ciel.orchestration — Orquestación

Primitivas de orquestación best-of-breed montadas sobre el Supervisor:

• Grafo de estado explícito (estilo LangGraph) con checkpoint: StateGraph, GraphNode, GraphEdge, GraphRunner, GraphState y stores.
• Flows event-driven (estilo CrewAI): Flow, FlowRunner, FlowStep.
• Group chat (estilo AutoGen): GroupChat, GroupChatManager, Agent.
• Root agent con sub-agents (estilo ADK): RootAgent, Specialist.
• Agencia autónoma en bucle (Fase 6): AutonomousAgent, EventLoop, Task.
• Sesiones: SessionStore.

ciel.orchestration

NodeFn = Callable[[Dict[str, Any]], Awaitable[Any]] module-attribute

__all__ = ['AgentStep', 'AgentSpec', 'GraphCheckpointStore', 'GraphEdge', 'GraphError', 'GraphNode', 'GraphPaused', 'GraphRunner', 'GraphState', 'GraphApprovalDenied', 'NodeFn', 'StateGraph', 'Flow', 'FlowCheckpointStore', 'FlowError', 'FlowRunner', 'FlowState', 'FlowStep', 'Agent', 'ChatMessage', 'GroupChat', 'GroupChatCheckpointStore', 'GroupChatError', 'GroupChatManager', 'GroupChatState', 'RootAgent', 'RootAgentError', 'RootCheckpointStore', 'RootRunner', 'RootState', 'Specialist', 'SessionError', 'SessionStore', 'AgentError', 'AutonomousAgent', 'EventLoop', 'EventLoopCheckpointStore', 'EventLoopStep', 'Task', 'TaskError'] module-attribute

Agent dataclass

Participante conversable del group chat.

responder recibe el GroupChatState y devuelve el texto de su réplica (o una corutina). Si terminate_if recibe el texto y devuelve True, ese mensaje finaliza la conversación.

Source code in src/ciel/orchestration/chat.py

@dataclass
class Agent:
    """Participante conversable del group chat.

    ``responder`` recibe el ``GroupChatState`` y devuelve el texto de su
    réplica (o una corutina). Si ``terminate_if`` recibe el texto y devuelve
    True, ese mensaje finaliza la conversación.
    """

    name: str
    responder: AgentFn
    system_message: str = ""
    terminate_if: Optional[Callable[[str], bool]] = None

    async def _reply(self, state: GroupChatState) -> str:
        res = self.responder(state)
        if hasattr(res, "__await__"):
            return await res
        return res

    def _is_terminator(self, text: str) -> bool:
        if self.terminate_if is None:
            return False
        return bool(self.terminate_if(text))

AgentError

Bases: Exception

Error base de la agencia autónoma (Fase 6).

Source code in src/ciel/orchestration/agent.py

class AgentError(Exception):
    """Error base de la agencia autónoma (Fase 6)."""

AgentSpec dataclass

Source code in src/ciel/orchestration/__init__.py

@dataclass
class AgentSpec:
    name: str
    steps: Sequence[AgentStep]
    topology: str = "pipeline"
    budget: Dict[str, Any] = field(default_factory=lambda: {"max_tools": 8})
    metadata: Dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> Dict[str, Any]:
        return {
            "name": self.name,
            "topology": self.topology,
            "budget": dict(self.budget),
            "steps": [
                {
                    "id": step.id,
                    "kind": step.kind,
                    "tool": step.tool,
                    "prompt": step.prompt,
                    "depends_on": list(step.depends_on),
                    "metadata": dict(step.metadata),
                }
                for step in self.steps
            ],
            "metadata": dict(self.metadata),
        }

    @classmethod
    def from_dict(cls, payload: Dict[str, Any]) -> AgentSpec:
        steps = [AgentStep(**item) for item in payload.get("steps", [])]
        return cls(
            name=payload.get("name", ""),
            steps=steps,
            topology=payload.get("topology", "pipeline"),
            budget=payload.get("budget", {}),
            metadata=payload.get("metadata", {}),
        )

    @classmethod
    def from_yaml(cls, payload: str) -> AgentSpec:
        import yaml

        data = yaml.safe_load(payload)
        if not isinstance(data, dict):
            raise ValueError("YAML root must be a mapping")
        return cls.from_dict(data)

AgentStep dataclass

Source code in src/ciel/orchestration/__init__.py

@dataclass
class AgentStep:
    id: str
    kind: str
    tool: Optional[str] = None
    prompt: Optional[str] = None
    depends_on: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)

AutonomousAgent

Agente autónomo que descompone un objetivo en tareas y las ejecuta.

OFFLINE-SAFE: por defecto el planner es un handler local que parte el objetivo en una sola tarea (o N tareas si se provee plan explícito); cada tarea se ejecuta con un handler local sobre task.payload (no red).

Source code in src/ciel/orchestration/agent.py

class AutonomousAgent:
    """Agente autónomo que descompone un objetivo en tareas y las ejecuta.

    OFFLINE-SAFE: por defecto el planner es un handler local que parte el
    objetivo en una sola tarea (o N tareas si se provee ``plan`` explícito);
    cada tarea se ejecuta con un handler local sobre ``task.payload`` (no red).
    """

    def __init__(
        self,
        name: str = "autonomous",
        *,
        supervisor: Optional[Supervisor] = None,
        checkpointer: Optional[EventLoopCheckpointStore] = None,
        session_store: Optional[SessionStore] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
        max_attempts: int = 5,
    ) -> None:
        self.name = name
        self.supervisor = supervisor or Supervisor()
        self.checkpointer = checkpointer
        self.session_store = session_store
        self.tenant_id = tenant_id
        self.session_id = session_id
        self.max_attempts = max_attempts

    def _loop(self, task: Task, handler: TaskHandler) -> EventLoop:
        return EventLoop(
            supervisor=self.supervisor,
            checkpointer=self.checkpointer,
            tenant_id=self.tenant_id,
            session_id=self.session_id,
            max_attempts=self.max_attempts,
        )

    async def run_task(self, task: Task, handler: TaskHandler) -> Task:
        loop = self._loop(task, handler)
        return await loop.run(task, handler)

    async def run_goal(
        self,
        goal: str,
        handler: TaskHandler,
        *,
        plan: Optional[Sequence[str]] = None,
    ) -> List[Task]:
        """Descompone ``goal`` en tareas (``plan`` o una sola) y las ejecuta.

        Cada tarea completada persiste un turno en ``SessionStore`` si está
        disponible (ADK session_state entre ejecuciones del agente).
        """
        steps = list(plan) if plan else [goal]
        tasks: List[Task] = [Task(goal=step) for step in steps]
        for task in tasks:
            await self.run_task(task, handler)
            self._persist_turn(task)
        return tasks

    def _persist_turn(self, task: Task) -> None:
        if self.session_store is None or self.session_id is None:
            return
        self.session_store.append_turn(
            tenant_id=self.tenant_id,
            session_id=self.session_id,
            turn={
                "goal": task.goal,
                "status": task.status,
                "result": task.result,
                "attempts": task.attempts,
            },
        )

run_goal(goal: str, handler: TaskHandler, *, plan: Optional[Sequence[str]] = None) -> List[Task] async

Descompone goal en tareas (plan o una sola) y las ejecuta.

Cada tarea completada persiste un turno en SessionStore si está disponible (ADK session_state entre ejecuciones del agente).

Source code in src/ciel/orchestration/agent.py

async def run_goal(
    self,
    goal: str,
    handler: TaskHandler,
    *,
    plan: Optional[Sequence[str]] = None,
) -> List[Task]:
    """Descompone ``goal`` en tareas (``plan`` o una sola) y las ejecuta.

    Cada tarea completada persiste un turno en ``SessionStore`` si está
    disponible (ADK session_state entre ejecuciones del agente).
    """
    steps = list(plan) if plan else [goal]
    tasks: List[Task] = [Task(goal=step) for step in steps]
    for task in tasks:
        await self.run_task(task, handler)
        self._persist_turn(task)
    return tasks

ChatMessage dataclass

Mensaje del transcripto (estilo AutoGen ChatMessage simplificado).

Source code in src/ciel/orchestration/chat.py

@dataclass
class ChatMessage:
    """Mensaje del transcripto (estilo AutoGen ChatMessage simplificado)."""

    role: str
    content: str
    round: int = 0

    def to_dict(self) -> Message:
        return {"role": self.role, "content": self.content, "round": self.round}

EventLoop

Bucle de eventos durable que ejecuta una Task con reintentos.

Montado sobre Supervisor (retry/timeout/budget por worker). Si el handler falla, el loop aplica backoff exponencial (capped) y reintenta hasta max_attempts. Tras cada intento persiste un checkpoint en MemoryStore; resume rehidrata y continúa.

OFFLINE-SAFE: el handler por defecto (y los demos) son funciones locales.

Source code in src/ciel/orchestration/agent.py

class EventLoop:
    """Bucle de eventos durable que ejecuta una ``Task`` con reintentos.

    Montado sobre ``Supervisor`` (retry/timeout/budget por worker). Si el
    handler falla, el loop aplica backoff exponencial (capped) y reintenta
    hasta ``max_attempts``. Tras cada intento persiste un checkpoint en
    ``MemoryStore``; ``resume`` rehidrata y continúa.

    OFFLINE-SAFE: el handler por defecto (y los demos) son funciones locales.
    """

    def __init__(
        self,
        *,
        supervisor: Optional[Supervisor] = None,
        checkpointer: Optional[EventLoopCheckpointStore] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
        max_attempts: int = 5,
        base_delay_s: float = 0.05,
        max_delay_s: float = 2.0,
        jitter: bool = False,
    ) -> None:
        self.supervisor = supervisor or Supervisor()
        self.checkpointer = checkpointer
        self.tenant_id = tenant_id
        self.session_id = session_id
        self.max_attempts = max_attempts
        self.base_delay_s = base_delay_s
        self.max_delay_s = max_delay_s
        self.jitter = jitter
        self.run_id: Optional[str] = None
        self.task: Optional[Task] = None
        self.steps: List[EventLoopStep] = []

    # -- helpers -------------------------------------------------------------
    @staticmethod
    async def _invoke(fn: TaskHandler, task: Task) -> Any:
        res = fn(task)
        if hasattr(res, "__await__"):
            return await res
        return res

    def _backoff(self, attempt: int) -> float:
        delay = min(self.base_delay_s * (2 ** (attempt - 1)), self.max_delay_s)
        if self.jitter:
            delay += (time.perf_counter() % 1) * 0.01
        return delay

    def _save(self, loop_state: Dict[str, Any]) -> None:
        if self.checkpointer is not None and self.run_id is not None and self.task is not None:
            self.checkpointer.save(
                run_id=self.run_id,
                loop_state=loop_state,
                task=self.task,
                tenant_id=self.tenant_id,
                session_id=self.session_id,
            )

    # -- run -----------------------------------------------------------------
    async def run(
        self,
        task: Task,
        handler: TaskHandler,
        *,
        run_id: Optional[str] = None,
    ) -> Task:
        """Ejecuta ``task`` con reintentos exponenciales hasta completar o agotar."""
        self.run_id = run_id or str(uuid.uuid4())
        self.task = task
        self.steps = []

        attempt = 0
        last_error: Optional[str] = None
        while attempt < self.max_attempts:
            attempt += 1
            task.attempts = attempt
            task.mark_running()
            start = time.perf_counter()
            try:
                result = await self.supervisor.run(
                    step_id=f"{self.run_id}:attempt:{attempt}",
                    worker=self._make_worker(handler, task),
                    payload={"attempt": attempt},
                    worker_id=f"loop-{attempt}",
                )
                latency = (time.perf_counter() - start) * 1000.0
                if result.failed:
                    # El Supervisor ya agotó sus reintentos internos.
                    last_error = result.error or "supervisor worker failed"
                    self.steps.append(
                        EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
                    )
                else:
                    task.mark_succeeded(result.output)
                    self.steps.append(
                        EventLoopStep(attempt=attempt, succeeded=True, output=result.output, latency_ms=latency)
                    )
                    self._save({"attempt": attempt, "status": task.status})
                    return task
            except Exception as exc:  # handler levantó fuera del supervisor
                last_error = str(exc)
                latency = (time.perf_counter() - start) * 1000.0
                self.steps.append(
                    EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
                )

            self._save({"attempt": attempt, "status": task.status})
            if attempt < self.max_attempts:
                await asyncio.sleep(self._backoff(attempt))

        task.mark_failed(last_error or "unknown failure")
        self._save({"attempt": attempt, "status": task.status})
        raise TaskError(
            f"task '{task.goal}' failed after {attempt} attempts: {task.error}"
        )

    def _make_worker(self, handler: TaskHandler, task: Task):
        async def _worker(ctx: WorkerContext) -> Any:
            return await self._invoke(handler, task)

        return _worker

    # -- resume --------------------------------------------------------------
    async def resume(self, *, run_id: str, handler: TaskHandler) -> Task:
        """Reanuda un loop interrumpido desde su último checkpoint.

        Rehidrata la ``Task`` persistida y continúa los reintentos. Si la tarea
        ya estaba en ``succeeded``/``failed`` al interrumpirse, la devuelve tal
        cual (idempotente). Cumple el criterio Fase 6: tras reinicio continúa y
        completa.
        """
        if self.checkpointer is None:
            raise EventLoopError("resume requires a checkpointer")
        self.run_id = run_id
        payload = self.checkpointer.load(
            run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
        )
        if payload is None:
            raise EventLoopError(f"no checkpoint found for run_id '{run_id}'")
        task = Task.from_snapshot(payload["task"])
        self.task = task
        loop_state = payload.get("loop_state", {})
        self.steps = []

        # Si ya terminó en el checkpoint previo, no re-ejecutar.
        if task.status in ("succeeded", "failed"):
            return task

        attempt = int(loop_state.get("attempt", 0))
        last_error: Optional[str] = None
        while attempt < self.max_attempts:
            attempt += 1
            task.attempts = attempt
            task.mark_running()
            start = time.perf_counter()
            try:
                result = await self.supervisor.run(
                    step_id=f"{self.run_id}:attempt:{attempt}",
                    worker=self._make_worker(handler, task),
                    payload={"attempt": attempt},
                    worker_id=f"loop-{attempt}",
                )
                latency = (time.perf_counter() - start) * 1000.0
                if result.failed:
                    last_error = result.error or "supervisor worker failed"
                    self.steps.append(
                        EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
                    )
                else:
                    task.mark_succeeded(result.output)
                    self.steps.append(
                        EventLoopStep(attempt=attempt, succeeded=True, output=result.output, latency_ms=latency)
                    )
                    self._save({"attempt": attempt, "status": task.status})
                    return task
            except Exception as exc:
                last_error = str(exc)
                latency = (time.perf_counter() - start) * 1000.0
                self.steps.append(
                    EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
                )

            self._save({"attempt": attempt, "status": task.status})
            if attempt < self.max_attempts:
                await asyncio.sleep(self._backoff(attempt))

        task.mark_failed(last_error or "unknown failure")
        self._save({"attempt": attempt, "status": task.status})
        raise TaskError(
            f"task '{task.goal}' failed after resume ({attempt} attempts): {task.error}"
        )

resume(*, run_id: str, handler: TaskHandler) -> Task async

Reanuda un loop interrumpido desde su último checkpoint.

Rehidrata la Task persistida y continúa los reintentos. Si la tarea ya estaba en succeeded/failed al interrumpirse, la devuelve tal cual (idempotente). Cumple el criterio Fase 6: tras reinicio continúa y completa.

Source code in src/ciel/orchestration/agent.py

async def resume(self, *, run_id: str, handler: TaskHandler) -> Task:
    """Reanuda un loop interrumpido desde su último checkpoint.

    Rehidrata la ``Task`` persistida y continúa los reintentos. Si la tarea
    ya estaba en ``succeeded``/``failed`` al interrumpirse, la devuelve tal
    cual (idempotente). Cumple el criterio Fase 6: tras reinicio continúa y
    completa.
    """
    if self.checkpointer is None:
        raise EventLoopError("resume requires a checkpointer")
    self.run_id = run_id
    payload = self.checkpointer.load(
        run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
    )
    if payload is None:
        raise EventLoopError(f"no checkpoint found for run_id '{run_id}'")
    task = Task.from_snapshot(payload["task"])
    self.task = task
    loop_state = payload.get("loop_state", {})
    self.steps = []

    # Si ya terminó en el checkpoint previo, no re-ejecutar.
    if task.status in ("succeeded", "failed"):
        return task

    attempt = int(loop_state.get("attempt", 0))
    last_error: Optional[str] = None
    while attempt < self.max_attempts:
        attempt += 1
        task.attempts = attempt
        task.mark_running()
        start = time.perf_counter()
        try:
            result = await self.supervisor.run(
                step_id=f"{self.run_id}:attempt:{attempt}",
                worker=self._make_worker(handler, task),
                payload={"attempt": attempt},
                worker_id=f"loop-{attempt}",
            )
            latency = (time.perf_counter() - start) * 1000.0
            if result.failed:
                last_error = result.error or "supervisor worker failed"
                self.steps.append(
                    EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
                )
            else:
                task.mark_succeeded(result.output)
                self.steps.append(
                    EventLoopStep(attempt=attempt, succeeded=True, output=result.output, latency_ms=latency)
                )
                self._save({"attempt": attempt, "status": task.status})
                return task
        except Exception as exc:
            last_error = str(exc)
            latency = (time.perf_counter() - start) * 1000.0
            self.steps.append(
                EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
            )

        self._save({"attempt": attempt, "status": task.status})
        if attempt < self.max_attempts:
            await asyncio.sleep(self._backoff(attempt))

    task.mark_failed(last_error or "unknown failure")
    self._save({"attempt": attempt, "status": task.status})
    raise TaskError(
        f"task '{task.goal}' failed after resume ({attempt} attempts): {task.error}"
    )

run(task: Task, handler: TaskHandler, *, run_id: Optional[str] = None) -> Task async

Ejecuta task con reintentos exponenciales hasta completar o agotar.

Source code in src/ciel/orchestration/agent.py

async def run(
    self,
    task: Task,
    handler: TaskHandler,
    *,
    run_id: Optional[str] = None,
) -> Task:
    """Ejecuta ``task`` con reintentos exponenciales hasta completar o agotar."""
    self.run_id = run_id or str(uuid.uuid4())
    self.task = task
    self.steps = []

    attempt = 0
    last_error: Optional[str] = None
    while attempt < self.max_attempts:
        attempt += 1
        task.attempts = attempt
        task.mark_running()
        start = time.perf_counter()
        try:
            result = await self.supervisor.run(
                step_id=f"{self.run_id}:attempt:{attempt}",
                worker=self._make_worker(handler, task),
                payload={"attempt": attempt},
                worker_id=f"loop-{attempt}",
            )
            latency = (time.perf_counter() - start) * 1000.0
            if result.failed:
                # El Supervisor ya agotó sus reintentos internos.
                last_error = result.error or "supervisor worker failed"
                self.steps.append(
                    EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
                )
            else:
                task.mark_succeeded(result.output)
                self.steps.append(
                    EventLoopStep(attempt=attempt, succeeded=True, output=result.output, latency_ms=latency)
                )
                self._save({"attempt": attempt, "status": task.status})
                return task
        except Exception as exc:  # handler levantó fuera del supervisor
            last_error = str(exc)
            latency = (time.perf_counter() - start) * 1000.0
            self.steps.append(
                EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
            )

        self._save({"attempt": attempt, "status": task.status})
        if attempt < self.max_attempts:
            await asyncio.sleep(self._backoff(attempt))

    task.mark_failed(last_error or "unknown failure")
    self._save({"attempt": attempt, "status": task.status})
    raise TaskError(
        f"task '{task.goal}' failed after {attempt} attempts: {task.error}"
    )

EventLoopCheckpointStore

Persiste el estado del EventLoop sobre MemoryStore.

Clave: loop:<run_id> (namespaced para no colisionar con otros módulos). Guarda el Task en curso + el estado del loop (intent, estado global).

Source code in src/ciel/orchestration/agent.py

class EventLoopCheckpointStore:
    """Persiste el estado del EventLoop sobre ``MemoryStore``.

    Clave: ``loop:<run_id>`` (namespaced para no colisionar con otros módulos).
    Guarda el ``Task`` en curso + el estado del loop (intent, estado global).
    """

    def __init__(self, memory_store: MemoryStore) -> None:
        self.memory = memory_store

    def _key(self, run_id: str) -> str:
        return f"loop:{run_id}"

    def save(
        self,
        *,
        run_id: str,
        loop_state: Dict[str, Any],
        task: Task,
        tenant_id: Optional[str],
        session_id: Optional[str],
    ) -> str:
        checkpoint_id = str(uuid.uuid4())
        payload = {
            "checkpoint_id": checkpoint_id,
            "run_id": run_id,
            "loop_state": dict(loop_state),
            "task": task.snapshot(),
        }
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id or run_id,
            key=self._key(run_id),
            value=payload,
        )
        return checkpoint_id

    def load(
        self, *, run_id: str, tenant_id: Optional[str], session_id: Optional[str]
    ) -> Optional[Dict[str, Any]]:
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=session_id or run_id, key=self._key(run_id)
        )
        return payload if isinstance(payload, dict) else None

EventLoopStep dataclass

Resultado de un intento de ejecución de una tarea en el loop.

Source code in src/ciel/orchestration/agent.py

@dataclass
class EventLoopStep:
    """Resultado de un intento de ejecución de una tarea en el loop."""

    attempt: int
    succeeded: bool
    output: Any = None
    error: Optional[str] = None
    latency_ms: float = 0.0

Flow

Constructor declarativo de flows event-driven.

API: add_start, add_listen, add_router; luego compile. Los IDs de paso se derivan del __name__ de la función salvo que se pasen explícitamente.

Source code in src/ciel/orchestration/flows.py

class Flow:
    """Constructor declarativo de flows event-driven.

    API: ``add_start``, ``add_listen``, ``add_router``; luego ``compile``.
    Los IDs de paso se derivan del ``__name__`` de la función salvo que se
    pasen explícitamente.
    """

    def __init__(self, name: str = "flow") -> None:
        self.name = name
        self._steps: Dict[str, FlowStep] = {}
        self._order: List[str] = []
        self._sources: Dict[str, set] = {}  # paso -> conjunto de pasos fuente
        self._router_branches: Dict[str, Dict[Any, str]] = {}  # router -> {valor: destino}

    # -- registro ------------------------------------------------------------
    def add_start(self, fn: StepFn, step_id: Optional[str] = None) -> "Flow":
        sid = step_id or getattr(fn, "__name__", "start")
        self._register(FlowStep(id=sid, kind="start", fn=fn))
        self._sources[sid] = set()
        return self

    def add_listen(self, source_id: str, fn: StepFn, step_id: Optional[str] = None) -> "Flow":
        sid = step_id or getattr(fn, "__name__", "listen")
        self._register(FlowStep(id=sid, kind="listen", fn=fn, source=source_id))
        self._sources[sid] = {source_id}
        return self

    def add_router(
        self,
        source_id: str,
        fn: StepFn,
        branches: Dict[Any, str],
        step_id: Optional[str] = None,
    ) -> "Flow":
        sid = step_id or getattr(fn, "__name__", "router")
        self._register(FlowStep(id=sid, kind="router", fn=fn, source=source_id, branches=dict(branches)))
        self._sources[sid] = {source_id}
        self._router_branches[sid] = dict(branches)
        return self

    def add_branch(self, fn: StepFn, step_id: Optional[str] = None) -> "Flow":
        """Paso rama activado EXCLUSIVAMENTE por un router (no tiene fuente de datos)."""
        sid = step_id or getattr(fn, "__name__", "branch")
        self._register(FlowStep(id=sid, kind="branch", fn=fn))
        self._sources[sid] = set()
        return self

    def _register(self, step: FlowStep) -> None:
        if step.id in self._steps:
            raise FlowError(f"step '{step.id}' already registered")
        self._steps[step.id] = step
        self._order.append(step.id)

    def steps(self) -> Dict[str, FlowStep]:
        return dict(self._steps)

    def compile(
        self,
        *,
        supervisor: Optional[Supervisor] = None,
        max_steps: int = 256,
        checkpointer: Optional["FlowCheckpointStore"] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> "FlowRunner":
        # Validación de que los destinos de router existen (evita carreras en runtime).
        for rid, branches in self._router_branches.items():
            for tid in branches.values():
                if tid not in self._steps:
                    raise FlowError(f"router '{rid}' branch target '{tid}' is not registered")
        return FlowRunner(
            flow=self,
            supervisor=supervisor or Supervisor(),
            max_steps=max_steps,
            checkpointer=checkpointer,
            tenant_id=tenant_id,
            session_id=session_id,
        )

add_branch(fn: StepFn, step_id: Optional[str] = None) -> 'Flow'

Paso rama activado EXCLUSIVAMENTE por un router (no tiene fuente de datos).

Source code in src/ciel/orchestration/flows.py

def add_branch(self, fn: StepFn, step_id: Optional[str] = None) -> "Flow":
    """Paso rama activado EXCLUSIVAMENTE por un router (no tiene fuente de datos)."""
    sid = step_id or getattr(fn, "__name__", "branch")
    self._register(FlowStep(id=sid, kind="branch", fn=fn))
    self._sources[sid] = set()
    return self

FlowCheckpointStore

Persistencia de checkpoints de flow sobre MemoryStore (multitenancy nativo).

Source code in src/ciel/orchestration/flows.py

class FlowCheckpointStore:
    """Persistencia de checkpoints de flow sobre ``MemoryStore`` (multitenancy nativo)."""

    def __init__(self, memory_store: MemoryStore) -> None:
        self.memory = memory_store

    def _key(self, run_id: str) -> str:
        return f"flow:{run_id}"

    def save(
        self,
        *,
        run_id: str,
        completed: Sequence[str],
        state: FlowState,
        finished: bool,
        tenant_id: Optional[str],
        session_id: Optional[str],
    ) -> str:
        checkpoint_id = str(uuid.uuid4())
        payload = {
            "checkpoint_id": checkpoint_id,
            "run_id": run_id,
            "completed": list(completed),
            "finished": finished,
            "state": state.snapshot(),
        }
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id or run_id,
            key=self._key(run_id),
            value=payload,
        )
        return checkpoint_id

    def load(self, *, run_id: str, tenant_id: Optional[str], session_id: Optional[str]) -> Optional[Dict[str, Any]]:
        payload = self.memory.get(tenant_id=tenant_id, session_id=session_id or run_id, key=self._key(run_id))
        return payload if isinstance(payload, dict) else None

FlowError

Bases: Exception

Source code in src/ciel/orchestration/flows.py

class FlowError(Exception):
    pass

FlowRunner

Ejecuta el flow paso a paso (event-driven) con checkpoint + resume.

Source code in src/ciel/orchestration/flows.py

class FlowRunner:
    """Ejecuta el flow paso a paso (event-driven) con checkpoint + resume."""

    def __init__(
        self,
        *,
        flow: Flow,
        supervisor: Supervisor,
        max_steps: int = 256,
        checkpointer: Optional[FlowCheckpointStore] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> None:
        self.flow = flow
        self.supervisor = supervisor
        self.max_steps = max_steps
        self.checkpointer = checkpointer
        self.tenant_id = tenant_id
        self.session_id = session_id
        self.run_id: Optional[str] = None
        self._activated: set = set()

    async def _run_step(self, step: FlowStep, state: FlowState, idx: int) -> Any:
        async def _worker(ctx: WorkerContext) -> Any:
            res = step.fn(state)
            if hasattr(res, "__await__"):
                return await res
            return res

        result = await self.supervisor.run(
            step_id=f"{self.run_id}:{step.id}:{idx}",
            worker=_worker,
            payload={"step": step.id, "kind": step.kind},
            worker_id=step.id,
        )
        if result.failed:
            raise FlowError(f"step '{step.id}' failed after {result.attempts} attempts: {result.error}")
        return result.output

    def _ready(self, state: FlowState) -> List[str]:
        out: List[str] = []
        for sid in self.flow._order:
            if sid in state.completed:
                continue
            step = self.flow._steps[sid]
            # Una rama SOLO está lista si un router la activó explícitamente.
            if step.kind == "branch":
                if sid in self._activated:
                    out.append(sid)
                continue
            if sid in self._activated:
                out.append(sid)
                continue
            srcs = self.flow._sources.get(sid, set())
            if all(s in state.completed for s in srcs):
                out.append(sid)
        return out

    async def _drive(self, state: FlowState) -> FlowState:
        idx = 0
        while True:
            ready = self._ready(state)
            if not ready:
                break
            sid = ready[0]  # determinista: orden de registro
            step = self.flow._steps[sid]
            out = await self._run_step(step, state, idx)
            state.results[sid] = out
            state.data.setdefault("__out__", {})[sid] = out
            state.completed.append(sid)
            state.last_event = sid
            idx += 1

            # Un router activa exactamente una rama tras completarse.
            if step.kind == "router":
                target = self.flow._router_branches[sid].get(out)
                if target is None:
                    raise FlowError(f"router '{sid}' returned {out!r} with no matching branch")
                self._activated.add(target)

            if self.checkpointer is not None:
                self.checkpointer.save(
                    run_id=self.run_id,
                    completed=state.completed,
                    state=state,
                    finished=False,
                    tenant_id=self.tenant_id,
                    session_id=self.session_id,
                )
            if idx >= self.max_steps:
                raise FlowError(f"exceeded max_steps={self.max_steps} (possible cycle)")
        state.last_event = state.completed[-1] if state.completed else None
        return state

    async def run(self, *, initial_data: Optional[Dict[str, Any]] = None, run_id: Optional[str] = None) -> FlowState:
        self.run_id = run_id or str(uuid.uuid4())
        self._activated = set()
        state = FlowState(data=dict(initial_data or {}))
        return await self._drive(state)

    async def resume(self, *, run_id: str) -> FlowState:
        """Reanuda un flow interrumpido desde su último checkpoint.

        Reconstruye el estado persistido y recomputa las ramas de router ya
        activadas (derivables de los routers completados + sus resultados).
        """
        if self.checkpointer is None:
            raise FlowError("resume requires a checkpointer")
        self.run_id = run_id
        payload = self.checkpointer.load(run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id)
        if payload is None:
            raise FlowError(f"no checkpoint found for run_id '{run_id}'")
        state = FlowState.from_snapshot(payload["state"])
        self._activated = set()
        for rid, branches in self.flow._router_branches.items():
            if rid in state.completed:
                target = branches.get(state.results.get(rid))
                if target:
                    self._activated.add(target)
        return await self._drive(state)

resume(*, run_id: str) -> FlowState async

Reanuda un flow interrumpido desde su último checkpoint.

Reconstruye el estado persistido y recomputa las ramas de router ya activadas (derivables de los routers completados + sus resultados).

Source code in src/ciel/orchestration/flows.py

async def resume(self, *, run_id: str) -> FlowState:
    """Reanuda un flow interrumpido desde su último checkpoint.

    Reconstruye el estado persistido y recomputa las ramas de router ya
    activadas (derivables de los routers completados + sus resultados).
    """
    if self.checkpointer is None:
        raise FlowError("resume requires a checkpointer")
    self.run_id = run_id
    payload = self.checkpointer.load(run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id)
    if payload is None:
        raise FlowError(f"no checkpoint found for run_id '{run_id}'")
    state = FlowState.from_snapshot(payload["state"])
    self._activated = set()
    for rid, branches in self.flow._router_branches.items():
        if rid in state.completed:
            target = branches.get(state.results.get(rid))
            if target:
                self._activated.add(target)
    return await self._drive(state)

FlowState dataclass

Estado mutable compartido entre pasos de un flow (estilo CrewAI state).

Source code in src/ciel/orchestration/flows.py

@dataclass
class FlowState:
    """Estado mutable compartido entre pasos de un flow (estilo CrewAI state)."""

    data: Dict[str, Any] = field(default_factory=dict)
    results: Dict[str, Any] = field(default_factory=dict)
    completed: List[str] = field(default_factory=list)
    last_event: Optional[str] = None

    def snapshot(self) -> Dict[str, Any]:
        return {
            "data": dict(self.data),
            "results": dict(self.results),
            "completed": list(self.completed),
            "last_event": self.last_event,
        }

    @classmethod
    def from_snapshot(cls, snap: Dict[str, Any]) -> "FlowState":
        return cls(
            data=dict(snap.get("data", {})),
            results=dict(snap.get("results", {})),
            completed=list(snap.get("completed", [])),
            last_event=snap.get("last_event"),
        )

FlowStep dataclass

Source code in src/ciel/orchestration/flows.py

@dataclass
class FlowStep:
    id: str
    kind: str  # "start" | "listen" | "router"
    fn: StepFn
    source: Optional[str] = None
    branches: Dict[Any, str] = field(default_factory=dict)  # router: valor -> paso destino

GraphApprovalDenied

Bases: Exception

Lanzada cuando un nodo pausado por HIL es denegado por el aprobadador.

Source code in src/ciel/orchestration/graph.py

class GraphApprovalDenied(Exception):
    """Lanzada cuando un nodo pausado por HIL es denegado por el aprobadador."""

GraphCheckpointStore

Persistencia de checkpoints del grafo sobre MemoryStore.

Mismo patrón que ciel.runtime.checkpoints.CheckpointStore: usa la clave (tenant_id, session_id, key) ya con multitenancy nativo.

Source code in src/ciel/orchestration/graph.py

class GraphCheckpointStore:
    """Persistencia de checkpoints del grafo sobre ``MemoryStore``.

    Mismo patrón que ``ciel.runtime.checkpoints.CheckpointStore``: usa la
    clave ``(tenant_id, session_id, key)`` ya con multitenancy nativo.
    """

    def __init__(self, memory_store: MemoryStore) -> None:
        self.memory = memory_store

    def _key(self, run_id: str) -> str:
        return f"graph:{run_id}"

    def save(self, *, run_id: str, step_index: int, state: GraphState, finished: bool, tenant_id: Optional[str], session_id: Optional[str], paused: bool = False, paused_node: Optional[str] = None) -> str:
        checkpoint_id = str(uuid.uuid4())
        payload = {
            "checkpoint_id": checkpoint_id,
            "run_id": run_id,
            "step_index": step_index,
            "finished": finished,
            "paused": paused,
            "paused_node": paused_node,
            "state": state.snapshot(),
        }
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id or run_id,
            key=self._key(run_id),
            value=payload,
        )
        return checkpoint_id

    def load(self, *, run_id: str, tenant_id: Optional[str], session_id: Optional[str]) -> Optional[Dict[str, Any]]:
        payload = self.memory.get(
            tenant_id=tenant_id,
            session_id=session_id or run_id,
            key=self._key(run_id),
        )
        return payload if isinstance(payload, dict) else None

GraphEdge dataclass

Source code in src/ciel/orchestration/graph.py

@dataclass
class GraphEdge:
    source: str
    target: str
    condition: Optional[EdgeCondition] = None

GraphError

Bases: Exception

Source code in src/ciel/orchestration/graph.py

class GraphError(Exception):
    pass

GraphNode dataclass

Source code in src/ciel/orchestration/graph.py

@dataclass
class GraphNode:
    id: str
    fn: NodeFn
    worker_id: str = "node"
    supervisor_kwargs: Dict[str, Any] = field(default_factory=dict)
    # Human-in-the-loop: si se define, el runner pausa ANTES de ejecutar el
    # nodo y exige aprobación RBAC con esta acción (p. ej. "approve:deploy").
    require_approval: Optional[str] = None

GraphPaused

Bases: Exception

Lanzada por run cuando un nodo exige aprobación (HIL).

Atributos

node_id: id del nodo pausado. action: acción RBAC requerida para aprobar (p. ej. "approve:deploy"). run_id: id del run pausado (para reanudar con approve).

Source code in src/ciel/orchestration/graph.py

class GraphPaused(Exception):
    """Lanzada por ``run`` cuando un nodo exige aprobación (HIL).

    Atributos:
        node_id: id del nodo pausado.
        action: acción RBAC requerida para aprobar (p. ej. ``"approve:deploy"``).
        run_id: id del run pausado (para reanudar con ``approve``).
    """

    def __init__(self, message: str, *, node_id: str, action: str, run_id: str) -> None:
        super().__init__(message)
        self.node_id = node_id
        self.action = action
        self.run_id = run_id

GraphRunner

Ejecuta el grafo nodo a nodo, con checkpoint + reanudación + time-travel.

Source code in src/ciel/orchestration/graph.py

class GraphRunner:
    """Ejecuta el grafo nodo a nodo, con checkpoint + reanudación + time-travel."""

    def __init__(
        self,
        *,
        graph: StateGraph,
        supervisor: Supervisor,
        max_steps: int = 64,
        checkpointer: Optional[GraphCheckpointStore] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> None:
        self.graph = graph
        self.supervisor = supervisor
        self.max_steps = max_steps
        self.checkpointer = checkpointer
        self.tenant_id = tenant_id
        self.session_id = session_id
        self.run_id: Optional[str] = None

    async def _run_node(self, node: GraphNode, state: GraphState, step_index: int) -> WorkerResult:
        async def _worker(ctx: WorkerContext) -> Any:
            result = node.fn(state.data)
            if hasattr(result, "__await__"):
                return await result
            return result

        return await self.supervisor.run(
            step_id=f"{self.run_id}:{node.id}:{step_index}",
            worker=_worker,
            payload={"node": node.id, "step_index": step_index},
            worker_id=node.worker_id,
            **node.supervisor_kwargs,
        )

    def _next(self, node_id: str, state: GraphState) -> Optional[str]:
        edges = self.graph.outgoing(node_id)
        if not edges:
            return None
        for edge in edges:
            if edge.condition is None or edge.condition(state.data):
                return edge.target
        # No hubo arista tomada: si hay un finish point, detener; si no, error.
        if self.graph._finish == node_id:
            return None
        return None

    async def run(self, *, initial_data: Optional[Dict[str, Any]] = None, run_id: Optional[str] = None) -> GraphState:
        self.run_id = run_id or str(uuid.uuid4())
        state = GraphState(data=dict(initial_data or {}))
        current = self.graph._entry
        step_index = 0
        start = time.perf_counter()

        while current is not None and step_index < self.max_steps:
            node = self.graph._nodes[current]
            state.current_node = current

            # Human-in-the-loop: si el nodo exige aprobación, pausamos ANTES de
            # ejecutarlo y persistimos un checkpoint marcado como pausado.
            if node.require_approval is not None:
                if self.checkpointer is not None:
                    self.checkpointer.save(
                        run_id=self.run_id,
                        step_index=step_index,
                        state=state,
                        finished=False,
                        tenant_id=self.tenant_id,
                        session_id=self.session_id,
                        paused=True,
                        paused_node=current,
                    )
                raise GraphPaused(
                    f"node '{current}' requires approval ('{node.require_approval}')",
                    node_id=current,
                    action=node.require_approval,
                    run_id=self.run_id,
                )

            result = await self._run_node(node, state, step_index)
            if result.failed:
                raise GraphError(
                    f"node '{current}' failed after {result.attempts} attempts: {result.error}"
                )
            state.data[f"__out__{current}"] = result.output
            state.last_output = result.output
            state.visited.append(current)
            step_index += 1

            finished = current == self.graph._finish
            if self.checkpointer is not None:
                self.checkpointer.save(
                    run_id=self.run_id,
                    step_index=step_index,
                    state=state,
                    finished=finished,
                    tenant_id=self.tenant_id,
                    session_id=self.session_id,
                )

            if finished:
                break
            current = self._next(current, state)

        if step_index >= self.max_steps:
            raise GraphError(f"exceeded max_steps={self.max_steps} (possible cycle without finish)")

        state.current_node = current
        return state

    async def approve(
        self,
        run_id: str,
        *,
        approver: Optional[str] = None,
        rbac: "object | None" = None,
        action: Optional[str] = None,
    ) -> GraphState:
        """Reanuda un grafo pausado por HIL tras aprobar el nodo pendiente.

        Si se pasa ``rbac`` (un ``RBACEngine``), valida que ``approver`` tenga
        el permiso ``action`` (por defecto el ``require_approval`` del nodo
        pausado). Si no tiene permiso, lanza ``RBACError`` (reutiliza
        ``ciel.enterprise.rbac``). Tras aprobar, ejecuta el nodo y continúa el
        grafo desde allí.

        Requiere un checkpointer (el grafo debe haber sido pausado con uno).
        """
        if self.checkpointer is None:
            raise GraphError("approve requires a checkpointer (pause happened with one)")
        self.run_id = run_id
        payload = self.checkpointer.load(
            run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
        )
        if payload is None:
            raise GraphError(f"no checkpoint found for run_id '{run_id}'")
        if not payload.get("paused"):
            raise GraphError(f"run_id '{run_id}' is not paused (no HIL approval pending)")
        if payload.get("finished"):
            state = GraphState.from_snapshot(payload["state"])
            state.current_node = None
            return state

        paused_node = payload["paused_node"]
        node = self.graph._nodes.get(paused_node)
        if node is None:
            raise GraphError(f"paused node '{paused_node}' no longer exists in graph")
        action = action or node.require_approval
        if rbac is not None and action is not None:
            rbac.check(approver, action, tenant_id=self.tenant_id)

        state = GraphState.from_snapshot(payload["state"])
        step_index = int(payload.get("step_index", 0))
        # Ejecuta el nodo pausado y continúa desde allí (no re-ejecuta previos).
        state.current_node = paused_node
        result = await self._run_node(node, state, step_index)
        if result.failed:
            raise GraphError(
                f"node '{paused_node}' failed after approval: {result.error}"
            )
        state.data[f"__out__{paused_node}"] = result.output
        state.last_output = result.output
        state.visited.append(paused_node)
        step_index += 1
        current = self._next(paused_node, state)

        while current is not None and step_index < self.max_steps:
            node = self.graph._nodes[current]
            state.current_node = current
            result = await self._run_node(node, state, step_index)
            if result.failed:
                raise GraphError(
                    f"node '{current}' failed after {result.attempts} attempts: {result.error}"
                )
            state.data[f"__out__{current}"] = result.output
            state.last_output = result.output
            state.visited.append(current)
            step_index += 1
            finished = current == self.graph._finish
            self.checkpointer.save(
                run_id=self.run_id,
                step_index=step_index,
                state=state,
                finished=finished,
                tenant_id=self.tenant_id,
                session_id=self.session_id,
            )
            if finished:
                break
            current = self._next(current, state)

        if step_index >= self.max_steps:
            raise GraphError(f"exceeded max_steps={self.max_steps} during approve")
        state.current_node = current
        return state

    async def deny(self, run_id: str, *, reason: Optional[str] = None) -> None:
        """Marca un grafo pausado por HIL como denegado y detiene el flujo.

        Persiste el checkpoint como no-pausado y no-finalizado; el estado
        queda libre para inspección pero el grafo no continúa. Lanza
        ``GraphApprovalDenied`` para que el llamador sepa que fue rechazado.
        """
        if self.checkpointer is None:
            raise GraphError("deny requires a checkpointer")
        self.run_id = run_id
        payload = self.checkpointer.load(
            run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
        )
        if payload is None:
            raise GraphError(f"no checkpoint found for run_id '{run_id}'")
        if not payload.get("paused"):
            raise GraphError(f"run_id '{run_id}' is not paused")
        # Persiste como no-pausado para que resume/approve no lo retomen.
        state = GraphState.from_snapshot(payload["state"])
        self.checkpointer.save(
            run_id=run_id,
            step_index=int(payload.get("step_index", 0)),
            state=state,
            finished=False,
            tenant_id=self.tenant_id,
            session_id=self.session_id,
            paused=False,
            paused_node=None,
        )
        raise GraphApprovalDenied(
            f"approval denied for run_id '{run_id}'"
 + (f": {reason}" if reason else "")
        )

    async def resume(self, *, run_id: str) -> GraphState:
        """Reanuda un grafo interrumpido desde su último checkpoint.

        Reconstruye el estado persistido y continúa desde el último nodo
        visitado. Si el checkpoint marca ``finished=True``, devuelve el estado
        tal cual (idempotente).
        """
        if self.checkpointer is None:
            raise GraphError("resume requires a checkpointer")
        self.run_id = run_id
        payload = self.checkpointer.load(
            run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
        )
        if payload is None:
            raise GraphError(f"no checkpoint found for run_id '{run_id}'")
        if payload.get("finished"):
            state = GraphState.from_snapshot(payload["state"])
            state.current_node = None
            return state
        if payload.get("paused"):
            raise GraphPaused(
                f"run_id '{run_id}' is paused waiting for HIL approval",
                node_id=payload.get("paused_node") or "",
                action=(self.graph._nodes.get(payload.get("paused_node"))
                        and self.graph._nodes[payload["paused_node"]].require_approval)
                or "approve:*",
                run_id=run_id,
            )

        state = GraphState.from_snapshot(payload["state"])
        # Retomar desde el nodo SIGUIENTE al último completado (no re-ejecutar
        # el último visitado). Si no hay aristas salientes, el grafo terminó.
        if state.visited:
            current = self._next(state.visited[-1], state)
        else:
            current = self.graph._entry
        step_index = int(payload.get("step_index", 0))

        while current is not None and step_index < self.max_steps:
            node = self.graph._nodes[current]
            state.current_node = current
            result = await self._run_node(node, state, step_index)
            if result.failed:
                raise GraphError(
                    f"node '{current}' failed after {result.attempts} attempts: {result.error}"
                )
            state.data[f"__out__{current}"] = result.output
            state.last_output = result.output
            state.visited.append(current)
            step_index += 1

            finished = current == self.graph._finish
            self.checkpointer.save(
                run_id=self.run_id,
                step_index=step_index,
                state=state,
                finished=finished,
                tenant_id=self.tenant_id,
                session_id=self.session_id,
            )
            if finished:
                break
            current = self._next(current, state)

        if step_index >= self.max_steps:
            raise GraphError(f"exceeded max_steps={self.max_steps} during resume")
        state.current_node = current
        return state

    async def run_from(self, *, run_id: str, up_to_node: str, initial_data: Optional[Dict[str, Any]] = None) -> GraphState:
        """Time-travel: re-ejecuta el grafo desde el inicio hasta ``up_to_node``.

        Útil para reproducir/depurar un sub-tramo del flujo de forma
        determinista, con el mismo ``run_id`` (sobrescribe el checkpoint).
        """
        # Ejecuta normalmente pero se detiene al alcanzar up_to_node (inclusive).
        self.run_id = run_id
        state = GraphState(data=dict(initial_data or {}))
        current = self.graph._entry
        step_index = 0
        while current is not None and step_index < self.max_steps:
            node = self.graph._nodes[current]
            state.current_node = current
            result = await self._run_node(node, state, step_index)
            if result.failed:
                raise GraphError(f"node '{current}' failed: {result.error}")
            state.data[f"__out__{current}"] = result.output
            state.last_output = result.output
            state.visited.append(current)
            step_index += 1
            if self.checkpointer is not None:
                self.checkpointer.save(
                    run_id=self.run_id,
                    step_index=step_index,
                    state=state,
                    finished=(current == up_to_node),
                    tenant_id=self.tenant_id,
                    session_id=self.session_id,
                )
            if current == up_to_node:
                break
            current = self._next(current, state)
        state.current_node = current
        return state

approve(run_id: str, *, approver: Optional[str] = None, rbac: 'object | None' = None, action: Optional[str] = None) -> GraphState async

Reanuda un grafo pausado por HIL tras aprobar el nodo pendiente.

Si se pasa rbac (un RBACEngine), valida que approver tenga el permiso action (por defecto el require_approval del nodo pausado). Si no tiene permiso, lanza RBACError (reutiliza ciel.enterprise.rbac). Tras aprobar, ejecuta el nodo y continúa el grafo desde allí.

Requiere un checkpointer (el grafo debe haber sido pausado con uno).

Source code in src/ciel/orchestration/graph.py

async def approve(
    self,
    run_id: str,
    *,
    approver: Optional[str] = None,
    rbac: "object | None" = None,
    action: Optional[str] = None,
) -> GraphState:
    """Reanuda un grafo pausado por HIL tras aprobar el nodo pendiente.

    Si se pasa ``rbac`` (un ``RBACEngine``), valida que ``approver`` tenga
    el permiso ``action`` (por defecto el ``require_approval`` del nodo
    pausado). Si no tiene permiso, lanza ``RBACError`` (reutiliza
    ``ciel.enterprise.rbac``). Tras aprobar, ejecuta el nodo y continúa el
    grafo desde allí.

    Requiere un checkpointer (el grafo debe haber sido pausado con uno).
    """
    if self.checkpointer is None:
        raise GraphError("approve requires a checkpointer (pause happened with one)")
    self.run_id = run_id
    payload = self.checkpointer.load(
        run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
    )
    if payload is None:
        raise GraphError(f"no checkpoint found for run_id '{run_id}'")
    if not payload.get("paused"):
        raise GraphError(f"run_id '{run_id}' is not paused (no HIL approval pending)")
    if payload.get("finished"):
        state = GraphState.from_snapshot(payload["state"])
        state.current_node = None
        return state

    paused_node = payload["paused_node"]
    node = self.graph._nodes.get(paused_node)
    if node is None:
        raise GraphError(f"paused node '{paused_node}' no longer exists in graph")
    action = action or node.require_approval
    if rbac is not None and action is not None:
        rbac.check(approver, action, tenant_id=self.tenant_id)

    state = GraphState.from_snapshot(payload["state"])
    step_index = int(payload.get("step_index", 0))
    # Ejecuta el nodo pausado y continúa desde allí (no re-ejecuta previos).
    state.current_node = paused_node
    result = await self._run_node(node, state, step_index)
    if result.failed:
        raise GraphError(
            f"node '{paused_node}' failed after approval: {result.error}"
        )
    state.data[f"__out__{paused_node}"] = result.output
    state.last_output = result.output
    state.visited.append(paused_node)
    step_index += 1
    current = self._next(paused_node, state)

    while current is not None and step_index < self.max_steps:
        node = self.graph._nodes[current]
        state.current_node = current
        result = await self._run_node(node, state, step_index)
        if result.failed:
            raise GraphError(
                f"node '{current}' failed after {result.attempts} attempts: {result.error}"
            )
        state.data[f"__out__{current}"] = result.output
        state.last_output = result.output
        state.visited.append(current)
        step_index += 1
        finished = current == self.graph._finish
        self.checkpointer.save(
            run_id=self.run_id,
            step_index=step_index,
            state=state,
            finished=finished,
            tenant_id=self.tenant_id,
            session_id=self.session_id,
        )
        if finished:
            break
        current = self._next(current, state)

    if step_index >= self.max_steps:
        raise GraphError(f"exceeded max_steps={self.max_steps} during approve")
    state.current_node = current
    return state

deny(run_id: str, *, reason: Optional[str] = None) -> None async

Marca un grafo pausado por HIL como denegado y detiene el flujo.

Persiste el checkpoint como no-pausado y no-finalizado; el estado queda libre para inspección pero el grafo no continúa. Lanza GraphApprovalDenied para que el llamador sepa que fue rechazado.

Source code in src/ciel/orchestration/graph.py

   async def deny(self, run_id: str, *, reason: Optional[str] = None) -> None:
       """Marca un grafo pausado por HIL como denegado y detiene el flujo.

       Persiste el checkpoint como no-pausado y no-finalizado; el estado
       queda libre para inspección pero el grafo no continúa. Lanza
       ``GraphApprovalDenied`` para que el llamador sepa que fue rechazado.
       """
       if self.checkpointer is None:
           raise GraphError("deny requires a checkpointer")
       self.run_id = run_id
       payload = self.checkpointer.load(
           run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
       )
       if payload is None:
           raise GraphError(f"no checkpoint found for run_id '{run_id}'")
       if not payload.get("paused"):
           raise GraphError(f"run_id '{run_id}' is not paused")
       # Persiste como no-pausado para que resume/approve no lo retomen.
       state = GraphState.from_snapshot(payload["state"])
       self.checkpointer.save(
           run_id=run_id,
           step_index=int(payload.get("step_index", 0)),
           state=state,
           finished=False,
           tenant_id=self.tenant_id,
           session_id=self.session_id,
           paused=False,
           paused_node=None,
       )
       raise GraphApprovalDenied(
           f"approval denied for run_id '{run_id}'"
+ (f": {reason}" if reason else "")
       )

resume(*, run_id: str) -> GraphState async

Reanuda un grafo interrumpido desde su último checkpoint.

Reconstruye el estado persistido y continúa desde el último nodo visitado. Si el checkpoint marca finished=True, devuelve el estado tal cual (idempotente).

Source code in src/ciel/orchestration/graph.py

async def resume(self, *, run_id: str) -> GraphState:
    """Reanuda un grafo interrumpido desde su último checkpoint.

    Reconstruye el estado persistido y continúa desde el último nodo
    visitado. Si el checkpoint marca ``finished=True``, devuelve el estado
    tal cual (idempotente).
    """
    if self.checkpointer is None:
        raise GraphError("resume requires a checkpointer")
    self.run_id = run_id
    payload = self.checkpointer.load(
        run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
    )
    if payload is None:
        raise GraphError(f"no checkpoint found for run_id '{run_id}'")
    if payload.get("finished"):
        state = GraphState.from_snapshot(payload["state"])
        state.current_node = None
        return state
    if payload.get("paused"):
        raise GraphPaused(
            f"run_id '{run_id}' is paused waiting for HIL approval",
            node_id=payload.get("paused_node") or "",
            action=(self.graph._nodes.get(payload.get("paused_node"))
                    and self.graph._nodes[payload["paused_node"]].require_approval)
            or "approve:*",
            run_id=run_id,
        )

    state = GraphState.from_snapshot(payload["state"])
    # Retomar desde el nodo SIGUIENTE al último completado (no re-ejecutar
    # el último visitado). Si no hay aristas salientes, el grafo terminó.
    if state.visited:
        current = self._next(state.visited[-1], state)
    else:
        current = self.graph._entry
    step_index = int(payload.get("step_index", 0))

    while current is not None and step_index < self.max_steps:
        node = self.graph._nodes[current]
        state.current_node = current
        result = await self._run_node(node, state, step_index)
        if result.failed:
            raise GraphError(
                f"node '{current}' failed after {result.attempts} attempts: {result.error}"
            )
        state.data[f"__out__{current}"] = result.output
        state.last_output = result.output
        state.visited.append(current)
        step_index += 1

        finished = current == self.graph._finish
        self.checkpointer.save(
            run_id=self.run_id,
            step_index=step_index,
            state=state,
            finished=finished,
            tenant_id=self.tenant_id,
            session_id=self.session_id,
        )
        if finished:
            break
        current = self._next(current, state)

    if step_index >= self.max_steps:
        raise GraphError(f"exceeded max_steps={self.max_steps} during resume")
    state.current_node = current
    return state

run_from(*, run_id: str, up_to_node: str, initial_data: Optional[Dict[str, Any]] = None) -> GraphState async

Time-travel: re-ejecuta el grafo desde el inicio hasta up_to_node.

Útil para reproducir/depurar un sub-tramo del flujo de forma determinista, con el mismo run_id (sobrescribe el checkpoint).

Source code in src/ciel/orchestration/graph.py

async def run_from(self, *, run_id: str, up_to_node: str, initial_data: Optional[Dict[str, Any]] = None) -> GraphState:
    """Time-travel: re-ejecuta el grafo desde el inicio hasta ``up_to_node``.

    Útil para reproducir/depurar un sub-tramo del flujo de forma
    determinista, con el mismo ``run_id`` (sobrescribe el checkpoint).
    """
    # Ejecuta normalmente pero se detiene al alcanzar up_to_node (inclusive).
    self.run_id = run_id
    state = GraphState(data=dict(initial_data or {}))
    current = self.graph._entry
    step_index = 0
    while current is not None and step_index < self.max_steps:
        node = self.graph._nodes[current]
        state.current_node = current
        result = await self._run_node(node, state, step_index)
        if result.failed:
            raise GraphError(f"node '{current}' failed: {result.error}")
        state.data[f"__out__{current}"] = result.output
        state.last_output = result.output
        state.visited.append(current)
        step_index += 1
        if self.checkpointer is not None:
            self.checkpointer.save(
                run_id=self.run_id,
                step_index=step_index,
                state=state,
                finished=(current == up_to_node),
                tenant_id=self.tenant_id,
                session_id=self.session_id,
            )
        if current == up_to_node:
            break
        current = self._next(current, state)
    state.current_node = current
    return state

GraphState dataclass

Estado mutable compartido entre nodos (estilo state machine LangGraph).

Source code in src/ciel/orchestration/graph.py

@dataclass
class GraphState:
    """Estado mutable compartido entre nodos (estilo state machine LangGraph)."""

    data: Dict[str, Any] = field(default_factory=dict)
    visited: List[str] = field(default_factory=list)
    current_node: Optional[str] = None
    last_output: Any = None

    def snapshot(self) -> Dict[str, Any]:
        return {
            "data": dict(self.data),
            "visited": list(self.visited),
            "current_node": self.current_node,
            "last_output": self.last_output,
        }

    @classmethod
    def from_snapshot(cls, snap: Dict[str, Any]) -> GraphState:
        return cls(
            data=dict(snap.get("data", {})),
            visited=list(snap.get("visited", [])),
            current_node=snap.get("current_node"),
            last_output=snap.get("last_output"),
        )

GroupChat

Configuración del group chat (participantes + estrategia de turnos).

Source code in src/ciel/orchestration/chat.py

class GroupChat:
    """Configuración del group chat (participantes + estrategia de turnos)."""

    def __init__(
        self,
        participants: List[Agent],
        *,
        max_rounds: int = 12,
        selector: Optional[Callable[["GroupChatState", List[Agent]], str]] = None,
        terminate_keyword: str = "TERMINATE",
    ) -> None:
        if not participants:
            raise GroupChatError("group chat requires at least one participant")
        self.participants = list(participants)
        self._by_name = {p.name: p for p in participants}
        self.max_rounds = max_rounds
        self.selector = selector
        self.terminate_keyword = terminate_keyword
        # round-robin a partir del índice 0
        self._next_index = 0

    def participant_names(self) -> List[str]:
        return [p.name for p in self.participants]

    def get(self, name: str) -> Agent:
        if name not in self._by_name:
            raise GroupChatError(f"unknown participant '{name}'")
        return self._by_name[name]

    def select_next(self, state: GroupChatState) -> Agent:
        if self.selector is not None:
            name = self.selector(state, self.participants)
            return self.get(name)
        # round-robin determinista (estilo AutoGen default)
        agent = self.participants[self._next_index % len(self.participants)]
        self._next_index += 1
        return agent

GroupChatCheckpointStore

Persistencia del transcripto de group chat sobre MemoryStore.

Source code in src/ciel/orchestration/chat.py

class GroupChatCheckpointStore:
    """Persistencia del transcripto de group chat sobre ``MemoryStore``."""

    def __init__(self, memory_store: MemoryStore) -> None:
        self.memory = memory_store

    def _key(self, run_id: str) -> str:
        return f"chat:{run_id}"

    def save(self, *, run_id: str, state: GroupChatState, tenant_id: Optional[str], session_id: Optional[str]) -> str:
        import uuid

        checkpoint_id = str(uuid.uuid4())
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id or run_id,
            key=self._key(run_id),
            value={"checkpoint_id": checkpoint_id, "run_id": run_id, "state": state.snapshot()},
        )
        return checkpoint_id

    def load(self, *, run_id: str, tenant_id: Optional[str], session_id: Optional[str]) -> Optional[Dict[str, Any]]:
        payload = self.memory.get(tenant_id=tenant_id, session_id=session_id or run_id, key=self._key(run_id))
        return payload if isinstance(payload, dict) else None

GroupChatError

Bases: Exception

Source code in src/ciel/orchestration/chat.py

class GroupChatError(Exception):
    pass

GroupChatManager

Ejecuta el diálogo: selecciona hablante, recolecta réplica, evalúa terminación. Montado SOBRE Supervisor para heredar retry/timeout/budget por participante (cada réplica es un worker del supervisor).

Source code in src/ciel/orchestration/chat.py

class GroupChatManager:
    """Ejecuta el diálogo: selecciona hablante, recolecta réplica, evalúa
    terminación. Montado SOBRE ``Supervisor`` para heredar retry/timeout/budget
    por participante (cada réplica es un worker del supervisor)."""

    def __init__(
        self,
        chat: GroupChat,
        *,
        supervisor: Optional[Supervisor] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> None:
        self.chat = chat
        self.supervisor = supervisor or Supervisor()
        self.tenant_id = tenant_id
        self.session_id = session_id

    async def _reply_via_supervisor(self, agent: Agent, state: GroupChatState, round_no: int) -> str:
        async def _worker(ctx: WorkerContext) -> str:
            return await agent._reply(state)

        result = await self.supervisor.run(
            step_id=f"chat:{agent.name}:{round_no}",
            worker=_worker,
            payload={"agent": agent.name, "round": round_no},
            worker_id=agent.name,
        )
        if result.failed:
            raise GroupChatError(f"agent '{agent.name}' failed in round {round_no}: {result.error}")
        return str(result.output)

    async def run(self, *, initial_message: Optional[str] = None, initial_sender: Optional[str] = None) -> GroupChatState:
        state = GroupChatState()
        if initial_message and initial_sender:
            state.transcript.append(ChatMessage(role=initial_sender, content=initial_message, round=0))

        for round_no in range(1, self.chat.max_rounds + 1):
            speaker = self.chat.select_next(state)
            text = await self._reply_via_supervisor(speaker, state, round_no)
            msg = ChatMessage(role=speaker.name, content=text, round=round_no)
            state.transcript.append(msg)
            state.rounds = round_no

            if self.chat.terminate_keyword and self.chat.terminate_keyword in text:
                state.terminated = True
                state.terminator = speaker.name
                break
            if speaker._is_terminator(text):
                state.terminated = True
                state.terminator = speaker.name
                break

        if state.rounds >= self.chat.max_rounds and not state.terminated:
            # Se agotaron las rondas sin señal explícita: marca terminado por límite.
            state.terminated = True
            state.terminator = None
        return state

GroupChatState dataclass

Estado compartido del group chat: transcripto + metadata por ronda.

Source code in src/ciel/orchestration/chat.py

@dataclass
class GroupChatState:
    """Estado compartido del group chat: transcripto + metadata por ronda."""

    transcript: List[ChatMessage] = field(default_factory=list)
    rounds: int = 0
    terminated: bool = False
    terminator: Optional[str] = None

    def last_message(self) -> Optional[ChatMessage]:
        return self.transcript[-1] if self.transcript else None

    def snapshot(self) -> Dict[str, Any]:
        return {
            "transcript": [m.to_dict() for m in self.transcript],
            "rounds": self.rounds,
            "terminated": self.terminated,
            "terminator": self.terminator,
        }

    @classmethod
    def from_snapshot(cls, snap: Dict[str, Any]) -> "GroupChatState":
        return cls(
            transcript=[ChatMessage(**m) for m in snap.get("transcript", [])],
            rounds=int(snap.get("rounds", 0)),
            terminated=bool(snap.get("terminated", False)),
            terminator=snap.get("terminator"),
        )

RootAgent

Builder declarativo: add_specialist + set_router; luego compile.

El router decide a qué specialist enrutar la petición. Si devuelve None, el root_handler (opcional) la atiende localmente.

Source code in src/ciel/orchestration/root.py

class RootAgent:
    """Builder declarativo: ``add_specialist`` + ``set_router``; luego ``compile``.

    El ``router`` decide a qué specialist enrutar la petición. Si devuelve
    ``None``, el ``root_handler`` (opcional) la atiende localmente.
    """

    def __init__(self, name: str = "root") -> None:
        self.name = name
        self._specialists: Dict[str, Specialist] = {}
        self._router: Optional[RoutingFn] = None
        self._root_handler: Optional[HandlerFn] = None

    def add_specialist(self, specialist: Specialist) -> "RootAgent":
        if specialist.name in self._specialists:
            raise RootAgentError(f"specialist '{specialist.name}' already registered")
        self._specialists[specialist.name] = specialist
        return self

    def set_router(self, router: RoutingFn) -> "RootAgent":
        self._router = router
        return self

    def set_root_handler(self, handler: HandlerFn) -> "RootAgent":
        self._root_handler = handler
        return self

    def specialists(self) -> Dict[str, Specialist]:
        return dict(self._specialists)

    def compile(
        self,
        *,
        supervisor: Optional[Supervisor] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> "RootRunner":
        if self._router is None and self._root_handler is None:
            raise RootAgentError("root agent needs a router and/or a root_handler")
        return RootRunner(
            agent=self,
            supervisor=supervisor or Supervisor(),
            tenant_id=tenant_id,
            session_id=session_id,
        )

RootAgentError

Bases: Exception

Source code in src/ciel/orchestration/root.py

class RootAgentError(Exception):
    pass

RootCheckpointStore

Persistencia de la decisión de enrutamiento sobre MemoryStore.

Source code in src/ciel/orchestration/root.py

class RootCheckpointStore:
    """Persistencia de la decisión de enrutamiento sobre ``MemoryStore``."""

    def __init__(self, memory_store: MemoryStore) -> None:
        self.memory = memory_store

    def _key(self, run_id: str) -> str:
        return f"root:{run_id}"

    def save(self, *, run_id: str, state: RootState, tenant_id: Optional[str], session_id: Optional[str]) -> str:
        import uuid

        checkpoint_id = str(uuid.uuid4())
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id or run_id,
            key=self._key(run_id),
            value={"checkpoint_id": checkpoint_id, "run_id": run_id, "state": state.snapshot()},
        )
        return checkpoint_id

    def load(self, *, run_id: str, tenant_id: Optional[str], session_id: Optional[str]) -> Optional[Dict[str, Any]]:
        payload = self.memory.get(tenant_id=tenant_id, session_id=session_id or run_id, key=self._key(run_id))
        return payload if isinstance(payload, dict) else None

RootRunner

Ejecuta el root agent: enruta la petición y ejecuta el specialist (o root).

Source code in src/ciel/orchestration/root.py

class RootRunner:
    """Ejecuta el root agent: enruta la petición y ejecuta el specialist (o root)."""

    def __init__(
        self,
        *,
        agent: RootAgent,
        supervisor: Supervisor,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
        session_store: Optional[SessionStore] = None,
    ) -> None:
        self.agent = agent
        self.supervisor = supervisor
        self.tenant_id = tenant_id
        self.session_id = session_id
        self.session_store = session_store or (SessionStore(supervisor._memory) if hasattr(supervisor, "_memory") else None)

    async def _run_worker(self, worker_id: str, fn: Callable[["RootState"], Any], state: RootState) -> Any:
        async def _worker(ctx: WorkerContext) -> Any:
            return await self._invoke(fn, state)

        result = await self.supervisor.run(
            step_id=f"root:{worker_id}",
            worker=_worker,
            payload={"worker": worker_id},
            worker_id=worker_id,
        )
        if result.failed:
            raise RootAgentError(f"worker '{worker_id}' failed after {result.attempts} attempts: {result.error}")
        return result.output

    @staticmethod
    async def _invoke(fn: Callable[["RootState"], Any], state: RootState) -> Any:
        res = fn(state)
        if hasattr(res, "__await__"):
            return await res
        return res

    async def route(
        self,
        prompt: str,
        *,
        session_id: Optional[str] = None,
        session_store: Optional[SessionStore] = None,
        tenant_id: Optional[str] = None,
    ) -> RootState:
        """Enruta y ejecuta la petición, manteniendo session state por tenant.

        Si se pasa ``session_id`` (+ ``session_store``/``tenant_id``), el estado
        del agente se reconstruye con el historial de turnos previos y el turno
        resultante se persiste para la próxima invocación (estilo ADK session
        state entre turnos). OFFLINE-SAFE: no requiere red ni proveedor.
        """
        sid = session_id if session_id is not None else self.session_id
        store = session_store if session_store is not None else self.session_store
        tid = tenant_id if tenant_id is not None else self.tenant_id

        # Rehidrata el historial de turnos previos de la session (si existe).
        prior_history: List[Dict[str, Any]] = []
        if store is not None and sid is not None:
            prior_history = store.history(tenant_id=tid, session_id=sid)

        state = RootState(prompt=prompt, history=list(prior_history))
        router = self.agent._router
        if router is not None:
            target = router(prompt)
            if isinstance(target, Awaitable):
                target = await target
            if target is not None:
                if target not in self.agent._specialists:
                    raise RootAgentError(f"router returned unknown specialist '{target}'")
                state.route = target
                state.result = await self._run_worker(target, self.agent._specialists[target]._handle, state)
                state.metadata["handled_by"] = target
                self._persist_turn(store, tid, sid, state)
                return state
        # Sin enrutamiento -> lo maneja el root (si lo hay).
        if self.agent._root_handler is not None:
            state.handled_by_root = True
            state.result = await self._run_worker("root", self.agent._root_handler, state)
            state.metadata["handled_by"] = self.agent.name
            self._persist_turn(store, tid, sid, state)
            return state
        raise RootAgentError("router returned None and no root_handler is configured")

    @staticmethod
    def _persist_turn(
        store: Optional[SessionStore],
        tid: Optional[str],
        sid: Optional[str],
        state: RootState,
    ) -> None:
        """Persiste el turno en la session si hay session store + id."""
        if store is None or sid is None:
            return
        turn = {
            "prompt": state.prompt,
            "route": state.route,
            "result": state.result,
            "handled_by_root": state.handled_by_root,
            "handled_by": state.metadata.get("handled_by"),
        }
        store.append_turn(tenant_id=tid, session_id=sid, turn=turn)

route(prompt: str, *, session_id: Optional[str] = None, session_store: Optional[SessionStore] = None, tenant_id: Optional[str] = None) -> RootState async

Enruta y ejecuta la petición, manteniendo session state por tenant.

Si se pasa session_id (+ session_store/tenant_id), el estado del agente se reconstruye con el historial de turnos previos y el turno resultante se persiste para la próxima invocación (estilo ADK session state entre turnos). OFFLINE-SAFE: no requiere red ni proveedor.

Source code in src/ciel/orchestration/root.py

async def route(
    self,
    prompt: str,
    *,
    session_id: Optional[str] = None,
    session_store: Optional[SessionStore] = None,
    tenant_id: Optional[str] = None,
) -> RootState:
    """Enruta y ejecuta la petición, manteniendo session state por tenant.

    Si se pasa ``session_id`` (+ ``session_store``/``tenant_id``), el estado
    del agente se reconstruye con el historial de turnos previos y el turno
    resultante se persiste para la próxima invocación (estilo ADK session
    state entre turnos). OFFLINE-SAFE: no requiere red ni proveedor.
    """
    sid = session_id if session_id is not None else self.session_id
    store = session_store if session_store is not None else self.session_store
    tid = tenant_id if tenant_id is not None else self.tenant_id

    # Rehidrata el historial de turnos previos de la session (si existe).
    prior_history: List[Dict[str, Any]] = []
    if store is not None and sid is not None:
        prior_history = store.history(tenant_id=tid, session_id=sid)

    state = RootState(prompt=prompt, history=list(prior_history))
    router = self.agent._router
    if router is not None:
        target = router(prompt)
        if isinstance(target, Awaitable):
            target = await target
        if target is not None:
            if target not in self.agent._specialists:
                raise RootAgentError(f"router returned unknown specialist '{target}'")
            state.route = target
            state.result = await self._run_worker(target, self.agent._specialists[target]._handle, state)
            state.metadata["handled_by"] = target
            self._persist_turn(store, tid, sid, state)
            return state
    # Sin enrutamiento -> lo maneja el root (si lo hay).
    if self.agent._root_handler is not None:
        state.handled_by_root = True
        state.result = await self._run_worker("root", self.agent._root_handler, state)
        state.metadata["handled_by"] = self.agent.name
        self._persist_turn(store, tid, sid, state)
        return state
    raise RootAgentError("router returned None and no root_handler is configured")

RootState dataclass

Estado compartido del root agent: prompt, ruta elegida y resultado.

history acumula los turnos previos de la session (estilo ADK session state) para que el agente recuerde contexto entre invocaciones.

Source code in src/ciel/orchestration/root.py

@dataclass
class RootState:
    """Estado compartido del root agent: prompt, ruta elegida y resultado.

    ``history`` acumula los turnos previos de la session (estilo ADK session
    state) para que el agente recuerde contexto entre invocaciones.
    """

    prompt: str = ""
    route: Optional[str] = None
    result: Any = None
    handled_by_root: bool = False
    metadata: Dict[str, Any] = field(default_factory=dict)
    history: List[Dict[str, Any]] = field(default_factory=list)

    def snapshot(self) -> Dict[str, Any]:
        return {
            "prompt": self.prompt,
            "route": self.route,
            "result": self.result,
            "handled_by_root": self.handled_by_root,
            "metadata": dict(self.metadata),
            "history": [dict(h) for h in self.history],
        }

    @classmethod
    def from_snapshot(cls, snap: Dict[str, Any]) -> "RootState":
        return cls(
            prompt=snap.get("prompt", ""),
            route=snap.get("route"),
            result=snap.get("result"),
            handled_by_root=bool(snap.get("handled_by_root", False)),
            metadata=dict(snap.get("metadata", {})),
            history=[dict(h) for h in snap.get("history", [])],
        )

SessionError

Bases: Exception

Source code in src/ciel/orchestration/session.py

class SessionError(Exception):
    pass

SessionStore

Estado de session durable por tenant, sobre MemoryStore.

Claves internas (todas namespaced con session: para no colisionar con otros módulos que usan el mismo MemoryStore):

• session:<sid>:turns -> {"version", "turns": [turn, ...]}
• session:<sid>:state -> {key: value} (estado arbitrario)
• session:<sid>:board_links-> {"links": [board_task_id, ...]}
• session_index:<tenant> -> {"sessions": [sid, ...]} (índice)

Source code in src/ciel/orchestration/session.py

class SessionStore:
    """Estado de session durable por tenant, sobre ``MemoryStore``.

    Claves internas (todas namespaced con ``session:`` para no colisionar con
    otros módulos que usan el mismo ``MemoryStore``):

    * ``session:<sid>:turns``      -> {"version", "turns": [turn, ...]}
    * ``session:<sid>:state``      -> {key: value}  (estado arbitrario)
    * ``session:<sid>:board_links``-> {"links": [board_task_id, ...]}
    * ``session_index:<tenant>``   -> {"sessions": [sid, ...]}  (índice)
    """

    def __init__(self, memory_store: MemoryStore) -> None:
        self.memory = memory_store

    # -- keys ---------------------------------------------------------------
    @staticmethod
    def _turns_key(session_id: str) -> str:
        return f"session:{session_id}:turns"

    @staticmethod
    def _state_key(session_id: str) -> str:
        return f"session:{session_id}:state"

    @staticmethod
    def _links_key(session_id: str) -> str:
        return f"session:{session_id}:board_links"

    @staticmethod
    def _index_key(tenant_id: Optional[str]) -> str:
        sentinel = tenant_id if tenant_id is not None else "__none__"
        return f"session_index:{sentinel}"

    # -- index --------------------------------------------------------------
    def _load_index(self, tenant_id: Optional[str]) -> List[str]:
        # El índice se guarda con session_id == tenant_id (determinista) para
        # no depender de una session concreta.
        sid = tenant_id if tenant_id is not None else "__none__"
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=sid, key=self._index_key(tenant_id)
        )
        if isinstance(payload, dict):
            return list(payload.get("sessions", []))
        return []

    def _register(self, *, tenant_id: Optional[str], session_id: str) -> None:
        sessions = self._load_index(tenant_id)
        if session_id not in sessions:
            sessions.append(session_id)
            sid = tenant_id if tenant_id is not None else "__none__"
            self.memory.set(
                tenant_id=tenant_id,
                session_id=sid,
                key=self._index_key(tenant_id),
                value={"sessions": sessions},
            )

    # -- turns --------------------------------------------------------------
    def append_turn(
        self, *, tenant_id: Optional[str], session_id: str, turn: Dict[str, Any]
    ) -> None:
        """Añade un turno al historial de la session y lo registra en el índice."""
        turn = dict(turn)
        turn.setdefault("ts", time.time())
        turns = self.history(tenant_id=tenant_id, session_id=session_id)
        turns.append(turn)
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id,
            key=self._turns_key(session_id),
            value={"version": SESSION_VERSION, "turns": turns},
        )
        self._register(tenant_id=tenant_id, session_id=session_id)

    def history(self, *, tenant_id: Optional[str], session_id: str) -> List[Dict[str, Any]]:
        """Recupera el historial de turnos (vacío si la session no existe)."""
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=session_id, key=self._turns_key(session_id)
        )
        if isinstance(payload, dict):
            return [dict(t) for t in payload.get("turns", [])]
        return []

    # -- arbitrary state ----------------------------------------------------
    def _load_state_blob(self, tenant_id: Optional[str], session_id: str) -> Dict[str, Any]:
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=session_id, key=self._state_key(session_id)
        )
        return dict(payload) if isinstance(payload, dict) else {}

    def save_state(
        self, *, tenant_id: Optional[str], session_id: str, key: str, value: Any
    ) -> None:
        state = self._load_state_blob(tenant_id, session_id)
        state[key] = value
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id,
            key=self._state_key(session_id),
            value=state,
        )
        self._register(tenant_id=tenant_id, session_id=session_id)

    def load_state(
        self, *, tenant_id: Optional[str], session_id: str, key: str, default: Any = None
    ) -> Any:
        state = self._load_state_blob(tenant_id, session_id)
        return state.get(key, default)

    # -- board integration --------------------------------------------------
    def link_board_task(
        self, *, tenant_id: Optional[str], session_id: str, board_task_id: str
    ) -> None:
        """Vincula una tarea del board (mismo tenant) con la session."""
        links = self.board_links(tenant_id=tenant_id, session_id=session_id)
        if board_task_id not in links:
            links.append(board_task_id)
            self.memory.set(
                tenant_id=tenant_id,
                session_id=session_id,
                key=self._links_key(session_id),
                value={"links": links},
            )

    def board_links(self, *, tenant_id: Optional[str], session_id: str) -> List[str]:
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=session_id, key=self._links_key(session_id)
        )
        if isinstance(payload, dict):
            return list(payload.get("links", []))
        return []

    # -- listing ------------------------------------------------------------
    def list_sessions(self, *, tenant_id: Optional[str]) -> List[str]:
        """Lista los ids de session registrados para un tenant."""
        return self._load_index(tenant_id)

append_turn(*, tenant_id: Optional[str], session_id: str, turn: Dict[str, Any]) -> None

Añade un turno al historial de la session y lo registra en el índice.

Source code in src/ciel/orchestration/session.py

def append_turn(
    self, *, tenant_id: Optional[str], session_id: str, turn: Dict[str, Any]
) -> None:
    """Añade un turno al historial de la session y lo registra en el índice."""
    turn = dict(turn)
    turn.setdefault("ts", time.time())
    turns = self.history(tenant_id=tenant_id, session_id=session_id)
    turns.append(turn)
    self.memory.set(
        tenant_id=tenant_id,
        session_id=session_id,
        key=self._turns_key(session_id),
        value={"version": SESSION_VERSION, "turns": turns},
    )
    self._register(tenant_id=tenant_id, session_id=session_id)

history(*, tenant_id: Optional[str], session_id: str) -> List[Dict[str, Any]]

Recupera el historial de turnos (vacío si la session no existe).

Source code in src/ciel/orchestration/session.py

def history(self, *, tenant_id: Optional[str], session_id: str) -> List[Dict[str, Any]]:
    """Recupera el historial de turnos (vacío si la session no existe)."""
    payload = self.memory.get(
        tenant_id=tenant_id, session_id=session_id, key=self._turns_key(session_id)
    )
    if isinstance(payload, dict):
        return [dict(t) for t in payload.get("turns", [])]
    return []

link_board_task(*, tenant_id: Optional[str], session_id: str, board_task_id: str) -> None

Vincula una tarea del board (mismo tenant) con la session.

Source code in src/ciel/orchestration/session.py

def link_board_task(
    self, *, tenant_id: Optional[str], session_id: str, board_task_id: str
) -> None:
    """Vincula una tarea del board (mismo tenant) con la session."""
    links = self.board_links(tenant_id=tenant_id, session_id=session_id)
    if board_task_id not in links:
        links.append(board_task_id)
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id,
            key=self._links_key(session_id),
            value={"links": links},
        )

list_sessions(*, tenant_id: Optional[str]) -> List[str]

Lista los ids de session registrados para un tenant.

Source code in src/ciel/orchestration/session.py

def list_sessions(self, *, tenant_id: Optional[str]) -> List[str]:
    """Lista los ids de session registrados para un tenant."""
    return self._load_index(tenant_id)

Specialist dataclass

Agente especialista enrutado por el root agent (ADK sub_agent).

Source code in src/ciel/orchestration/root.py

@dataclass
class Specialist:
    """Agente especialista enrutado por el root agent (ADK sub_agent)."""

    name: str
    handler: HandlerFn
    description: str = ""

    async def _handle(self, state: RootState) -> Any:
        res = self.handler(state)
        if hasattr(res, "__await__"):
            return await res
        return res

StateGraph

Grafo de estado explícito, cíclico/condicional, con checkpoint.

• add_node registra un nodo ejecutable.
• add_edge registra una transición incondicional.
• add_conditional_edges registra transiciones con guarda booleana.
• set_entry_point / set_finish_point definen inicio y fin.

Source code in src/ciel/orchestration/graph.py

class StateGraph:
    """Grafo de estado explícito, cíclico/condicional, con checkpoint.

    - ``add_node`` registra un nodo ejecutable.
    - ``add_edge`` registra una transición incondicional.
    - ``add_conditional_edges`` registra transiciones con guarda booleana.
    - ``set_entry_point`` / ``set_finish_point`` definen inicio y fin.
    """

    def __init__(self, name: str = "graph") -> None:
        self.name = name
        self._nodes: Dict[str, GraphNode] = {}
        self._edges: List[GraphEdge] = []
        self._entry: Optional[str] = None
        self._finish: Optional[str] = None

    def add_node(
        self,
        node_id: str,
        fn: NodeFn,
        *,
        worker_id: str = "node",
        supervisor_kwargs: Optional[Dict[str, Any]] = None,
        require_approval: Optional[str] = None,
    ) -> "StateGraph":
        if node_id in self._nodes:
            raise GraphError(f"node '{node_id}' already exists")
        self._nodes[node_id] = GraphNode(
            id=node_id,
            fn=fn,
            worker_id=worker_id,
            supervisor_kwargs=supervisor_kwargs or {},
            require_approval=require_approval,
        )
        return self

    def add_edge(self, source: str, target: str) -> "StateGraph":
        self._require_node(source)
        self._require_node(target)
        self._edges.append(GraphEdge(source=source, target=target))
        return self

    def add_conditional_edges(
        self,
        source: str,
        targets: Sequence[str],
        condition: EdgeCondition,
    ) -> "StateGraph":
        self._require_node(source)
        for t in targets:
            self._require_node(t)
            self._edges.append(GraphEdge(source=source, target=t, condition=condition))
        return self

    def set_entry_point(self, node_id: str) -> "StateGraph":
        self._require_node(node_id)
        self._entry = node_id
        return self

    def set_finish_point(self, node_id: str) -> "StateGraph":
        self._require_node(node_id)
        self._finish = node_id
        return self

    @property
    def nodes(self) -> Dict[str, GraphNode]:
        return dict(self._nodes)

    def outgoing(self, node_id: str) -> List[GraphEdge]:
        return [e for e in self._edges if e.source == node_id]

    def _require_node(self, node_id: str) -> None:
        if node_id not in self._nodes:
            raise GraphError(f"unknown node '{node_id}'")

    def compile(
        self,
        *,
        supervisor: Optional[Supervisor] = None,
        max_steps: int = 64,
        checkpointer: Optional["GraphCheckpointStore"] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> "GraphRunner":
        if self._entry is None:
            raise GraphError("entry point not set; call set_entry_point(...)")
        return GraphRunner(
            graph=self,
            supervisor=supervisor or Supervisor(),
            max_steps=max_steps,
            checkpointer=checkpointer,
            tenant_id=tenant_id,
            session_id=session_id,
        )

Task dataclass

Unidad de trabajo autónomo durable (estilo ADK/AutoGen task).

status: pending | running | succeeded | failed. attempts se incrementa por cada ejecución del handler.

Source code in src/ciel/orchestration/agent.py

@dataclass
class Task:
    """Unidad de trabajo autónomo durable (estilo ADK/AutoGen task).

    ``status``: ``pending`` | ``running`` | ``succeeded`` | ``failed``.
    ``attempts`` se incrementa por cada ejecución del handler.
    """

    goal: str
    payload: Dict[str, Any] = field(default_factory=dict)
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    status: str = "pending"
    attempts: int = 0
    result: Any = None
    error: Optional[str] = None
    created_at: float = field(default_factory=lambda: time.time())
    updated_at: float = field(default_factory=lambda: time.time())

    def snapshot(self) -> Dict[str, Any]:
        return {
            "id": self.id,
            "goal": self.goal,
            "payload": dict(self.payload),
            "status": self.status,
            "attempts": self.attempts,
            "result": self.result,
            "error": self.error,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
        }

    @classmethod
    def from_snapshot(cls, snap: Dict[str, Any]) -> "Task":
        return cls(
            id=snap.get("id", str(uuid.uuid4())),
            goal=snap.get("goal", ""),
            payload=dict(snap.get("payload", {})),
            status=snap.get("status", "pending"),
            attempts=snap.get("attempts", 0),
            result=snap.get("result"),
            error=snap.get("error"),
            created_at=snap.get("created_at", time.time()),
            updated_at=snap.get("updated_at", time.time()),
        )

    def mark_running(self) -> None:
        self.status = "running"
        self.updated_at = time.time()

    def mark_succeeded(self, result: Any) -> None:
        self.status = "succeeded"
        self.result = result
        self.error = None
        self.updated_at = time.time()

    def mark_failed(self, error: str) -> None:
        self.status = "failed"
        self.error = error
        self.updated_at = time.time()

TaskError

Bases: AgentError

Error de una tarea individual tras agotar reintentos.

Source code in src/ciel/orchestration/agent.py

class TaskError(AgentError):
    """Error de una tarea individual tras agotar reintentos."""

EdgeCondition = Callable[[Dict[str, Any]], bool] module-attribute

NodeFn = Callable[[Dict[str, Any]], Awaitable[Any]] module-attribute

Worker = Callable[[WorkerContext], Any] module-attribute

GraphApprovalDenied

Bases: Exception

Lanzada cuando un nodo pausado por HIL es denegado por el aprobadador.

Source code in src/ciel/orchestration/graph.py

class GraphApprovalDenied(Exception):
    """Lanzada cuando un nodo pausado por HIL es denegado por el aprobadador."""

GraphCheckpointStore

Persistencia de checkpoints del grafo sobre MemoryStore.

Mismo patrón que ciel.runtime.checkpoints.CheckpointStore: usa la clave (tenant_id, session_id, key) ya con multitenancy nativo.

Source code in src/ciel/orchestration/graph.py

class GraphCheckpointStore:
    """Persistencia de checkpoints del grafo sobre ``MemoryStore``.

    Mismo patrón que ``ciel.runtime.checkpoints.CheckpointStore``: usa la
    clave ``(tenant_id, session_id, key)`` ya con multitenancy nativo.
    """

    def __init__(self, memory_store: MemoryStore) -> None:
        self.memory = memory_store

    def _key(self, run_id: str) -> str:
        return f"graph:{run_id}"

    def save(self, *, run_id: str, step_index: int, state: GraphState, finished: bool, tenant_id: Optional[str], session_id: Optional[str], paused: bool = False, paused_node: Optional[str] = None) -> str:
        checkpoint_id = str(uuid.uuid4())
        payload = {
            "checkpoint_id": checkpoint_id,
            "run_id": run_id,
            "step_index": step_index,
            "finished": finished,
            "paused": paused,
            "paused_node": paused_node,
            "state": state.snapshot(),
        }
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id or run_id,
            key=self._key(run_id),
            value=payload,
        )
        return checkpoint_id

    def load(self, *, run_id: str, tenant_id: Optional[str], session_id: Optional[str]) -> Optional[Dict[str, Any]]:
        payload = self.memory.get(
            tenant_id=tenant_id,
            session_id=session_id or run_id,
            key=self._key(run_id),
        )
        return payload if isinstance(payload, dict) else None

GraphEdge dataclass

Source code in src/ciel/orchestration/graph.py

@dataclass
class GraphEdge:
    source: str
    target: str
    condition: Optional[EdgeCondition] = None

GraphError

Bases: Exception

Source code in src/ciel/orchestration/graph.py

class GraphError(Exception):
    pass

GraphNode dataclass

Source code in src/ciel/orchestration/graph.py

@dataclass
class GraphNode:
    id: str
    fn: NodeFn
    worker_id: str = "node"
    supervisor_kwargs: Dict[str, Any] = field(default_factory=dict)
    # Human-in-the-loop: si se define, el runner pausa ANTES de ejecutar el
    # nodo y exige aprobación RBAC con esta acción (p. ej. "approve:deploy").
    require_approval: Optional[str] = None

GraphPaused

Bases: Exception

Lanzada por run cuando un nodo exige aprobación (HIL).

Atributos

node_id: id del nodo pausado. action: acción RBAC requerida para aprobar (p. ej. "approve:deploy"). run_id: id del run pausado (para reanudar con approve).

Source code in src/ciel/orchestration/graph.py

class GraphPaused(Exception):
    """Lanzada por ``run`` cuando un nodo exige aprobación (HIL).

    Atributos:
        node_id: id del nodo pausado.
        action: acción RBAC requerida para aprobar (p. ej. ``"approve:deploy"``).
        run_id: id del run pausado (para reanudar con ``approve``).
    """

    def __init__(self, message: str, *, node_id: str, action: str, run_id: str) -> None:
        super().__init__(message)
        self.node_id = node_id
        self.action = action
        self.run_id = run_id

GraphRunner

Ejecuta el grafo nodo a nodo, con checkpoint + reanudación + time-travel.

Source code in src/ciel/orchestration/graph.py

class GraphRunner:
    """Ejecuta el grafo nodo a nodo, con checkpoint + reanudación + time-travel."""

    def __init__(
        self,
        *,
        graph: StateGraph,
        supervisor: Supervisor,
        max_steps: int = 64,
        checkpointer: Optional[GraphCheckpointStore] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> None:
        self.graph = graph
        self.supervisor = supervisor
        self.max_steps = max_steps
        self.checkpointer = checkpointer
        self.tenant_id = tenant_id
        self.session_id = session_id
        self.run_id: Optional[str] = None

    async def _run_node(self, node: GraphNode, state: GraphState, step_index: int) -> WorkerResult:
        async def _worker(ctx: WorkerContext) -> Any:
            result = node.fn(state.data)
            if hasattr(result, "__await__"):
                return await result
            return result

        return await self.supervisor.run(
            step_id=f"{self.run_id}:{node.id}:{step_index}",
            worker=_worker,
            payload={"node": node.id, "step_index": step_index},
            worker_id=node.worker_id,
            **node.supervisor_kwargs,
        )

    def _next(self, node_id: str, state: GraphState) -> Optional[str]:
        edges = self.graph.outgoing(node_id)
        if not edges:
            return None
        for edge in edges:
            if edge.condition is None or edge.condition(state.data):
                return edge.target
        # No hubo arista tomada: si hay un finish point, detener; si no, error.
        if self.graph._finish == node_id:
            return None
        return None

    async def run(self, *, initial_data: Optional[Dict[str, Any]] = None, run_id: Optional[str] = None) -> GraphState:
        self.run_id = run_id or str(uuid.uuid4())
        state = GraphState(data=dict(initial_data or {}))
        current = self.graph._entry
        step_index = 0
        start = time.perf_counter()

        while current is not None and step_index < self.max_steps:
            node = self.graph._nodes[current]
            state.current_node = current

            # Human-in-the-loop: si el nodo exige aprobación, pausamos ANTES de
            # ejecutarlo y persistimos un checkpoint marcado como pausado.
            if node.require_approval is not None:
                if self.checkpointer is not None:
                    self.checkpointer.save(
                        run_id=self.run_id,
                        step_index=step_index,
                        state=state,
                        finished=False,
                        tenant_id=self.tenant_id,
                        session_id=self.session_id,
                        paused=True,
                        paused_node=current,
                    )
                raise GraphPaused(
                    f"node '{current}' requires approval ('{node.require_approval}')",
                    node_id=current,
                    action=node.require_approval,
                    run_id=self.run_id,
                )

            result = await self._run_node(node, state, step_index)
            if result.failed:
                raise GraphError(
                    f"node '{current}' failed after {result.attempts} attempts: {result.error}"
                )
            state.data[f"__out__{current}"] = result.output
            state.last_output = result.output
            state.visited.append(current)
            step_index += 1

            finished = current == self.graph._finish
            if self.checkpointer is not None:
                self.checkpointer.save(
                    run_id=self.run_id,
                    step_index=step_index,
                    state=state,
                    finished=finished,
                    tenant_id=self.tenant_id,
                    session_id=self.session_id,
                )

            if finished:
                break
            current = self._next(current, state)

        if step_index >= self.max_steps:
            raise GraphError(f"exceeded max_steps={self.max_steps} (possible cycle without finish)")

        state.current_node = current
        return state

    async def approve(
        self,
        run_id: str,
        *,
        approver: Optional[str] = None,
        rbac: "object | None" = None,
        action: Optional[str] = None,
    ) -> GraphState:
        """Reanuda un grafo pausado por HIL tras aprobar el nodo pendiente.

        Si se pasa ``rbac`` (un ``RBACEngine``), valida que ``approver`` tenga
        el permiso ``action`` (por defecto el ``require_approval`` del nodo
        pausado). Si no tiene permiso, lanza ``RBACError`` (reutiliza
        ``ciel.enterprise.rbac``). Tras aprobar, ejecuta el nodo y continúa el
        grafo desde allí.

        Requiere un checkpointer (el grafo debe haber sido pausado con uno).
        """
        if self.checkpointer is None:
            raise GraphError("approve requires a checkpointer (pause happened with one)")
        self.run_id = run_id
        payload = self.checkpointer.load(
            run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
        )
        if payload is None:
            raise GraphError(f"no checkpoint found for run_id '{run_id}'")
        if not payload.get("paused"):
            raise GraphError(f"run_id '{run_id}' is not paused (no HIL approval pending)")
        if payload.get("finished"):
            state = GraphState.from_snapshot(payload["state"])
            state.current_node = None
            return state

        paused_node = payload["paused_node"]
        node = self.graph._nodes.get(paused_node)
        if node is None:
            raise GraphError(f"paused node '{paused_node}' no longer exists in graph")
        action = action or node.require_approval
        if rbac is not None and action is not None:
            rbac.check(approver, action, tenant_id=self.tenant_id)

        state = GraphState.from_snapshot(payload["state"])
        step_index = int(payload.get("step_index", 0))
        # Ejecuta el nodo pausado y continúa desde allí (no re-ejecuta previos).
        state.current_node = paused_node
        result = await self._run_node(node, state, step_index)
        if result.failed:
            raise GraphError(
                f"node '{paused_node}' failed after approval: {result.error}"
            )
        state.data[f"__out__{paused_node}"] = result.output
        state.last_output = result.output
        state.visited.append(paused_node)
        step_index += 1
        current = self._next(paused_node, state)

        while current is not None and step_index < self.max_steps:
            node = self.graph._nodes[current]
            state.current_node = current
            result = await self._run_node(node, state, step_index)
            if result.failed:
                raise GraphError(
                    f"node '{current}' failed after {result.attempts} attempts: {result.error}"
                )
            state.data[f"__out__{current}"] = result.output
            state.last_output = result.output
            state.visited.append(current)
            step_index += 1
            finished = current == self.graph._finish
            self.checkpointer.save(
                run_id=self.run_id,
                step_index=step_index,
                state=state,
                finished=finished,
                tenant_id=self.tenant_id,
                session_id=self.session_id,
            )
            if finished:
                break
            current = self._next(current, state)

        if step_index >= self.max_steps:
            raise GraphError(f"exceeded max_steps={self.max_steps} during approve")
        state.current_node = current
        return state

    async def deny(self, run_id: str, *, reason: Optional[str] = None) -> None:
        """Marca un grafo pausado por HIL como denegado y detiene el flujo.

        Persiste el checkpoint como no-pausado y no-finalizado; el estado
        queda libre para inspección pero el grafo no continúa. Lanza
        ``GraphApprovalDenied`` para que el llamador sepa que fue rechazado.
        """
        if self.checkpointer is None:
            raise GraphError("deny requires a checkpointer")
        self.run_id = run_id
        payload = self.checkpointer.load(
            run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
        )
        if payload is None:
            raise GraphError(f"no checkpoint found for run_id '{run_id}'")
        if not payload.get("paused"):
            raise GraphError(f"run_id '{run_id}' is not paused")
        # Persiste como no-pausado para que resume/approve no lo retomen.
        state = GraphState.from_snapshot(payload["state"])
        self.checkpointer.save(
            run_id=run_id,
            step_index=int(payload.get("step_index", 0)),
            state=state,
            finished=False,
            tenant_id=self.tenant_id,
            session_id=self.session_id,
            paused=False,
            paused_node=None,
        )
        raise GraphApprovalDenied(
            f"approval denied for run_id '{run_id}'"
 + (f": {reason}" if reason else "")
        )

    async def resume(self, *, run_id: str) -> GraphState:
        """Reanuda un grafo interrumpido desde su último checkpoint.

        Reconstruye el estado persistido y continúa desde el último nodo
        visitado. Si el checkpoint marca ``finished=True``, devuelve el estado
        tal cual (idempotente).
        """
        if self.checkpointer is None:
            raise GraphError("resume requires a checkpointer")
        self.run_id = run_id
        payload = self.checkpointer.load(
            run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
        )
        if payload is None:
            raise GraphError(f"no checkpoint found for run_id '{run_id}'")
        if payload.get("finished"):
            state = GraphState.from_snapshot(payload["state"])
            state.current_node = None
            return state
        if payload.get("paused"):
            raise GraphPaused(
                f"run_id '{run_id}' is paused waiting for HIL approval",
                node_id=payload.get("paused_node") or "",
                action=(self.graph._nodes.get(payload.get("paused_node"))
                        and self.graph._nodes[payload["paused_node"]].require_approval)
                or "approve:*",
                run_id=run_id,
            )

        state = GraphState.from_snapshot(payload["state"])
        # Retomar desde el nodo SIGUIENTE al último completado (no re-ejecutar
        # el último visitado). Si no hay aristas salientes, el grafo terminó.
        if state.visited:
            current = self._next(state.visited[-1], state)
        else:
            current = self.graph._entry
        step_index = int(payload.get("step_index", 0))

        while current is not None and step_index < self.max_steps:
            node = self.graph._nodes[current]
            state.current_node = current
            result = await self._run_node(node, state, step_index)
            if result.failed:
                raise GraphError(
                    f"node '{current}' failed after {result.attempts} attempts: {result.error}"
                )
            state.data[f"__out__{current}"] = result.output
            state.last_output = result.output
            state.visited.append(current)
            step_index += 1

            finished = current == self.graph._finish
            self.checkpointer.save(
                run_id=self.run_id,
                step_index=step_index,
                state=state,
                finished=finished,
                tenant_id=self.tenant_id,
                session_id=self.session_id,
            )
            if finished:
                break
            current = self._next(current, state)

        if step_index >= self.max_steps:
            raise GraphError(f"exceeded max_steps={self.max_steps} during resume")
        state.current_node = current
        return state

    async def run_from(self, *, run_id: str, up_to_node: str, initial_data: Optional[Dict[str, Any]] = None) -> GraphState:
        """Time-travel: re-ejecuta el grafo desde el inicio hasta ``up_to_node``.

        Útil para reproducir/depurar un sub-tramo del flujo de forma
        determinista, con el mismo ``run_id`` (sobrescribe el checkpoint).
        """
        # Ejecuta normalmente pero se detiene al alcanzar up_to_node (inclusive).
        self.run_id = run_id
        state = GraphState(data=dict(initial_data or {}))
        current = self.graph._entry
        step_index = 0
        while current is not None and step_index < self.max_steps:
            node = self.graph._nodes[current]
            state.current_node = current
            result = await self._run_node(node, state, step_index)
            if result.failed:
                raise GraphError(f"node '{current}' failed: {result.error}")
            state.data[f"__out__{current}"] = result.output
            state.last_output = result.output
            state.visited.append(current)
            step_index += 1
            if self.checkpointer is not None:
                self.checkpointer.save(
                    run_id=self.run_id,
                    step_index=step_index,
                    state=state,
                    finished=(current == up_to_node),
                    tenant_id=self.tenant_id,
                    session_id=self.session_id,
                )
            if current == up_to_node:
                break
            current = self._next(current, state)
        state.current_node = current
        return state

approve(run_id: str, *, approver: Optional[str] = None, rbac: 'object | None' = None, action: Optional[str] = None) -> GraphState async

Reanuda un grafo pausado por HIL tras aprobar el nodo pendiente.

Si se pasa rbac (un RBACEngine), valida que approver tenga el permiso action (por defecto el require_approval del nodo pausado). Si no tiene permiso, lanza RBACError (reutiliza ciel.enterprise.rbac). Tras aprobar, ejecuta el nodo y continúa el grafo desde allí.

Requiere un checkpointer (el grafo debe haber sido pausado con uno).

Source code in src/ciel/orchestration/graph.py

async def approve(
    self,
    run_id: str,
    *,
    approver: Optional[str] = None,
    rbac: "object | None" = None,
    action: Optional[str] = None,
) -> GraphState:
    """Reanuda un grafo pausado por HIL tras aprobar el nodo pendiente.

    Si se pasa ``rbac`` (un ``RBACEngine``), valida que ``approver`` tenga
    el permiso ``action`` (por defecto el ``require_approval`` del nodo
    pausado). Si no tiene permiso, lanza ``RBACError`` (reutiliza
    ``ciel.enterprise.rbac``). Tras aprobar, ejecuta el nodo y continúa el
    grafo desde allí.

    Requiere un checkpointer (el grafo debe haber sido pausado con uno).
    """
    if self.checkpointer is None:
        raise GraphError("approve requires a checkpointer (pause happened with one)")
    self.run_id = run_id
    payload = self.checkpointer.load(
        run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
    )
    if payload is None:
        raise GraphError(f"no checkpoint found for run_id '{run_id}'")
    if not payload.get("paused"):
        raise GraphError(f"run_id '{run_id}' is not paused (no HIL approval pending)")
    if payload.get("finished"):
        state = GraphState.from_snapshot(payload["state"])
        state.current_node = None
        return state

    paused_node = payload["paused_node"]
    node = self.graph._nodes.get(paused_node)
    if node is None:
        raise GraphError(f"paused node '{paused_node}' no longer exists in graph")
    action = action or node.require_approval
    if rbac is not None and action is not None:
        rbac.check(approver, action, tenant_id=self.tenant_id)

    state = GraphState.from_snapshot(payload["state"])
    step_index = int(payload.get("step_index", 0))
    # Ejecuta el nodo pausado y continúa desde allí (no re-ejecuta previos).
    state.current_node = paused_node
    result = await self._run_node(node, state, step_index)
    if result.failed:
        raise GraphError(
            f"node '{paused_node}' failed after approval: {result.error}"
        )
    state.data[f"__out__{paused_node}"] = result.output
    state.last_output = result.output
    state.visited.append(paused_node)
    step_index += 1
    current = self._next(paused_node, state)

    while current is not None and step_index < self.max_steps:
        node = self.graph._nodes[current]
        state.current_node = current
        result = await self._run_node(node, state, step_index)
        if result.failed:
            raise GraphError(
                f"node '{current}' failed after {result.attempts} attempts: {result.error}"
            )
        state.data[f"__out__{current}"] = result.output
        state.last_output = result.output
        state.visited.append(current)
        step_index += 1
        finished = current == self.graph._finish
        self.checkpointer.save(
            run_id=self.run_id,
            step_index=step_index,
            state=state,
            finished=finished,
            tenant_id=self.tenant_id,
            session_id=self.session_id,
        )
        if finished:
            break
        current = self._next(current, state)

    if step_index >= self.max_steps:
        raise GraphError(f"exceeded max_steps={self.max_steps} during approve")
    state.current_node = current
    return state

deny(run_id: str, *, reason: Optional[str] = None) -> None async

Marca un grafo pausado por HIL como denegado y detiene el flujo.

Persiste el checkpoint como no-pausado y no-finalizado; el estado queda libre para inspección pero el grafo no continúa. Lanza GraphApprovalDenied para que el llamador sepa que fue rechazado.

Source code in src/ciel/orchestration/graph.py

   async def deny(self, run_id: str, *, reason: Optional[str] = None) -> None:
       """Marca un grafo pausado por HIL como denegado y detiene el flujo.

       Persiste el checkpoint como no-pausado y no-finalizado; el estado
       queda libre para inspección pero el grafo no continúa. Lanza
       ``GraphApprovalDenied`` para que el llamador sepa que fue rechazado.
       """
       if self.checkpointer is None:
           raise GraphError("deny requires a checkpointer")
       self.run_id = run_id
       payload = self.checkpointer.load(
           run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
       )
       if payload is None:
           raise GraphError(f"no checkpoint found for run_id '{run_id}'")
       if not payload.get("paused"):
           raise GraphError(f"run_id '{run_id}' is not paused")
       # Persiste como no-pausado para que resume/approve no lo retomen.
       state = GraphState.from_snapshot(payload["state"])
       self.checkpointer.save(
           run_id=run_id,
           step_index=int(payload.get("step_index", 0)),
           state=state,
           finished=False,
           tenant_id=self.tenant_id,
           session_id=self.session_id,
           paused=False,
           paused_node=None,
       )
       raise GraphApprovalDenied(
           f"approval denied for run_id '{run_id}'"
+ (f": {reason}" if reason else "")
       )

resume(*, run_id: str) -> GraphState async

Reanuda un grafo interrumpido desde su último checkpoint.

Reconstruye el estado persistido y continúa desde el último nodo visitado. Si el checkpoint marca finished=True, devuelve el estado tal cual (idempotente).

Source code in src/ciel/orchestration/graph.py

async def resume(self, *, run_id: str) -> GraphState:
    """Reanuda un grafo interrumpido desde su último checkpoint.

    Reconstruye el estado persistido y continúa desde el último nodo
    visitado. Si el checkpoint marca ``finished=True``, devuelve el estado
    tal cual (idempotente).
    """
    if self.checkpointer is None:
        raise GraphError("resume requires a checkpointer")
    self.run_id = run_id
    payload = self.checkpointer.load(
        run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
    )
    if payload is None:
        raise GraphError(f"no checkpoint found for run_id '{run_id}'")
    if payload.get("finished"):
        state = GraphState.from_snapshot(payload["state"])
        state.current_node = None
        return state
    if payload.get("paused"):
        raise GraphPaused(
            f"run_id '{run_id}' is paused waiting for HIL approval",
            node_id=payload.get("paused_node") or "",
            action=(self.graph._nodes.get(payload.get("paused_node"))
                    and self.graph._nodes[payload["paused_node"]].require_approval)
            or "approve:*",
            run_id=run_id,
        )

    state = GraphState.from_snapshot(payload["state"])
    # Retomar desde el nodo SIGUIENTE al último completado (no re-ejecutar
    # el último visitado). Si no hay aristas salientes, el grafo terminó.
    if state.visited:
        current = self._next(state.visited[-1], state)
    else:
        current = self.graph._entry
    step_index = int(payload.get("step_index", 0))

    while current is not None and step_index < self.max_steps:
        node = self.graph._nodes[current]
        state.current_node = current
        result = await self._run_node(node, state, step_index)
        if result.failed:
            raise GraphError(
                f"node '{current}' failed after {result.attempts} attempts: {result.error}"
            )
        state.data[f"__out__{current}"] = result.output
        state.last_output = result.output
        state.visited.append(current)
        step_index += 1

        finished = current == self.graph._finish
        self.checkpointer.save(
            run_id=self.run_id,
            step_index=step_index,
            state=state,
            finished=finished,
            tenant_id=self.tenant_id,
            session_id=self.session_id,
        )
        if finished:
            break
        current = self._next(current, state)

    if step_index >= self.max_steps:
        raise GraphError(f"exceeded max_steps={self.max_steps} during resume")
    state.current_node = current
    return state

run_from(*, run_id: str, up_to_node: str, initial_data: Optional[Dict[str, Any]] = None) -> GraphState async

Time-travel: re-ejecuta el grafo desde el inicio hasta up_to_node.

Útil para reproducir/depurar un sub-tramo del flujo de forma determinista, con el mismo run_id (sobrescribe el checkpoint).

Source code in src/ciel/orchestration/graph.py

async def run_from(self, *, run_id: str, up_to_node: str, initial_data: Optional[Dict[str, Any]] = None) -> GraphState:
    """Time-travel: re-ejecuta el grafo desde el inicio hasta ``up_to_node``.

    Útil para reproducir/depurar un sub-tramo del flujo de forma
    determinista, con el mismo ``run_id`` (sobrescribe el checkpoint).
    """
    # Ejecuta normalmente pero se detiene al alcanzar up_to_node (inclusive).
    self.run_id = run_id
    state = GraphState(data=dict(initial_data or {}))
    current = self.graph._entry
    step_index = 0
    while current is not None and step_index < self.max_steps:
        node = self.graph._nodes[current]
        state.current_node = current
        result = await self._run_node(node, state, step_index)
        if result.failed:
            raise GraphError(f"node '{current}' failed: {result.error}")
        state.data[f"__out__{current}"] = result.output
        state.last_output = result.output
        state.visited.append(current)
        step_index += 1
        if self.checkpointer is not None:
            self.checkpointer.save(
                run_id=self.run_id,
                step_index=step_index,
                state=state,
                finished=(current == up_to_node),
                tenant_id=self.tenant_id,
                session_id=self.session_id,
            )
        if current == up_to_node:
            break
        current = self._next(current, state)
    state.current_node = current
    return state

GraphState dataclass

Estado mutable compartido entre nodos (estilo state machine LangGraph).

Source code in src/ciel/orchestration/graph.py

@dataclass
class GraphState:
    """Estado mutable compartido entre nodos (estilo state machine LangGraph)."""

    data: Dict[str, Any] = field(default_factory=dict)
    visited: List[str] = field(default_factory=list)
    current_node: Optional[str] = None
    last_output: Any = None

    def snapshot(self) -> Dict[str, Any]:
        return {
            "data": dict(self.data),
            "visited": list(self.visited),
            "current_node": self.current_node,
            "last_output": self.last_output,
        }

    @classmethod
    def from_snapshot(cls, snap: Dict[str, Any]) -> GraphState:
        return cls(
            data=dict(snap.get("data", {})),
            visited=list(snap.get("visited", [])),
            current_node=snap.get("current_node"),
            last_output=snap.get("last_output"),
        )

MemoryStore

Bases: SqliteStateBackend

Alias retrocompatible a :class:SqliteStateBackend (SQLite en disco).

MemoryStore es SQLite en disco desde F5 — el nombre engaña. Para F15 se refactorizó para heredar de StateBackend de modo que cualquier store que hoy recibe un MemoryStore puede recibir también un PostgresStateBackend compartido sin cambios de API.

Se sigue construyendo con la misma firma: MemoryStore(db_path).

Source code in src/ciel/runtime/memory.py

class MemoryStore(SqliteStateBackend):
    """Alias retrocompatible a :class:`SqliteStateBackend` (SQLite en disco).

    ``MemoryStore`` es SQLite en disco desde F5 — el nombre engaña. Para
    F15 se refactorizó para heredar de ``StateBackend`` de modo que cualquier
    store que hoy recibe un ``MemoryStore`` puede recibir también un
    ``PostgresStateBackend`` compartido sin cambios de API.

    Se sigue construyendo con la misma firma: ``MemoryStore(db_path)``.
    """

    def __init__(self, db_path: str) -> None:
        super().__init__(db_path)

StateGraph

Grafo de estado explícito, cíclico/condicional, con checkpoint.

• add_node registra un nodo ejecutable.
• add_edge registra una transición incondicional.
• add_conditional_edges registra transiciones con guarda booleana.
• set_entry_point / set_finish_point definen inicio y fin.

Source code in src/ciel/orchestration/graph.py

class StateGraph:
    """Grafo de estado explícito, cíclico/condicional, con checkpoint.

    - ``add_node`` registra un nodo ejecutable.
    - ``add_edge`` registra una transición incondicional.
    - ``add_conditional_edges`` registra transiciones con guarda booleana.
    - ``set_entry_point`` / ``set_finish_point`` definen inicio y fin.
    """

    def __init__(self, name: str = "graph") -> None:
        self.name = name
        self._nodes: Dict[str, GraphNode] = {}
        self._edges: List[GraphEdge] = []
        self._entry: Optional[str] = None
        self._finish: Optional[str] = None

    def add_node(
        self,
        node_id: str,
        fn: NodeFn,
        *,
        worker_id: str = "node",
        supervisor_kwargs: Optional[Dict[str, Any]] = None,
        require_approval: Optional[str] = None,
    ) -> "StateGraph":
        if node_id in self._nodes:
            raise GraphError(f"node '{node_id}' already exists")
        self._nodes[node_id] = GraphNode(
            id=node_id,
            fn=fn,
            worker_id=worker_id,
            supervisor_kwargs=supervisor_kwargs or {},
            require_approval=require_approval,
        )
        return self

    def add_edge(self, source: str, target: str) -> "StateGraph":
        self._require_node(source)
        self._require_node(target)
        self._edges.append(GraphEdge(source=source, target=target))
        return self

    def add_conditional_edges(
        self,
        source: str,
        targets: Sequence[str],
        condition: EdgeCondition,
    ) -> "StateGraph":
        self._require_node(source)
        for t in targets:
            self._require_node(t)
            self._edges.append(GraphEdge(source=source, target=t, condition=condition))
        return self

    def set_entry_point(self, node_id: str) -> "StateGraph":
        self._require_node(node_id)
        self._entry = node_id
        return self

    def set_finish_point(self, node_id: str) -> "StateGraph":
        self._require_node(node_id)
        self._finish = node_id
        return self

    @property
    def nodes(self) -> Dict[str, GraphNode]:
        return dict(self._nodes)

    def outgoing(self, node_id: str) -> List[GraphEdge]:
        return [e for e in self._edges if e.source == node_id]

    def _require_node(self, node_id: str) -> None:
        if node_id not in self._nodes:
            raise GraphError(f"unknown node '{node_id}'")

    def compile(
        self,
        *,
        supervisor: Optional[Supervisor] = None,
        max_steps: int = 64,
        checkpointer: Optional["GraphCheckpointStore"] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> "GraphRunner":
        if self._entry is None:
            raise GraphError("entry point not set; call set_entry_point(...)")
        return GraphRunner(
            graph=self,
            supervisor=supervisor or Supervisor(),
            max_steps=max_steps,
            checkpointer=checkpointer,
            tenant_id=tenant_id,
            session_id=session_id,
        )

Supervisor

Source code in src/ciel/orchestration/supervisor.py

class Supervisor:
    def __init__(
        self,
        max_attempts: int = 2,
        timeout_s: float = 2.0,
        budget: Optional[Any] = None,
        rate_limiter: Optional[Any] = None,
        agent_counter: Optional[Any] = None,
        rate_limit: int = 0,
    ) -> None:
        self.max_attempts = max_attempts
        self.timeout_s = timeout_s
        self.budget = budget
        self.rate_limiter = rate_limiter
        self.agent_counter = agent_counter
        self.rate_limit = rate_limit
        self._results: Dict[str, WorkerResult] = {}

    async def run(
        self,
        step_id: str,
        worker: Worker,
        payload: Optional[Dict[str, Any]] = None,
        worker_id: str = "worker-1",
    ) -> WorkerResult:
        key = f"{step_id}:{worker_id}"
        if self.agent_counter is not None and self.budget is not None:
            exceeded = self.agent_counter.exceed(self.budget)
            if exceeded:
                result = WorkerResult(
                    worker_id=worker_id,
                    attempts=0,
                    failed=True,
                    error=f"budget rejection before run: {exceeded}",
                )
                self._results[key] = result
                return result
        if self.rate_limiter is not None and self.rate_limit > 0:
            rate_key = step_id or worker_id
            exceeded = self.rate_limiter.check(rate_key, self.rate_limit)
            if exceeded:
                result = WorkerResult(
                    worker_id=worker_id,
                    attempts=0,
                    failed=True,
                    error=f"rate limit rejection before run: {exceeded}",
                )
                self._results[key] = result
                return result
        attempt = 0
        start = time.perf_counter()
        while attempt < self.max_attempts:
            attempt += 1
            if self.agent_counter is not None:
                self.agent_counter.consume_tool(1)
            ctx = WorkerContext(step_id=step_id, worker_id=worker_id, payload=payload)
            try:
                output = await asyncio.wait_for(worker(ctx), timeout=self.timeout_s)
                result = WorkerResult(
                    worker_id=worker_id,
                    output=output,
                    attempts=attempt,
                    latency_ms=(time.perf_counter() - start) * 1000.0,
                )
                self._results[key] = result
                return result
            except Exception as exc:
                if attempt >= self.max_attempts:
                    result = WorkerResult(
                        worker_id=worker_id,
                        attempts=attempt,
                        latency_ms=(time.perf_counter() - start) * 1000.0,
                        failed=True,
                        error=str(exc),
                    )
                    self._results[key] = result
                    return result
        return self._results[key]

    def results(self) -> Dict[str, WorkerResult]:
        return dict(self._results)

WorkerContext dataclass

Source code in src/ciel/orchestration/supervisor.py

@dataclass
class WorkerContext:
    step_id: str
    worker_id: str
    payload: Optional[Dict[str, Any]] = None

WorkerResult dataclass

Source code in src/ciel/orchestration/supervisor.py

@dataclass
class WorkerResult:
    worker_id: str
    output: Any = None
    error: Optional[str] = None
    attempts: int = 0
    latency_ms: float = 0.0
    failed: bool = False

StepFn = Callable[['FlowState'], Any] module-attribute

SyncOrAsync = Any module-attribute

Worker = Callable[[WorkerContext], Any] module-attribute

_ = Worker module-attribute

Flow

Constructor declarativo de flows event-driven.

API: add_start, add_listen, add_router; luego compile. Los IDs de paso se derivan del __name__ de la función salvo que se pasen explícitamente.

Source code in src/ciel/orchestration/flows.py

class Flow:
    """Constructor declarativo de flows event-driven.

    API: ``add_start``, ``add_listen``, ``add_router``; luego ``compile``.
    Los IDs de paso se derivan del ``__name__`` de la función salvo que se
    pasen explícitamente.
    """

    def __init__(self, name: str = "flow") -> None:
        self.name = name
        self._steps: Dict[str, FlowStep] = {}
        self._order: List[str] = []
        self._sources: Dict[str, set] = {}  # paso -> conjunto de pasos fuente
        self._router_branches: Dict[str, Dict[Any, str]] = {}  # router -> {valor: destino}

    # -- registro ------------------------------------------------------------
    def add_start(self, fn: StepFn, step_id: Optional[str] = None) -> "Flow":
        sid = step_id or getattr(fn, "__name__", "start")
        self._register(FlowStep(id=sid, kind="start", fn=fn))
        self._sources[sid] = set()
        return self

    def add_listen(self, source_id: str, fn: StepFn, step_id: Optional[str] = None) -> "Flow":
        sid = step_id or getattr(fn, "__name__", "listen")
        self._register(FlowStep(id=sid, kind="listen", fn=fn, source=source_id))
        self._sources[sid] = {source_id}
        return self

    def add_router(
        self,
        source_id: str,
        fn: StepFn,
        branches: Dict[Any, str],
        step_id: Optional[str] = None,
    ) -> "Flow":
        sid = step_id or getattr(fn, "__name__", "router")
        self._register(FlowStep(id=sid, kind="router", fn=fn, source=source_id, branches=dict(branches)))
        self._sources[sid] = {source_id}
        self._router_branches[sid] = dict(branches)
        return self

    def add_branch(self, fn: StepFn, step_id: Optional[str] = None) -> "Flow":
        """Paso rama activado EXCLUSIVAMENTE por un router (no tiene fuente de datos)."""
        sid = step_id or getattr(fn, "__name__", "branch")
        self._register(FlowStep(id=sid, kind="branch", fn=fn))
        self._sources[sid] = set()
        return self

    def _register(self, step: FlowStep) -> None:
        if step.id in self._steps:
            raise FlowError(f"step '{step.id}' already registered")
        self._steps[step.id] = step
        self._order.append(step.id)

    def steps(self) -> Dict[str, FlowStep]:
        return dict(self._steps)

    def compile(
        self,
        *,
        supervisor: Optional[Supervisor] = None,
        max_steps: int = 256,
        checkpointer: Optional["FlowCheckpointStore"] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> "FlowRunner":
        # Validación de que los destinos de router existen (evita carreras en runtime).
        for rid, branches in self._router_branches.items():
            for tid in branches.values():
                if tid not in self._steps:
                    raise FlowError(f"router '{rid}' branch target '{tid}' is not registered")
        return FlowRunner(
            flow=self,
            supervisor=supervisor or Supervisor(),
            max_steps=max_steps,
            checkpointer=checkpointer,
            tenant_id=tenant_id,
            session_id=session_id,
        )

add_branch(fn: StepFn, step_id: Optional[str] = None) -> 'Flow'

Paso rama activado EXCLUSIVAMENTE por un router (no tiene fuente de datos).

Source code in src/ciel/orchestration/flows.py

def add_branch(self, fn: StepFn, step_id: Optional[str] = None) -> "Flow":
    """Paso rama activado EXCLUSIVAMENTE por un router (no tiene fuente de datos)."""
    sid = step_id or getattr(fn, "__name__", "branch")
    self._register(FlowStep(id=sid, kind="branch", fn=fn))
    self._sources[sid] = set()
    return self

FlowCheckpointStore

Persistencia de checkpoints de flow sobre MemoryStore (multitenancy nativo).

Source code in src/ciel/orchestration/flows.py

class FlowCheckpointStore:
    """Persistencia de checkpoints de flow sobre ``MemoryStore`` (multitenancy nativo)."""

    def __init__(self, memory_store: MemoryStore) -> None:
        self.memory = memory_store

    def _key(self, run_id: str) -> str:
        return f"flow:{run_id}"

    def save(
        self,
        *,
        run_id: str,
        completed: Sequence[str],
        state: FlowState,
        finished: bool,
        tenant_id: Optional[str],
        session_id: Optional[str],
    ) -> str:
        checkpoint_id = str(uuid.uuid4())
        payload = {
            "checkpoint_id": checkpoint_id,
            "run_id": run_id,
            "completed": list(completed),
            "finished": finished,
            "state": state.snapshot(),
        }
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id or run_id,
            key=self._key(run_id),
            value=payload,
        )
        return checkpoint_id

    def load(self, *, run_id: str, tenant_id: Optional[str], session_id: Optional[str]) -> Optional[Dict[str, Any]]:
        payload = self.memory.get(tenant_id=tenant_id, session_id=session_id or run_id, key=self._key(run_id))
        return payload if isinstance(payload, dict) else None

FlowError

Bases: Exception

Source code in src/ciel/orchestration/flows.py

class FlowError(Exception):
    pass

FlowRunner

Ejecuta el flow paso a paso (event-driven) con checkpoint + resume.

Source code in src/ciel/orchestration/flows.py

class FlowRunner:
    """Ejecuta el flow paso a paso (event-driven) con checkpoint + resume."""

    def __init__(
        self,
        *,
        flow: Flow,
        supervisor: Supervisor,
        max_steps: int = 256,
        checkpointer: Optional[FlowCheckpointStore] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> None:
        self.flow = flow
        self.supervisor = supervisor
        self.max_steps = max_steps
        self.checkpointer = checkpointer
        self.tenant_id = tenant_id
        self.session_id = session_id
        self.run_id: Optional[str] = None
        self._activated: set = set()

    async def _run_step(self, step: FlowStep, state: FlowState, idx: int) -> Any:
        async def _worker(ctx: WorkerContext) -> Any:
            res = step.fn(state)
            if hasattr(res, "__await__"):
                return await res
            return res

        result = await self.supervisor.run(
            step_id=f"{self.run_id}:{step.id}:{idx}",
            worker=_worker,
            payload={"step": step.id, "kind": step.kind},
            worker_id=step.id,
        )
        if result.failed:
            raise FlowError(f"step '{step.id}' failed after {result.attempts} attempts: {result.error}")
        return result.output

    def _ready(self, state: FlowState) -> List[str]:
        out: List[str] = []
        for sid in self.flow._order:
            if sid in state.completed:
                continue
            step = self.flow._steps[sid]
            # Una rama SOLO está lista si un router la activó explícitamente.
            if step.kind == "branch":
                if sid in self._activated:
                    out.append(sid)
                continue
            if sid in self._activated:
                out.append(sid)
                continue
            srcs = self.flow._sources.get(sid, set())
            if all(s in state.completed for s in srcs):
                out.append(sid)
        return out

    async def _drive(self, state: FlowState) -> FlowState:
        idx = 0
        while True:
            ready = self._ready(state)
            if not ready:
                break
            sid = ready[0]  # determinista: orden de registro
            step = self.flow._steps[sid]
            out = await self._run_step(step, state, idx)
            state.results[sid] = out
            state.data.setdefault("__out__", {})[sid] = out
            state.completed.append(sid)
            state.last_event = sid
            idx += 1

            # Un router activa exactamente una rama tras completarse.
            if step.kind == "router":
                target = self.flow._router_branches[sid].get(out)
                if target is None:
                    raise FlowError(f"router '{sid}' returned {out!r} with no matching branch")
                self._activated.add(target)

            if self.checkpointer is not None:
                self.checkpointer.save(
                    run_id=self.run_id,
                    completed=state.completed,
                    state=state,
                    finished=False,
                    tenant_id=self.tenant_id,
                    session_id=self.session_id,
                )
            if idx >= self.max_steps:
                raise FlowError(f"exceeded max_steps={self.max_steps} (possible cycle)")
        state.last_event = state.completed[-1] if state.completed else None
        return state

    async def run(self, *, initial_data: Optional[Dict[str, Any]] = None, run_id: Optional[str] = None) -> FlowState:
        self.run_id = run_id or str(uuid.uuid4())
        self._activated = set()
        state = FlowState(data=dict(initial_data or {}))
        return await self._drive(state)

    async def resume(self, *, run_id: str) -> FlowState:
        """Reanuda un flow interrumpido desde su último checkpoint.

        Reconstruye el estado persistido y recomputa las ramas de router ya
        activadas (derivables de los routers completados + sus resultados).
        """
        if self.checkpointer is None:
            raise FlowError("resume requires a checkpointer")
        self.run_id = run_id
        payload = self.checkpointer.load(run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id)
        if payload is None:
            raise FlowError(f"no checkpoint found for run_id '{run_id}'")
        state = FlowState.from_snapshot(payload["state"])
        self._activated = set()
        for rid, branches in self.flow._router_branches.items():
            if rid in state.completed:
                target = branches.get(state.results.get(rid))
                if target:
                    self._activated.add(target)
        return await self._drive(state)

resume(*, run_id: str) -> FlowState async

Reanuda un flow interrumpido desde su último checkpoint.

Reconstruye el estado persistido y recomputa las ramas de router ya activadas (derivables de los routers completados + sus resultados).

Source code in src/ciel/orchestration/flows.py

async def resume(self, *, run_id: str) -> FlowState:
    """Reanuda un flow interrumpido desde su último checkpoint.

    Reconstruye el estado persistido y recomputa las ramas de router ya
    activadas (derivables de los routers completados + sus resultados).
    """
    if self.checkpointer is None:
        raise FlowError("resume requires a checkpointer")
    self.run_id = run_id
    payload = self.checkpointer.load(run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id)
    if payload is None:
        raise FlowError(f"no checkpoint found for run_id '{run_id}'")
    state = FlowState.from_snapshot(payload["state"])
    self._activated = set()
    for rid, branches in self.flow._router_branches.items():
        if rid in state.completed:
            target = branches.get(state.results.get(rid))
            if target:
                self._activated.add(target)
    return await self._drive(state)

FlowState dataclass

Estado mutable compartido entre pasos de un flow (estilo CrewAI state).

Source code in src/ciel/orchestration/flows.py

@dataclass
class FlowState:
    """Estado mutable compartido entre pasos de un flow (estilo CrewAI state)."""

    data: Dict[str, Any] = field(default_factory=dict)
    results: Dict[str, Any] = field(default_factory=dict)
    completed: List[str] = field(default_factory=list)
    last_event: Optional[str] = None

    def snapshot(self) -> Dict[str, Any]:
        return {
            "data": dict(self.data),
            "results": dict(self.results),
            "completed": list(self.completed),
            "last_event": self.last_event,
        }

    @classmethod
    def from_snapshot(cls, snap: Dict[str, Any]) -> "FlowState":
        return cls(
            data=dict(snap.get("data", {})),
            results=dict(snap.get("results", {})),
            completed=list(snap.get("completed", [])),
            last_event=snap.get("last_event"),
        )

FlowStep dataclass

Source code in src/ciel/orchestration/flows.py

@dataclass
class FlowStep:
    id: str
    kind: str  # "start" | "listen" | "router"
    fn: StepFn
    source: Optional[str] = None
    branches: Dict[Any, str] = field(default_factory=dict)  # router: valor -> paso destino

MemoryStore

Bases: SqliteStateBackend

Alias retrocompatible a :class:SqliteStateBackend (SQLite en disco).

MemoryStore es SQLite en disco desde F5 — el nombre engaña. Para F15 se refactorizó para heredar de StateBackend de modo que cualquier store que hoy recibe un MemoryStore puede recibir también un PostgresStateBackend compartido sin cambios de API.

Se sigue construyendo con la misma firma: MemoryStore(db_path).

Source code in src/ciel/runtime/memory.py

class MemoryStore(SqliteStateBackend):
    """Alias retrocompatible a :class:`SqliteStateBackend` (SQLite en disco).

    ``MemoryStore`` es SQLite en disco desde F5 — el nombre engaña. Para
    F15 se refactorizó para heredar de ``StateBackend`` de modo que cualquier
    store que hoy recibe un ``MemoryStore`` puede recibir también un
    ``PostgresStateBackend`` compartido sin cambios de API.

    Se sigue construyendo con la misma firma: ``MemoryStore(db_path)``.
    """

    def __init__(self, db_path: str) -> None:
        super().__init__(db_path)

Supervisor

Source code in src/ciel/orchestration/supervisor.py

class Supervisor:
    def __init__(
        self,
        max_attempts: int = 2,
        timeout_s: float = 2.0,
        budget: Optional[Any] = None,
        rate_limiter: Optional[Any] = None,
        agent_counter: Optional[Any] = None,
        rate_limit: int = 0,
    ) -> None:
        self.max_attempts = max_attempts
        self.timeout_s = timeout_s
        self.budget = budget
        self.rate_limiter = rate_limiter
        self.agent_counter = agent_counter
        self.rate_limit = rate_limit
        self._results: Dict[str, WorkerResult] = {}

    async def run(
        self,
        step_id: str,
        worker: Worker,
        payload: Optional[Dict[str, Any]] = None,
        worker_id: str = "worker-1",
    ) -> WorkerResult:
        key = f"{step_id}:{worker_id}"
        if self.agent_counter is not None and self.budget is not None:
            exceeded = self.agent_counter.exceed(self.budget)
            if exceeded:
                result = WorkerResult(
                    worker_id=worker_id,
                    attempts=0,
                    failed=True,
                    error=f"budget rejection before run: {exceeded}",
                )
                self._results[key] = result
                return result
        if self.rate_limiter is not None and self.rate_limit > 0:
            rate_key = step_id or worker_id
            exceeded = self.rate_limiter.check(rate_key, self.rate_limit)
            if exceeded:
                result = WorkerResult(
                    worker_id=worker_id,
                    attempts=0,
                    failed=True,
                    error=f"rate limit rejection before run: {exceeded}",
                )
                self._results[key] = result
                return result
        attempt = 0
        start = time.perf_counter()
        while attempt < self.max_attempts:
            attempt += 1
            if self.agent_counter is not None:
                self.agent_counter.consume_tool(1)
            ctx = WorkerContext(step_id=step_id, worker_id=worker_id, payload=payload)
            try:
                output = await asyncio.wait_for(worker(ctx), timeout=self.timeout_s)
                result = WorkerResult(
                    worker_id=worker_id,
                    output=output,
                    attempts=attempt,
                    latency_ms=(time.perf_counter() - start) * 1000.0,
                )
                self._results[key] = result
                return result
            except Exception as exc:
                if attempt >= self.max_attempts:
                    result = WorkerResult(
                        worker_id=worker_id,
                        attempts=attempt,
                        latency_ms=(time.perf_counter() - start) * 1000.0,
                        failed=True,
                        error=str(exc),
                    )
                    self._results[key] = result
                    return result
        return self._results[key]

    def results(self) -> Dict[str, WorkerResult]:
        return dict(self._results)

WorkerContext dataclass

Source code in src/ciel/orchestration/supervisor.py

@dataclass
class WorkerContext:
    step_id: str
    worker_id: str
    payload: Optional[Dict[str, Any]] = None

AgentFn = Callable[['GroupChatState'], Any] module-attribute

Message = Dict[str, Any] module-attribute

Agent dataclass

Participante conversable del group chat.

responder recibe el GroupChatState y devuelve el texto de su réplica (o una corutina). Si terminate_if recibe el texto y devuelve True, ese mensaje finaliza la conversación.

Source code in src/ciel/orchestration/chat.py

@dataclass
class Agent:
    """Participante conversable del group chat.

    ``responder`` recibe el ``GroupChatState`` y devuelve el texto de su
    réplica (o una corutina). Si ``terminate_if`` recibe el texto y devuelve
    True, ese mensaje finaliza la conversación.
    """

    name: str
    responder: AgentFn
    system_message: str = ""
    terminate_if: Optional[Callable[[str], bool]] = None

    async def _reply(self, state: GroupChatState) -> str:
        res = self.responder(state)
        if hasattr(res, "__await__"):
            return await res
        return res

    def _is_terminator(self, text: str) -> bool:
        if self.terminate_if is None:
            return False
        return bool(self.terminate_if(text))

ChatMessage dataclass

Mensaje del transcripto (estilo AutoGen ChatMessage simplificado).

Source code in src/ciel/orchestration/chat.py

@dataclass
class ChatMessage:
    """Mensaje del transcripto (estilo AutoGen ChatMessage simplificado)."""

    role: str
    content: str
    round: int = 0

    def to_dict(self) -> Message:
        return {"role": self.role, "content": self.content, "round": self.round}

GroupChat

Configuración del group chat (participantes + estrategia de turnos).

Source code in src/ciel/orchestration/chat.py

class GroupChat:
    """Configuración del group chat (participantes + estrategia de turnos)."""

    def __init__(
        self,
        participants: List[Agent],
        *,
        max_rounds: int = 12,
        selector: Optional[Callable[["GroupChatState", List[Agent]], str]] = None,
        terminate_keyword: str = "TERMINATE",
    ) -> None:
        if not participants:
            raise GroupChatError("group chat requires at least one participant")
        self.participants = list(participants)
        self._by_name = {p.name: p for p in participants}
        self.max_rounds = max_rounds
        self.selector = selector
        self.terminate_keyword = terminate_keyword
        # round-robin a partir del índice 0
        self._next_index = 0

    def participant_names(self) -> List[str]:
        return [p.name for p in self.participants]

    def get(self, name: str) -> Agent:
        if name not in self._by_name:
            raise GroupChatError(f"unknown participant '{name}'")
        return self._by_name[name]

    def select_next(self, state: GroupChatState) -> Agent:
        if self.selector is not None:
            name = self.selector(state, self.participants)
            return self.get(name)
        # round-robin determinista (estilo AutoGen default)
        agent = self.participants[self._next_index % len(self.participants)]
        self._next_index += 1
        return agent

GroupChatCheckpointStore

Persistencia del transcripto de group chat sobre MemoryStore.

Source code in src/ciel/orchestration/chat.py

class GroupChatCheckpointStore:
    """Persistencia del transcripto de group chat sobre ``MemoryStore``."""

    def __init__(self, memory_store: MemoryStore) -> None:
        self.memory = memory_store

    def _key(self, run_id: str) -> str:
        return f"chat:{run_id}"

    def save(self, *, run_id: str, state: GroupChatState, tenant_id: Optional[str], session_id: Optional[str]) -> str:
        import uuid

        checkpoint_id = str(uuid.uuid4())
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id or run_id,
            key=self._key(run_id),
            value={"checkpoint_id": checkpoint_id, "run_id": run_id, "state": state.snapshot()},
        )
        return checkpoint_id

    def load(self, *, run_id: str, tenant_id: Optional[str], session_id: Optional[str]) -> Optional[Dict[str, Any]]:
        payload = self.memory.get(tenant_id=tenant_id, session_id=session_id or run_id, key=self._key(run_id))
        return payload if isinstance(payload, dict) else None

GroupChatError

Bases: Exception

Source code in src/ciel/orchestration/chat.py

class GroupChatError(Exception):
    pass

GroupChatManager

Ejecuta el diálogo: selecciona hablante, recolecta réplica, evalúa terminación. Montado SOBRE Supervisor para heredar retry/timeout/budget por participante (cada réplica es un worker del supervisor).

Source code in src/ciel/orchestration/chat.py

class GroupChatManager:
    """Ejecuta el diálogo: selecciona hablante, recolecta réplica, evalúa
    terminación. Montado SOBRE ``Supervisor`` para heredar retry/timeout/budget
    por participante (cada réplica es un worker del supervisor)."""

    def __init__(
        self,
        chat: GroupChat,
        *,
        supervisor: Optional[Supervisor] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> None:
        self.chat = chat
        self.supervisor = supervisor or Supervisor()
        self.tenant_id = tenant_id
        self.session_id = session_id

    async def _reply_via_supervisor(self, agent: Agent, state: GroupChatState, round_no: int) -> str:
        async def _worker(ctx: WorkerContext) -> str:
            return await agent._reply(state)

        result = await self.supervisor.run(
            step_id=f"chat:{agent.name}:{round_no}",
            worker=_worker,
            payload={"agent": agent.name, "round": round_no},
            worker_id=agent.name,
        )
        if result.failed:
            raise GroupChatError(f"agent '{agent.name}' failed in round {round_no}: {result.error}")
        return str(result.output)

    async def run(self, *, initial_message: Optional[str] = None, initial_sender: Optional[str] = None) -> GroupChatState:
        state = GroupChatState()
        if initial_message and initial_sender:
            state.transcript.append(ChatMessage(role=initial_sender, content=initial_message, round=0))

        for round_no in range(1, self.chat.max_rounds + 1):
            speaker = self.chat.select_next(state)
            text = await self._reply_via_supervisor(speaker, state, round_no)
            msg = ChatMessage(role=speaker.name, content=text, round=round_no)
            state.transcript.append(msg)
            state.rounds = round_no

            if self.chat.terminate_keyword and self.chat.terminate_keyword in text:
                state.terminated = True
                state.terminator = speaker.name
                break
            if speaker._is_terminator(text):
                state.terminated = True
                state.terminator = speaker.name
                break

        if state.rounds >= self.chat.max_rounds and not state.terminated:
            # Se agotaron las rondas sin señal explícita: marca terminado por límite.
            state.terminated = True
            state.terminator = None
        return state

GroupChatState dataclass

Estado compartido del group chat: transcripto + metadata por ronda.

Source code in src/ciel/orchestration/chat.py

@dataclass
class GroupChatState:
    """Estado compartido del group chat: transcripto + metadata por ronda."""

    transcript: List[ChatMessage] = field(default_factory=list)
    rounds: int = 0
    terminated: bool = False
    terminator: Optional[str] = None

    def last_message(self) -> Optional[ChatMessage]:
        return self.transcript[-1] if self.transcript else None

    def snapshot(self) -> Dict[str, Any]:
        return {
            "transcript": [m.to_dict() for m in self.transcript],
            "rounds": self.rounds,
            "terminated": self.terminated,
            "terminator": self.terminator,
        }

    @classmethod
    def from_snapshot(cls, snap: Dict[str, Any]) -> "GroupChatState":
        return cls(
            transcript=[ChatMessage(**m) for m in snap.get("transcript", [])],
            rounds=int(snap.get("rounds", 0)),
            terminated=bool(snap.get("terminated", False)),
            terminator=snap.get("terminator"),
        )

MemoryStore

Bases: SqliteStateBackend

Alias retrocompatible a :class:SqliteStateBackend (SQLite en disco).

MemoryStore es SQLite en disco desde F5 — el nombre engaña. Para F15 se refactorizó para heredar de StateBackend de modo que cualquier store que hoy recibe un MemoryStore puede recibir también un PostgresStateBackend compartido sin cambios de API.

Se sigue construyendo con la misma firma: MemoryStore(db_path).

Source code in src/ciel/runtime/memory.py

class MemoryStore(SqliteStateBackend):
    """Alias retrocompatible a :class:`SqliteStateBackend` (SQLite en disco).

    ``MemoryStore`` es SQLite en disco desde F5 — el nombre engaña. Para
    F15 se refactorizó para heredar de ``StateBackend`` de modo que cualquier
    store que hoy recibe un ``MemoryStore`` puede recibir también un
    ``PostgresStateBackend`` compartido sin cambios de API.

    Se sigue construyendo con la misma firma: ``MemoryStore(db_path)``.
    """

    def __init__(self, db_path: str) -> None:
        super().__init__(db_path)

Supervisor

Source code in src/ciel/orchestration/supervisor.py

class Supervisor:
    def __init__(
        self,
        max_attempts: int = 2,
        timeout_s: float = 2.0,
        budget: Optional[Any] = None,
        rate_limiter: Optional[Any] = None,
        agent_counter: Optional[Any] = None,
        rate_limit: int = 0,
    ) -> None:
        self.max_attempts = max_attempts
        self.timeout_s = timeout_s
        self.budget = budget
        self.rate_limiter = rate_limiter
        self.agent_counter = agent_counter
        self.rate_limit = rate_limit
        self._results: Dict[str, WorkerResult] = {}

    async def run(
        self,
        step_id: str,
        worker: Worker,
        payload: Optional[Dict[str, Any]] = None,
        worker_id: str = "worker-1",
    ) -> WorkerResult:
        key = f"{step_id}:{worker_id}"
        if self.agent_counter is not None and self.budget is not None:
            exceeded = self.agent_counter.exceed(self.budget)
            if exceeded:
                result = WorkerResult(
                    worker_id=worker_id,
                    attempts=0,
                    failed=True,
                    error=f"budget rejection before run: {exceeded}",
                )
                self._results[key] = result
                return result
        if self.rate_limiter is not None and self.rate_limit > 0:
            rate_key = step_id or worker_id
            exceeded = self.rate_limiter.check(rate_key, self.rate_limit)
            if exceeded:
                result = WorkerResult(
                    worker_id=worker_id,
                    attempts=0,
                    failed=True,
                    error=f"rate limit rejection before run: {exceeded}",
                )
                self._results[key] = result
                return result
        attempt = 0
        start = time.perf_counter()
        while attempt < self.max_attempts:
            attempt += 1
            if self.agent_counter is not None:
                self.agent_counter.consume_tool(1)
            ctx = WorkerContext(step_id=step_id, worker_id=worker_id, payload=payload)
            try:
                output = await asyncio.wait_for(worker(ctx), timeout=self.timeout_s)
                result = WorkerResult(
                    worker_id=worker_id,
                    output=output,
                    attempts=attempt,
                    latency_ms=(time.perf_counter() - start) * 1000.0,
                )
                self._results[key] = result
                return result
            except Exception as exc:
                if attempt >= self.max_attempts:
                    result = WorkerResult(
                        worker_id=worker_id,
                        attempts=attempt,
                        latency_ms=(time.perf_counter() - start) * 1000.0,
                        failed=True,
                        error=str(exc),
                    )
                    self._results[key] = result
                    return result
        return self._results[key]

    def results(self) -> Dict[str, WorkerResult]:
        return dict(self._results)

WorkerContext dataclass

Source code in src/ciel/orchestration/supervisor.py

@dataclass
class WorkerContext:
    step_id: str
    worker_id: str
    payload: Optional[Dict[str, Any]] = None

HandlerFn = Callable[['RootState'], Any] module-attribute

RoutingFn = Callable[[str], Any] module-attribute

MemoryStore

Bases: SqliteStateBackend

Alias retrocompatible a :class:SqliteStateBackend (SQLite en disco).

MemoryStore es SQLite en disco desde F5 — el nombre engaña. Para F15 se refactorizó para heredar de StateBackend de modo que cualquier store que hoy recibe un MemoryStore puede recibir también un PostgresStateBackend compartido sin cambios de API.

Se sigue construyendo con la misma firma: MemoryStore(db_path).

Source code in src/ciel/runtime/memory.py

class MemoryStore(SqliteStateBackend):
    """Alias retrocompatible a :class:`SqliteStateBackend` (SQLite en disco).

    ``MemoryStore`` es SQLite en disco desde F5 — el nombre engaña. Para
    F15 se refactorizó para heredar de ``StateBackend`` de modo que cualquier
    store que hoy recibe un ``MemoryStore`` puede recibir también un
    ``PostgresStateBackend`` compartido sin cambios de API.

    Se sigue construyendo con la misma firma: ``MemoryStore(db_path)``.
    """

    def __init__(self, db_path: str) -> None:
        super().__init__(db_path)

RootAgent

Builder declarativo: add_specialist + set_router; luego compile.

El router decide a qué specialist enrutar la petición. Si devuelve None, el root_handler (opcional) la atiende localmente.

Source code in src/ciel/orchestration/root.py

class RootAgent:
    """Builder declarativo: ``add_specialist`` + ``set_router``; luego ``compile``.

    El ``router`` decide a qué specialist enrutar la petición. Si devuelve
    ``None``, el ``root_handler`` (opcional) la atiende localmente.
    """

    def __init__(self, name: str = "root") -> None:
        self.name = name
        self._specialists: Dict[str, Specialist] = {}
        self._router: Optional[RoutingFn] = None
        self._root_handler: Optional[HandlerFn] = None

    def add_specialist(self, specialist: Specialist) -> "RootAgent":
        if specialist.name in self._specialists:
            raise RootAgentError(f"specialist '{specialist.name}' already registered")
        self._specialists[specialist.name] = specialist
        return self

    def set_router(self, router: RoutingFn) -> "RootAgent":
        self._router = router
        return self

    def set_root_handler(self, handler: HandlerFn) -> "RootAgent":
        self._root_handler = handler
        return self

    def specialists(self) -> Dict[str, Specialist]:
        return dict(self._specialists)

    def compile(
        self,
        *,
        supervisor: Optional[Supervisor] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> "RootRunner":
        if self._router is None and self._root_handler is None:
            raise RootAgentError("root agent needs a router and/or a root_handler")
        return RootRunner(
            agent=self,
            supervisor=supervisor or Supervisor(),
            tenant_id=tenant_id,
            session_id=session_id,
        )

RootAgentError

Bases: Exception

Source code in src/ciel/orchestration/root.py

class RootAgentError(Exception):
    pass

RootCheckpointStore

Persistencia de la decisión de enrutamiento sobre MemoryStore.

Source code in src/ciel/orchestration/root.py

class RootCheckpointStore:
    """Persistencia de la decisión de enrutamiento sobre ``MemoryStore``."""

    def __init__(self, memory_store: MemoryStore) -> None:
        self.memory = memory_store

    def _key(self, run_id: str) -> str:
        return f"root:{run_id}"

    def save(self, *, run_id: str, state: RootState, tenant_id: Optional[str], session_id: Optional[str]) -> str:
        import uuid

        checkpoint_id = str(uuid.uuid4())
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id or run_id,
            key=self._key(run_id),
            value={"checkpoint_id": checkpoint_id, "run_id": run_id, "state": state.snapshot()},
        )
        return checkpoint_id

    def load(self, *, run_id: str, tenant_id: Optional[str], session_id: Optional[str]) -> Optional[Dict[str, Any]]:
        payload = self.memory.get(tenant_id=tenant_id, session_id=session_id or run_id, key=self._key(run_id))
        return payload if isinstance(payload, dict) else None

RootRunner

Ejecuta el root agent: enruta la petición y ejecuta el specialist (o root).

Source code in src/ciel/orchestration/root.py

class RootRunner:
    """Ejecuta el root agent: enruta la petición y ejecuta el specialist (o root)."""

    def __init__(
        self,
        *,
        agent: RootAgent,
        supervisor: Supervisor,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
        session_store: Optional[SessionStore] = None,
    ) -> None:
        self.agent = agent
        self.supervisor = supervisor
        self.tenant_id = tenant_id
        self.session_id = session_id
        self.session_store = session_store or (SessionStore(supervisor._memory) if hasattr(supervisor, "_memory") else None)

    async def _run_worker(self, worker_id: str, fn: Callable[["RootState"], Any], state: RootState) -> Any:
        async def _worker(ctx: WorkerContext) -> Any:
            return await self._invoke(fn, state)

        result = await self.supervisor.run(
            step_id=f"root:{worker_id}",
            worker=_worker,
            payload={"worker": worker_id},
            worker_id=worker_id,
        )
        if result.failed:
            raise RootAgentError(f"worker '{worker_id}' failed after {result.attempts} attempts: {result.error}")
        return result.output

    @staticmethod
    async def _invoke(fn: Callable[["RootState"], Any], state: RootState) -> Any:
        res = fn(state)
        if hasattr(res, "__await__"):
            return await res
        return res

    async def route(
        self,
        prompt: str,
        *,
        session_id: Optional[str] = None,
        session_store: Optional[SessionStore] = None,
        tenant_id: Optional[str] = None,
    ) -> RootState:
        """Enruta y ejecuta la petición, manteniendo session state por tenant.

        Si se pasa ``session_id`` (+ ``session_store``/``tenant_id``), el estado
        del agente se reconstruye con el historial de turnos previos y el turno
        resultante se persiste para la próxima invocación (estilo ADK session
        state entre turnos). OFFLINE-SAFE: no requiere red ni proveedor.
        """
        sid = session_id if session_id is not None else self.session_id
        store = session_store if session_store is not None else self.session_store
        tid = tenant_id if tenant_id is not None else self.tenant_id

        # Rehidrata el historial de turnos previos de la session (si existe).
        prior_history: List[Dict[str, Any]] = []
        if store is not None and sid is not None:
            prior_history = store.history(tenant_id=tid, session_id=sid)

        state = RootState(prompt=prompt, history=list(prior_history))
        router = self.agent._router
        if router is not None:
            target = router(prompt)
            if isinstance(target, Awaitable):
                target = await target
            if target is not None:
                if target not in self.agent._specialists:
                    raise RootAgentError(f"router returned unknown specialist '{target}'")
                state.route = target
                state.result = await self._run_worker(target, self.agent._specialists[target]._handle, state)
                state.metadata["handled_by"] = target
                self._persist_turn(store, tid, sid, state)
                return state
        # Sin enrutamiento -> lo maneja el root (si lo hay).
        if self.agent._root_handler is not None:
            state.handled_by_root = True
            state.result = await self._run_worker("root", self.agent._root_handler, state)
            state.metadata["handled_by"] = self.agent.name
            self._persist_turn(store, tid, sid, state)
            return state
        raise RootAgentError("router returned None and no root_handler is configured")

    @staticmethod
    def _persist_turn(
        store: Optional[SessionStore],
        tid: Optional[str],
        sid: Optional[str],
        state: RootState,
    ) -> None:
        """Persiste el turno en la session si hay session store + id."""
        if store is None or sid is None:
            return
        turn = {
            "prompt": state.prompt,
            "route": state.route,
            "result": state.result,
            "handled_by_root": state.handled_by_root,
            "handled_by": state.metadata.get("handled_by"),
        }
        store.append_turn(tenant_id=tid, session_id=sid, turn=turn)

route(prompt: str, *, session_id: Optional[str] = None, session_store: Optional[SessionStore] = None, tenant_id: Optional[str] = None) -> RootState async

Enruta y ejecuta la petición, manteniendo session state por tenant.

Si se pasa session_id (+ session_store/tenant_id), el estado del agente se reconstruye con el historial de turnos previos y el turno resultante se persiste para la próxima invocación (estilo ADK session state entre turnos). OFFLINE-SAFE: no requiere red ni proveedor.

Source code in src/ciel/orchestration/root.py

async def route(
    self,
    prompt: str,
    *,
    session_id: Optional[str] = None,
    session_store: Optional[SessionStore] = None,
    tenant_id: Optional[str] = None,
) -> RootState:
    """Enruta y ejecuta la petición, manteniendo session state por tenant.

    Si se pasa ``session_id`` (+ ``session_store``/``tenant_id``), el estado
    del agente se reconstruye con el historial de turnos previos y el turno
    resultante se persiste para la próxima invocación (estilo ADK session
    state entre turnos). OFFLINE-SAFE: no requiere red ni proveedor.
    """
    sid = session_id if session_id is not None else self.session_id
    store = session_store if session_store is not None else self.session_store
    tid = tenant_id if tenant_id is not None else self.tenant_id

    # Rehidrata el historial de turnos previos de la session (si existe).
    prior_history: List[Dict[str, Any]] = []
    if store is not None and sid is not None:
        prior_history = store.history(tenant_id=tid, session_id=sid)

    state = RootState(prompt=prompt, history=list(prior_history))
    router = self.agent._router
    if router is not None:
        target = router(prompt)
        if isinstance(target, Awaitable):
            target = await target
        if target is not None:
            if target not in self.agent._specialists:
                raise RootAgentError(f"router returned unknown specialist '{target}'")
            state.route = target
            state.result = await self._run_worker(target, self.agent._specialists[target]._handle, state)
            state.metadata["handled_by"] = target
            self._persist_turn(store, tid, sid, state)
            return state
    # Sin enrutamiento -> lo maneja el root (si lo hay).
    if self.agent._root_handler is not None:
        state.handled_by_root = True
        state.result = await self._run_worker("root", self.agent._root_handler, state)
        state.metadata["handled_by"] = self.agent.name
        self._persist_turn(store, tid, sid, state)
        return state
    raise RootAgentError("router returned None and no root_handler is configured")

RootState dataclass

Estado compartido del root agent: prompt, ruta elegida y resultado.

history acumula los turnos previos de la session (estilo ADK session state) para que el agente recuerde contexto entre invocaciones.

Source code in src/ciel/orchestration/root.py

@dataclass
class RootState:
    """Estado compartido del root agent: prompt, ruta elegida y resultado.

    ``history`` acumula los turnos previos de la session (estilo ADK session
    state) para que el agente recuerde contexto entre invocaciones.
    """

    prompt: str = ""
    route: Optional[str] = None
    result: Any = None
    handled_by_root: bool = False
    metadata: Dict[str, Any] = field(default_factory=dict)
    history: List[Dict[str, Any]] = field(default_factory=list)

    def snapshot(self) -> Dict[str, Any]:
        return {
            "prompt": self.prompt,
            "route": self.route,
            "result": self.result,
            "handled_by_root": self.handled_by_root,
            "metadata": dict(self.metadata),
            "history": [dict(h) for h in self.history],
        }

    @classmethod
    def from_snapshot(cls, snap: Dict[str, Any]) -> "RootState":
        return cls(
            prompt=snap.get("prompt", ""),
            route=snap.get("route"),
            result=snap.get("result"),
            handled_by_root=bool(snap.get("handled_by_root", False)),
            metadata=dict(snap.get("metadata", {})),
            history=[dict(h) for h in snap.get("history", [])],
        )

SessionStore

Estado de session durable por tenant, sobre MemoryStore.

Claves internas (todas namespaced con session: para no colisionar con otros módulos que usan el mismo MemoryStore):

• session:<sid>:turns -> {"version", "turns": [turn, ...]}
• session:<sid>:state -> {key: value} (estado arbitrario)
• session:<sid>:board_links-> {"links": [board_task_id, ...]}
• session_index:<tenant> -> {"sessions": [sid, ...]} (índice)

Source code in src/ciel/orchestration/session.py

class SessionStore:
    """Estado de session durable por tenant, sobre ``MemoryStore``.

    Claves internas (todas namespaced con ``session:`` para no colisionar con
    otros módulos que usan el mismo ``MemoryStore``):

    * ``session:<sid>:turns``      -> {"version", "turns": [turn, ...]}
    * ``session:<sid>:state``      -> {key: value}  (estado arbitrario)
    * ``session:<sid>:board_links``-> {"links": [board_task_id, ...]}
    * ``session_index:<tenant>``   -> {"sessions": [sid, ...]}  (índice)
    """

    def __init__(self, memory_store: MemoryStore) -> None:
        self.memory = memory_store

    # -- keys ---------------------------------------------------------------
    @staticmethod
    def _turns_key(session_id: str) -> str:
        return f"session:{session_id}:turns"

    @staticmethod
    def _state_key(session_id: str) -> str:
        return f"session:{session_id}:state"

    @staticmethod
    def _links_key(session_id: str) -> str:
        return f"session:{session_id}:board_links"

    @staticmethod
    def _index_key(tenant_id: Optional[str]) -> str:
        sentinel = tenant_id if tenant_id is not None else "__none__"
        return f"session_index:{sentinel}"

    # -- index --------------------------------------------------------------
    def _load_index(self, tenant_id: Optional[str]) -> List[str]:
        # El índice se guarda con session_id == tenant_id (determinista) para
        # no depender de una session concreta.
        sid = tenant_id if tenant_id is not None else "__none__"
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=sid, key=self._index_key(tenant_id)
        )
        if isinstance(payload, dict):
            return list(payload.get("sessions", []))
        return []

    def _register(self, *, tenant_id: Optional[str], session_id: str) -> None:
        sessions = self._load_index(tenant_id)
        if session_id not in sessions:
            sessions.append(session_id)
            sid = tenant_id if tenant_id is not None else "__none__"
            self.memory.set(
                tenant_id=tenant_id,
                session_id=sid,
                key=self._index_key(tenant_id),
                value={"sessions": sessions},
            )

    # -- turns --------------------------------------------------------------
    def append_turn(
        self, *, tenant_id: Optional[str], session_id: str, turn: Dict[str, Any]
    ) -> None:
        """Añade un turno al historial de la session y lo registra en el índice."""
        turn = dict(turn)
        turn.setdefault("ts", time.time())
        turns = self.history(tenant_id=tenant_id, session_id=session_id)
        turns.append(turn)
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id,
            key=self._turns_key(session_id),
            value={"version": SESSION_VERSION, "turns": turns},
        )
        self._register(tenant_id=tenant_id, session_id=session_id)

    def history(self, *, tenant_id: Optional[str], session_id: str) -> List[Dict[str, Any]]:
        """Recupera el historial de turnos (vacío si la session no existe)."""
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=session_id, key=self._turns_key(session_id)
        )
        if isinstance(payload, dict):
            return [dict(t) for t in payload.get("turns", [])]
        return []

    # -- arbitrary state ----------------------------------------------------
    def _load_state_blob(self, tenant_id: Optional[str], session_id: str) -> Dict[str, Any]:
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=session_id, key=self._state_key(session_id)
        )
        return dict(payload) if isinstance(payload, dict) else {}

    def save_state(
        self, *, tenant_id: Optional[str], session_id: str, key: str, value: Any
    ) -> None:
        state = self._load_state_blob(tenant_id, session_id)
        state[key] = value
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id,
            key=self._state_key(session_id),
            value=state,
        )
        self._register(tenant_id=tenant_id, session_id=session_id)

    def load_state(
        self, *, tenant_id: Optional[str], session_id: str, key: str, default: Any = None
    ) -> Any:
        state = self._load_state_blob(tenant_id, session_id)
        return state.get(key, default)

    # -- board integration --------------------------------------------------
    def link_board_task(
        self, *, tenant_id: Optional[str], session_id: str, board_task_id: str
    ) -> None:
        """Vincula una tarea del board (mismo tenant) con la session."""
        links = self.board_links(tenant_id=tenant_id, session_id=session_id)
        if board_task_id not in links:
            links.append(board_task_id)
            self.memory.set(
                tenant_id=tenant_id,
                session_id=session_id,
                key=self._links_key(session_id),
                value={"links": links},
            )

    def board_links(self, *, tenant_id: Optional[str], session_id: str) -> List[str]:
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=session_id, key=self._links_key(session_id)
        )
        if isinstance(payload, dict):
            return list(payload.get("links", []))
        return []

    # -- listing ------------------------------------------------------------
    def list_sessions(self, *, tenant_id: Optional[str]) -> List[str]:
        """Lista los ids de session registrados para un tenant."""
        return self._load_index(tenant_id)

append_turn(*, tenant_id: Optional[str], session_id: str, turn: Dict[str, Any]) -> None

Añade un turno al historial de la session y lo registra en el índice.

Source code in src/ciel/orchestration/session.py

def append_turn(
    self, *, tenant_id: Optional[str], session_id: str, turn: Dict[str, Any]
) -> None:
    """Añade un turno al historial de la session y lo registra en el índice."""
    turn = dict(turn)
    turn.setdefault("ts", time.time())
    turns = self.history(tenant_id=tenant_id, session_id=session_id)
    turns.append(turn)
    self.memory.set(
        tenant_id=tenant_id,
        session_id=session_id,
        key=self._turns_key(session_id),
        value={"version": SESSION_VERSION, "turns": turns},
    )
    self._register(tenant_id=tenant_id, session_id=session_id)

history(*, tenant_id: Optional[str], session_id: str) -> List[Dict[str, Any]]

Recupera el historial de turnos (vacío si la session no existe).

Source code in src/ciel/orchestration/session.py

def history(self, *, tenant_id: Optional[str], session_id: str) -> List[Dict[str, Any]]:
    """Recupera el historial de turnos (vacío si la session no existe)."""
    payload = self.memory.get(
        tenant_id=tenant_id, session_id=session_id, key=self._turns_key(session_id)
    )
    if isinstance(payload, dict):
        return [dict(t) for t in payload.get("turns", [])]
    return []

link_board_task(*, tenant_id: Optional[str], session_id: str, board_task_id: str) -> None

Vincula una tarea del board (mismo tenant) con la session.

Source code in src/ciel/orchestration/session.py

def link_board_task(
    self, *, tenant_id: Optional[str], session_id: str, board_task_id: str
) -> None:
    """Vincula una tarea del board (mismo tenant) con la session."""
    links = self.board_links(tenant_id=tenant_id, session_id=session_id)
    if board_task_id not in links:
        links.append(board_task_id)
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id,
            key=self._links_key(session_id),
            value={"links": links},
        )

list_sessions(*, tenant_id: Optional[str]) -> List[str]

Lista los ids de session registrados para un tenant.

Source code in src/ciel/orchestration/session.py

def list_sessions(self, *, tenant_id: Optional[str]) -> List[str]:
    """Lista los ids de session registrados para un tenant."""
    return self._load_index(tenant_id)

Specialist dataclass

Agente especialista enrutado por el root agent (ADK sub_agent).

Source code in src/ciel/orchestration/root.py

@dataclass
class Specialist:
    """Agente especialista enrutado por el root agent (ADK sub_agent)."""

    name: str
    handler: HandlerFn
    description: str = ""

    async def _handle(self, state: RootState) -> Any:
        res = self.handler(state)
        if hasattr(res, "__await__"):
            return await res
        return res

Supervisor

Source code in src/ciel/orchestration/supervisor.py

class Supervisor:
    def __init__(
        self,
        max_attempts: int = 2,
        timeout_s: float = 2.0,
        budget: Optional[Any] = None,
        rate_limiter: Optional[Any] = None,
        agent_counter: Optional[Any] = None,
        rate_limit: int = 0,
    ) -> None:
        self.max_attempts = max_attempts
        self.timeout_s = timeout_s
        self.budget = budget
        self.rate_limiter = rate_limiter
        self.agent_counter = agent_counter
        self.rate_limit = rate_limit
        self._results: Dict[str, WorkerResult] = {}

    async def run(
        self,
        step_id: str,
        worker: Worker,
        payload: Optional[Dict[str, Any]] = None,
        worker_id: str = "worker-1",
    ) -> WorkerResult:
        key = f"{step_id}:{worker_id}"
        if self.agent_counter is not None and self.budget is not None:
            exceeded = self.agent_counter.exceed(self.budget)
            if exceeded:
                result = WorkerResult(
                    worker_id=worker_id,
                    attempts=0,
                    failed=True,
                    error=f"budget rejection before run: {exceeded}",
                )
                self._results[key] = result
                return result
        if self.rate_limiter is not None and self.rate_limit > 0:
            rate_key = step_id or worker_id
            exceeded = self.rate_limiter.check(rate_key, self.rate_limit)
            if exceeded:
                result = WorkerResult(
                    worker_id=worker_id,
                    attempts=0,
                    failed=True,
                    error=f"rate limit rejection before run: {exceeded}",
                )
                self._results[key] = result
                return result
        attempt = 0
        start = time.perf_counter()
        while attempt < self.max_attempts:
            attempt += 1
            if self.agent_counter is not None:
                self.agent_counter.consume_tool(1)
            ctx = WorkerContext(step_id=step_id, worker_id=worker_id, payload=payload)
            try:
                output = await asyncio.wait_for(worker(ctx), timeout=self.timeout_s)
                result = WorkerResult(
                    worker_id=worker_id,
                    output=output,
                    attempts=attempt,
                    latency_ms=(time.perf_counter() - start) * 1000.0,
                )
                self._results[key] = result
                return result
            except Exception as exc:
                if attempt >= self.max_attempts:
                    result = WorkerResult(
                        worker_id=worker_id,
                        attempts=attempt,
                        latency_ms=(time.perf_counter() - start) * 1000.0,
                        failed=True,
                        error=str(exc),
                    )
                    self._results[key] = result
                    return result
        return self._results[key]

    def results(self) -> Dict[str, WorkerResult]:
        return dict(self._results)

WorkerContext dataclass

Source code in src/ciel/orchestration/supervisor.py

@dataclass
class WorkerContext:
    step_id: str
    worker_id: str
    payload: Optional[Dict[str, Any]] = None

TaskHandler = Callable[['Task'], Any] module-attribute

__all__ = ['AgentError', 'EventLoopError', 'TaskError', 'Task', 'EventLoopCheckpointStore', 'EventLoop', 'EventLoopStep', 'AutonomousAgent'] module-attribute

AgentError

Bases: Exception

Error base de la agencia autónoma (Fase 6).

Source code in src/ciel/orchestration/agent.py

class AgentError(Exception):
    """Error base de la agencia autónoma (Fase 6)."""

AutonomousAgent

Agente autónomo que descompone un objetivo en tareas y las ejecuta.

OFFLINE-SAFE: por defecto el planner es un handler local que parte el objetivo en una sola tarea (o N tareas si se provee plan explícito); cada tarea se ejecuta con un handler local sobre task.payload (no red).

Source code in src/ciel/orchestration/agent.py

class AutonomousAgent:
    """Agente autónomo que descompone un objetivo en tareas y las ejecuta.

    OFFLINE-SAFE: por defecto el planner es un handler local que parte el
    objetivo en una sola tarea (o N tareas si se provee ``plan`` explícito);
    cada tarea se ejecuta con un handler local sobre ``task.payload`` (no red).
    """

    def __init__(
        self,
        name: str = "autonomous",
        *,
        supervisor: Optional[Supervisor] = None,
        checkpointer: Optional[EventLoopCheckpointStore] = None,
        session_store: Optional[SessionStore] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
        max_attempts: int = 5,
    ) -> None:
        self.name = name
        self.supervisor = supervisor or Supervisor()
        self.checkpointer = checkpointer
        self.session_store = session_store
        self.tenant_id = tenant_id
        self.session_id = session_id
        self.max_attempts = max_attempts

    def _loop(self, task: Task, handler: TaskHandler) -> EventLoop:
        return EventLoop(
            supervisor=self.supervisor,
            checkpointer=self.checkpointer,
            tenant_id=self.tenant_id,
            session_id=self.session_id,
            max_attempts=self.max_attempts,
        )

    async def run_task(self, task: Task, handler: TaskHandler) -> Task:
        loop = self._loop(task, handler)
        return await loop.run(task, handler)

    async def run_goal(
        self,
        goal: str,
        handler: TaskHandler,
        *,
        plan: Optional[Sequence[str]] = None,
    ) -> List[Task]:
        """Descompone ``goal`` en tareas (``plan`` o una sola) y las ejecuta.

        Cada tarea completada persiste un turno en ``SessionStore`` si está
        disponible (ADK session_state entre ejecuciones del agente).
        """
        steps = list(plan) if plan else [goal]
        tasks: List[Task] = [Task(goal=step) for step in steps]
        for task in tasks:
            await self.run_task(task, handler)
            self._persist_turn(task)
        return tasks

    def _persist_turn(self, task: Task) -> None:
        if self.session_store is None or self.session_id is None:
            return
        self.session_store.append_turn(
            tenant_id=self.tenant_id,
            session_id=self.session_id,
            turn={
                "goal": task.goal,
                "status": task.status,
                "result": task.result,
                "attempts": task.attempts,
            },
        )

run_goal(goal: str, handler: TaskHandler, *, plan: Optional[Sequence[str]] = None) -> List[Task] async

Descompone goal en tareas (plan o una sola) y las ejecuta.

Cada tarea completada persiste un turno en SessionStore si está disponible (ADK session_state entre ejecuciones del agente).

Source code in src/ciel/orchestration/agent.py

async def run_goal(
    self,
    goal: str,
    handler: TaskHandler,
    *,
    plan: Optional[Sequence[str]] = None,
) -> List[Task]:
    """Descompone ``goal`` en tareas (``plan`` o una sola) y las ejecuta.

    Cada tarea completada persiste un turno en ``SessionStore`` si está
    disponible (ADK session_state entre ejecuciones del agente).
    """
    steps = list(plan) if plan else [goal]
    tasks: List[Task] = [Task(goal=step) for step in steps]
    for task in tasks:
        await self.run_task(task, handler)
        self._persist_turn(task)
    return tasks

EventLoop

Bucle de eventos durable que ejecuta una Task con reintentos.

Montado sobre Supervisor (retry/timeout/budget por worker). Si el handler falla, el loop aplica backoff exponencial (capped) y reintenta hasta max_attempts. Tras cada intento persiste un checkpoint en MemoryStore; resume rehidrata y continúa.

OFFLINE-SAFE: el handler por defecto (y los demos) son funciones locales.

Source code in src/ciel/orchestration/agent.py

class EventLoop:
    """Bucle de eventos durable que ejecuta una ``Task`` con reintentos.

    Montado sobre ``Supervisor`` (retry/timeout/budget por worker). Si el
    handler falla, el loop aplica backoff exponencial (capped) y reintenta
    hasta ``max_attempts``. Tras cada intento persiste un checkpoint en
    ``MemoryStore``; ``resume`` rehidrata y continúa.

    OFFLINE-SAFE: el handler por defecto (y los demos) son funciones locales.
    """

    def __init__(
        self,
        *,
        supervisor: Optional[Supervisor] = None,
        checkpointer: Optional[EventLoopCheckpointStore] = None,
        tenant_id: Optional[str] = None,
        session_id: Optional[str] = None,
        max_attempts: int = 5,
        base_delay_s: float = 0.05,
        max_delay_s: float = 2.0,
        jitter: bool = False,
    ) -> None:
        self.supervisor = supervisor or Supervisor()
        self.checkpointer = checkpointer
        self.tenant_id = tenant_id
        self.session_id = session_id
        self.max_attempts = max_attempts
        self.base_delay_s = base_delay_s
        self.max_delay_s = max_delay_s
        self.jitter = jitter
        self.run_id: Optional[str] = None
        self.task: Optional[Task] = None
        self.steps: List[EventLoopStep] = []

    # -- helpers -------------------------------------------------------------
    @staticmethod
    async def _invoke(fn: TaskHandler, task: Task) -> Any:
        res = fn(task)
        if hasattr(res, "__await__"):
            return await res
        return res

    def _backoff(self, attempt: int) -> float:
        delay = min(self.base_delay_s * (2 ** (attempt - 1)), self.max_delay_s)
        if self.jitter:
            delay += (time.perf_counter() % 1) * 0.01
        return delay

    def _save(self, loop_state: Dict[str, Any]) -> None:
        if self.checkpointer is not None and self.run_id is not None and self.task is not None:
            self.checkpointer.save(
                run_id=self.run_id,
                loop_state=loop_state,
                task=self.task,
                tenant_id=self.tenant_id,
                session_id=self.session_id,
            )

    # -- run -----------------------------------------------------------------
    async def run(
        self,
        task: Task,
        handler: TaskHandler,
        *,
        run_id: Optional[str] = None,
    ) -> Task:
        """Ejecuta ``task`` con reintentos exponenciales hasta completar o agotar."""
        self.run_id = run_id or str(uuid.uuid4())
        self.task = task
        self.steps = []

        attempt = 0
        last_error: Optional[str] = None
        while attempt < self.max_attempts:
            attempt += 1
            task.attempts = attempt
            task.mark_running()
            start = time.perf_counter()
            try:
                result = await self.supervisor.run(
                    step_id=f"{self.run_id}:attempt:{attempt}",
                    worker=self._make_worker(handler, task),
                    payload={"attempt": attempt},
                    worker_id=f"loop-{attempt}",
                )
                latency = (time.perf_counter() - start) * 1000.0
                if result.failed:
                    # El Supervisor ya agotó sus reintentos internos.
                    last_error = result.error or "supervisor worker failed"
                    self.steps.append(
                        EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
                    )
                else:
                    task.mark_succeeded(result.output)
                    self.steps.append(
                        EventLoopStep(attempt=attempt, succeeded=True, output=result.output, latency_ms=latency)
                    )
                    self._save({"attempt": attempt, "status": task.status})
                    return task
            except Exception as exc:  # handler levantó fuera del supervisor
                last_error = str(exc)
                latency = (time.perf_counter() - start) * 1000.0
                self.steps.append(
                    EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
                )

            self._save({"attempt": attempt, "status": task.status})
            if attempt < self.max_attempts:
                await asyncio.sleep(self._backoff(attempt))

        task.mark_failed(last_error or "unknown failure")
        self._save({"attempt": attempt, "status": task.status})
        raise TaskError(
            f"task '{task.goal}' failed after {attempt} attempts: {task.error}"
        )

    def _make_worker(self, handler: TaskHandler, task: Task):
        async def _worker(ctx: WorkerContext) -> Any:
            return await self._invoke(handler, task)

        return _worker

    # -- resume --------------------------------------------------------------
    async def resume(self, *, run_id: str, handler: TaskHandler) -> Task:
        """Reanuda un loop interrumpido desde su último checkpoint.

        Rehidrata la ``Task`` persistida y continúa los reintentos. Si la tarea
        ya estaba en ``succeeded``/``failed`` al interrumpirse, la devuelve tal
        cual (idempotente). Cumple el criterio Fase 6: tras reinicio continúa y
        completa.
        """
        if self.checkpointer is None:
            raise EventLoopError("resume requires a checkpointer")
        self.run_id = run_id
        payload = self.checkpointer.load(
            run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
        )
        if payload is None:
            raise EventLoopError(f"no checkpoint found for run_id '{run_id}'")
        task = Task.from_snapshot(payload["task"])
        self.task = task
        loop_state = payload.get("loop_state", {})
        self.steps = []

        # Si ya terminó en el checkpoint previo, no re-ejecutar.
        if task.status in ("succeeded", "failed"):
            return task

        attempt = int(loop_state.get("attempt", 0))
        last_error: Optional[str] = None
        while attempt < self.max_attempts:
            attempt += 1
            task.attempts = attempt
            task.mark_running()
            start = time.perf_counter()
            try:
                result = await self.supervisor.run(
                    step_id=f"{self.run_id}:attempt:{attempt}",
                    worker=self._make_worker(handler, task),
                    payload={"attempt": attempt},
                    worker_id=f"loop-{attempt}",
                )
                latency = (time.perf_counter() - start) * 1000.0
                if result.failed:
                    last_error = result.error or "supervisor worker failed"
                    self.steps.append(
                        EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
                    )
                else:
                    task.mark_succeeded(result.output)
                    self.steps.append(
                        EventLoopStep(attempt=attempt, succeeded=True, output=result.output, latency_ms=latency)
                    )
                    self._save({"attempt": attempt, "status": task.status})
                    return task
            except Exception as exc:
                last_error = str(exc)
                latency = (time.perf_counter() - start) * 1000.0
                self.steps.append(
                    EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
                )

            self._save({"attempt": attempt, "status": task.status})
            if attempt < self.max_attempts:
                await asyncio.sleep(self._backoff(attempt))

        task.mark_failed(last_error or "unknown failure")
        self._save({"attempt": attempt, "status": task.status})
        raise TaskError(
            f"task '{task.goal}' failed after resume ({attempt} attempts): {task.error}"
        )

resume(*, run_id: str, handler: TaskHandler) -> Task async

Reanuda un loop interrumpido desde su último checkpoint.

Rehidrata la Task persistida y continúa los reintentos. Si la tarea ya estaba en succeeded/failed al interrumpirse, la devuelve tal cual (idempotente). Cumple el criterio Fase 6: tras reinicio continúa y completa.

Source code in src/ciel/orchestration/agent.py

async def resume(self, *, run_id: str, handler: TaskHandler) -> Task:
    """Reanuda un loop interrumpido desde su último checkpoint.

    Rehidrata la ``Task`` persistida y continúa los reintentos. Si la tarea
    ya estaba en ``succeeded``/``failed`` al interrumpirse, la devuelve tal
    cual (idempotente). Cumple el criterio Fase 6: tras reinicio continúa y
    completa.
    """
    if self.checkpointer is None:
        raise EventLoopError("resume requires a checkpointer")
    self.run_id = run_id
    payload = self.checkpointer.load(
        run_id=run_id, tenant_id=self.tenant_id, session_id=self.session_id
    )
    if payload is None:
        raise EventLoopError(f"no checkpoint found for run_id '{run_id}'")
    task = Task.from_snapshot(payload["task"])
    self.task = task
    loop_state = payload.get("loop_state", {})
    self.steps = []

    # Si ya terminó en el checkpoint previo, no re-ejecutar.
    if task.status in ("succeeded", "failed"):
        return task

    attempt = int(loop_state.get("attempt", 0))
    last_error: Optional[str] = None
    while attempt < self.max_attempts:
        attempt += 1
        task.attempts = attempt
        task.mark_running()
        start = time.perf_counter()
        try:
            result = await self.supervisor.run(
                step_id=f"{self.run_id}:attempt:{attempt}",
                worker=self._make_worker(handler, task),
                payload={"attempt": attempt},
                worker_id=f"loop-{attempt}",
            )
            latency = (time.perf_counter() - start) * 1000.0
            if result.failed:
                last_error = result.error or "supervisor worker failed"
                self.steps.append(
                    EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
                )
            else:
                task.mark_succeeded(result.output)
                self.steps.append(
                    EventLoopStep(attempt=attempt, succeeded=True, output=result.output, latency_ms=latency)
                )
                self._save({"attempt": attempt, "status": task.status})
                return task
        except Exception as exc:
            last_error = str(exc)
            latency = (time.perf_counter() - start) * 1000.0
            self.steps.append(
                EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
            )

        self._save({"attempt": attempt, "status": task.status})
        if attempt < self.max_attempts:
            await asyncio.sleep(self._backoff(attempt))

    task.mark_failed(last_error or "unknown failure")
    self._save({"attempt": attempt, "status": task.status})
    raise TaskError(
        f"task '{task.goal}' failed after resume ({attempt} attempts): {task.error}"
    )

run(task: Task, handler: TaskHandler, *, run_id: Optional[str] = None) -> Task async

Ejecuta task con reintentos exponenciales hasta completar o agotar.

Source code in src/ciel/orchestration/agent.py

async def run(
    self,
    task: Task,
    handler: TaskHandler,
    *,
    run_id: Optional[str] = None,
) -> Task:
    """Ejecuta ``task`` con reintentos exponenciales hasta completar o agotar."""
    self.run_id = run_id or str(uuid.uuid4())
    self.task = task
    self.steps = []

    attempt = 0
    last_error: Optional[str] = None
    while attempt < self.max_attempts:
        attempt += 1
        task.attempts = attempt
        task.mark_running()
        start = time.perf_counter()
        try:
            result = await self.supervisor.run(
                step_id=f"{self.run_id}:attempt:{attempt}",
                worker=self._make_worker(handler, task),
                payload={"attempt": attempt},
                worker_id=f"loop-{attempt}",
            )
            latency = (time.perf_counter() - start) * 1000.0
            if result.failed:
                # El Supervisor ya agotó sus reintentos internos.
                last_error = result.error or "supervisor worker failed"
                self.steps.append(
                    EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
                )
            else:
                task.mark_succeeded(result.output)
                self.steps.append(
                    EventLoopStep(attempt=attempt, succeeded=True, output=result.output, latency_ms=latency)
                )
                self._save({"attempt": attempt, "status": task.status})
                return task
        except Exception as exc:  # handler levantó fuera del supervisor
            last_error = str(exc)
            latency = (time.perf_counter() - start) * 1000.0
            self.steps.append(
                EventLoopStep(attempt=attempt, succeeded=False, error=last_error, latency_ms=latency)
            )

        self._save({"attempt": attempt, "status": task.status})
        if attempt < self.max_attempts:
            await asyncio.sleep(self._backoff(attempt))

    task.mark_failed(last_error or "unknown failure")
    self._save({"attempt": attempt, "status": task.status})
    raise TaskError(
        f"task '{task.goal}' failed after {attempt} attempts: {task.error}"
    )

EventLoopCheckpointStore

Persiste el estado del EventLoop sobre MemoryStore.

Clave: loop:<run_id> (namespaced para no colisionar con otros módulos). Guarda el Task en curso + el estado del loop (intent, estado global).

Source code in src/ciel/orchestration/agent.py

class EventLoopCheckpointStore:
    """Persiste el estado del EventLoop sobre ``MemoryStore``.

    Clave: ``loop:<run_id>`` (namespaced para no colisionar con otros módulos).
    Guarda el ``Task`` en curso + el estado del loop (intent, estado global).
    """

    def __init__(self, memory_store: MemoryStore) -> None:
        self.memory = memory_store

    def _key(self, run_id: str) -> str:
        return f"loop:{run_id}"

    def save(
        self,
        *,
        run_id: str,
        loop_state: Dict[str, Any],
        task: Task,
        tenant_id: Optional[str],
        session_id: Optional[str],
    ) -> str:
        checkpoint_id = str(uuid.uuid4())
        payload = {
            "checkpoint_id": checkpoint_id,
            "run_id": run_id,
            "loop_state": dict(loop_state),
            "task": task.snapshot(),
        }
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id or run_id,
            key=self._key(run_id),
            value=payload,
        )
        return checkpoint_id

    def load(
        self, *, run_id: str, tenant_id: Optional[str], session_id: Optional[str]
    ) -> Optional[Dict[str, Any]]:
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=session_id or run_id, key=self._key(run_id)
        )
        return payload if isinstance(payload, dict) else None

EventLoopError

Bases: AgentError

Error del bucle de eventos (reintentos agotados, estado inválido).

Source code in src/ciel/orchestration/agent.py

class EventLoopError(AgentError):
    """Error del bucle de eventos (reintentos agotados, estado inválido)."""

EventLoopStep dataclass

Resultado de un intento de ejecución de una tarea en el loop.

Source code in src/ciel/orchestration/agent.py

@dataclass
class EventLoopStep:
    """Resultado de un intento de ejecución de una tarea en el loop."""

    attempt: int
    succeeded: bool
    output: Any = None
    error: Optional[str] = None
    latency_ms: float = 0.0

MemoryStore

Bases: SqliteStateBackend

Alias retrocompatible a :class:SqliteStateBackend (SQLite en disco).

MemoryStore es SQLite en disco desde F5 — el nombre engaña. Para F15 se refactorizó para heredar de StateBackend de modo que cualquier store que hoy recibe un MemoryStore puede recibir también un PostgresStateBackend compartido sin cambios de API.

Se sigue construyendo con la misma firma: MemoryStore(db_path).

Source code in src/ciel/runtime/memory.py

class MemoryStore(SqliteStateBackend):
    """Alias retrocompatible a :class:`SqliteStateBackend` (SQLite en disco).

    ``MemoryStore`` es SQLite en disco desde F5 — el nombre engaña. Para
    F15 se refactorizó para heredar de ``StateBackend`` de modo que cualquier
    store que hoy recibe un ``MemoryStore`` puede recibir también un
    ``PostgresStateBackend`` compartido sin cambios de API.

    Se sigue construyendo con la misma firma: ``MemoryStore(db_path)``.
    """

    def __init__(self, db_path: str) -> None:
        super().__init__(db_path)

SessionStore

Estado de session durable por tenant, sobre MemoryStore.

Claves internas (todas namespaced con session: para no colisionar con otros módulos que usan el mismo MemoryStore):

• session:<sid>:turns -> {"version", "turns": [turn, ...]}
• session:<sid>:state -> {key: value} (estado arbitrario)
• session:<sid>:board_links-> {"links": [board_task_id, ...]}
• session_index:<tenant> -> {"sessions": [sid, ...]} (índice)

Source code in src/ciel/orchestration/session.py

class SessionStore:
    """Estado de session durable por tenant, sobre ``MemoryStore``.

    Claves internas (todas namespaced con ``session:`` para no colisionar con
    otros módulos que usan el mismo ``MemoryStore``):

    * ``session:<sid>:turns``      -> {"version", "turns": [turn, ...]}
    * ``session:<sid>:state``      -> {key: value}  (estado arbitrario)
    * ``session:<sid>:board_links``-> {"links": [board_task_id, ...]}
    * ``session_index:<tenant>``   -> {"sessions": [sid, ...]}  (índice)
    """

    def __init__(self, memory_store: MemoryStore) -> None:
        self.memory = memory_store

    # -- keys ---------------------------------------------------------------
    @staticmethod
    def _turns_key(session_id: str) -> str:
        return f"session:{session_id}:turns"

    @staticmethod
    def _state_key(session_id: str) -> str:
        return f"session:{session_id}:state"

    @staticmethod
    def _links_key(session_id: str) -> str:
        return f"session:{session_id}:board_links"

    @staticmethod
    def _index_key(tenant_id: Optional[str]) -> str:
        sentinel = tenant_id if tenant_id is not None else "__none__"
        return f"session_index:{sentinel}"

    # -- index --------------------------------------------------------------
    def _load_index(self, tenant_id: Optional[str]) -> List[str]:
        # El índice se guarda con session_id == tenant_id (determinista) para
        # no depender de una session concreta.
        sid = tenant_id if tenant_id is not None else "__none__"
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=sid, key=self._index_key(tenant_id)
        )
        if isinstance(payload, dict):
            return list(payload.get("sessions", []))
        return []

    def _register(self, *, tenant_id: Optional[str], session_id: str) -> None:
        sessions = self._load_index(tenant_id)
        if session_id not in sessions:
            sessions.append(session_id)
            sid = tenant_id if tenant_id is not None else "__none__"
            self.memory.set(
                tenant_id=tenant_id,
                session_id=sid,
                key=self._index_key(tenant_id),
                value={"sessions": sessions},
            )

    # -- turns --------------------------------------------------------------
    def append_turn(
        self, *, tenant_id: Optional[str], session_id: str, turn: Dict[str, Any]
    ) -> None:
        """Añade un turno al historial de la session y lo registra en el índice."""
        turn = dict(turn)
        turn.setdefault("ts", time.time())
        turns = self.history(tenant_id=tenant_id, session_id=session_id)
        turns.append(turn)
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id,
            key=self._turns_key(session_id),
            value={"version": SESSION_VERSION, "turns": turns},
        )
        self._register(tenant_id=tenant_id, session_id=session_id)

    def history(self, *, tenant_id: Optional[str], session_id: str) -> List[Dict[str, Any]]:
        """Recupera el historial de turnos (vacío si la session no existe)."""
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=session_id, key=self._turns_key(session_id)
        )
        if isinstance(payload, dict):
            return [dict(t) for t in payload.get("turns", [])]
        return []

    # -- arbitrary state ----------------------------------------------------
    def _load_state_blob(self, tenant_id: Optional[str], session_id: str) -> Dict[str, Any]:
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=session_id, key=self._state_key(session_id)
        )
        return dict(payload) if isinstance(payload, dict) else {}

    def save_state(
        self, *, tenant_id: Optional[str], session_id: str, key: str, value: Any
    ) -> None:
        state = self._load_state_blob(tenant_id, session_id)
        state[key] = value
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id,
            key=self._state_key(session_id),
            value=state,
        )
        self._register(tenant_id=tenant_id, session_id=session_id)

    def load_state(
        self, *, tenant_id: Optional[str], session_id: str, key: str, default: Any = None
    ) -> Any:
        state = self._load_state_blob(tenant_id, session_id)
        return state.get(key, default)

    # -- board integration --------------------------------------------------
    def link_board_task(
        self, *, tenant_id: Optional[str], session_id: str, board_task_id: str
    ) -> None:
        """Vincula una tarea del board (mismo tenant) con la session."""
        links = self.board_links(tenant_id=tenant_id, session_id=session_id)
        if board_task_id not in links:
            links.append(board_task_id)
            self.memory.set(
                tenant_id=tenant_id,
                session_id=session_id,
                key=self._links_key(session_id),
                value={"links": links},
            )

    def board_links(self, *, tenant_id: Optional[str], session_id: str) -> List[str]:
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=session_id, key=self._links_key(session_id)
        )
        if isinstance(payload, dict):
            return list(payload.get("links", []))
        return []

    # -- listing ------------------------------------------------------------
    def list_sessions(self, *, tenant_id: Optional[str]) -> List[str]:
        """Lista los ids de session registrados para un tenant."""
        return self._load_index(tenant_id)

append_turn(*, tenant_id: Optional[str], session_id: str, turn: Dict[str, Any]) -> None

Añade un turno al historial de la session y lo registra en el índice.

Source code in src/ciel/orchestration/session.py

def append_turn(
    self, *, tenant_id: Optional[str], session_id: str, turn: Dict[str, Any]
) -> None:
    """Añade un turno al historial de la session y lo registra en el índice."""
    turn = dict(turn)
    turn.setdefault("ts", time.time())
    turns = self.history(tenant_id=tenant_id, session_id=session_id)
    turns.append(turn)
    self.memory.set(
        tenant_id=tenant_id,
        session_id=session_id,
        key=self._turns_key(session_id),
        value={"version": SESSION_VERSION, "turns": turns},
    )
    self._register(tenant_id=tenant_id, session_id=session_id)

history(*, tenant_id: Optional[str], session_id: str) -> List[Dict[str, Any]]

Recupera el historial de turnos (vacío si la session no existe).

Source code in src/ciel/orchestration/session.py

def history(self, *, tenant_id: Optional[str], session_id: str) -> List[Dict[str, Any]]:
    """Recupera el historial de turnos (vacío si la session no existe)."""
    payload = self.memory.get(
        tenant_id=tenant_id, session_id=session_id, key=self._turns_key(session_id)
    )
    if isinstance(payload, dict):
        return [dict(t) for t in payload.get("turns", [])]
    return []

link_board_task(*, tenant_id: Optional[str], session_id: str, board_task_id: str) -> None

Vincula una tarea del board (mismo tenant) con la session.

Source code in src/ciel/orchestration/session.py

def link_board_task(
    self, *, tenant_id: Optional[str], session_id: str, board_task_id: str
) -> None:
    """Vincula una tarea del board (mismo tenant) con la session."""
    links = self.board_links(tenant_id=tenant_id, session_id=session_id)
    if board_task_id not in links:
        links.append(board_task_id)
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id,
            key=self._links_key(session_id),
            value={"links": links},
        )

list_sessions(*, tenant_id: Optional[str]) -> List[str]

Lista los ids de session registrados para un tenant.

Source code in src/ciel/orchestration/session.py

def list_sessions(self, *, tenant_id: Optional[str]) -> List[str]:
    """Lista los ids de session registrados para un tenant."""
    return self._load_index(tenant_id)

Supervisor

Source code in src/ciel/orchestration/supervisor.py

class Supervisor:
    def __init__(
        self,
        max_attempts: int = 2,
        timeout_s: float = 2.0,
        budget: Optional[Any] = None,
        rate_limiter: Optional[Any] = None,
        agent_counter: Optional[Any] = None,
        rate_limit: int = 0,
    ) -> None:
        self.max_attempts = max_attempts
        self.timeout_s = timeout_s
        self.budget = budget
        self.rate_limiter = rate_limiter
        self.agent_counter = agent_counter
        self.rate_limit = rate_limit
        self._results: Dict[str, WorkerResult] = {}

    async def run(
        self,
        step_id: str,
        worker: Worker,
        payload: Optional[Dict[str, Any]] = None,
        worker_id: str = "worker-1",
    ) -> WorkerResult:
        key = f"{step_id}:{worker_id}"
        if self.agent_counter is not None and self.budget is not None:
            exceeded = self.agent_counter.exceed(self.budget)
            if exceeded:
                result = WorkerResult(
                    worker_id=worker_id,
                    attempts=0,
                    failed=True,
                    error=f"budget rejection before run: {exceeded}",
                )
                self._results[key] = result
                return result
        if self.rate_limiter is not None and self.rate_limit > 0:
            rate_key = step_id or worker_id
            exceeded = self.rate_limiter.check(rate_key, self.rate_limit)
            if exceeded:
                result = WorkerResult(
                    worker_id=worker_id,
                    attempts=0,
                    failed=True,
                    error=f"rate limit rejection before run: {exceeded}",
                )
                self._results[key] = result
                return result
        attempt = 0
        start = time.perf_counter()
        while attempt < self.max_attempts:
            attempt += 1
            if self.agent_counter is not None:
                self.agent_counter.consume_tool(1)
            ctx = WorkerContext(step_id=step_id, worker_id=worker_id, payload=payload)
            try:
                output = await asyncio.wait_for(worker(ctx), timeout=self.timeout_s)
                result = WorkerResult(
                    worker_id=worker_id,
                    output=output,
                    attempts=attempt,
                    latency_ms=(time.perf_counter() - start) * 1000.0,
                )
                self._results[key] = result
                return result
            except Exception as exc:
                if attempt >= self.max_attempts:
                    result = WorkerResult(
                        worker_id=worker_id,
                        attempts=attempt,
                        latency_ms=(time.perf_counter() - start) * 1000.0,
                        failed=True,
                        error=str(exc),
                    )
                    self._results[key] = result
                    return result
        return self._results[key]

    def results(self) -> Dict[str, WorkerResult]:
        return dict(self._results)

Task dataclass

Unidad de trabajo autónomo durable (estilo ADK/AutoGen task).

status: pending | running | succeeded | failed. attempts se incrementa por cada ejecución del handler.

Source code in src/ciel/orchestration/agent.py

@dataclass
class Task:
    """Unidad de trabajo autónomo durable (estilo ADK/AutoGen task).

    ``status``: ``pending`` | ``running`` | ``succeeded`` | ``failed``.
    ``attempts`` se incrementa por cada ejecución del handler.
    """

    goal: str
    payload: Dict[str, Any] = field(default_factory=dict)
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    status: str = "pending"
    attempts: int = 0
    result: Any = None
    error: Optional[str] = None
    created_at: float = field(default_factory=lambda: time.time())
    updated_at: float = field(default_factory=lambda: time.time())

    def snapshot(self) -> Dict[str, Any]:
        return {
            "id": self.id,
            "goal": self.goal,
            "payload": dict(self.payload),
            "status": self.status,
            "attempts": self.attempts,
            "result": self.result,
            "error": self.error,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
        }

    @classmethod
    def from_snapshot(cls, snap: Dict[str, Any]) -> "Task":
        return cls(
            id=snap.get("id", str(uuid.uuid4())),
            goal=snap.get("goal", ""),
            payload=dict(snap.get("payload", {})),
            status=snap.get("status", "pending"),
            attempts=snap.get("attempts", 0),
            result=snap.get("result"),
            error=snap.get("error"),
            created_at=snap.get("created_at", time.time()),
            updated_at=snap.get("updated_at", time.time()),
        )

    def mark_running(self) -> None:
        self.status = "running"
        self.updated_at = time.time()

    def mark_succeeded(self, result: Any) -> None:
        self.status = "succeeded"
        self.result = result
        self.error = None
        self.updated_at = time.time()

    def mark_failed(self, error: str) -> None:
        self.status = "failed"
        self.error = error
        self.updated_at = time.time()

TaskError

Bases: AgentError

Error de una tarea individual tras agotar reintentos.

Source code in src/ciel/orchestration/agent.py

class TaskError(AgentError):
    """Error de una tarea individual tras agotar reintentos."""

WorkerContext dataclass

Source code in src/ciel/orchestration/supervisor.py

@dataclass
class WorkerContext:
    step_id: str
    worker_id: str
    payload: Optional[Dict[str, Any]] = None

SESSION_VERSION = 1 module-attribute

__all__ = ['SessionStore', 'SessionError'] module-attribute

MemoryStore

Bases: SqliteStateBackend

Alias retrocompatible a :class:SqliteStateBackend (SQLite en disco).

MemoryStore es SQLite en disco desde F5 — el nombre engaña. Para F15 se refactorizó para heredar de StateBackend de modo que cualquier store que hoy recibe un MemoryStore puede recibir también un PostgresStateBackend compartido sin cambios de API.

Se sigue construyendo con la misma firma: MemoryStore(db_path).

Source code in src/ciel/runtime/memory.py

class MemoryStore(SqliteStateBackend):
    """Alias retrocompatible a :class:`SqliteStateBackend` (SQLite en disco).

    ``MemoryStore`` es SQLite en disco desde F5 — el nombre engaña. Para
    F15 se refactorizó para heredar de ``StateBackend`` de modo que cualquier
    store que hoy recibe un ``MemoryStore`` puede recibir también un
    ``PostgresStateBackend`` compartido sin cambios de API.

    Se sigue construyendo con la misma firma: ``MemoryStore(db_path)``.
    """

    def __init__(self, db_path: str) -> None:
        super().__init__(db_path)

SessionError

Bases: Exception

Source code in src/ciel/orchestration/session.py

class SessionError(Exception):
    pass

SessionStore

Estado de session durable por tenant, sobre MemoryStore.

Claves internas (todas namespaced con session: para no colisionar con otros módulos que usan el mismo MemoryStore):

• session:<sid>:turns -> {"version", "turns": [turn, ...]}
• session:<sid>:state -> {key: value} (estado arbitrario)
• session:<sid>:board_links-> {"links": [board_task_id, ...]}
• session_index:<tenant> -> {"sessions": [sid, ...]} (índice)

Source code in src/ciel/orchestration/session.py

class SessionStore:
    """Estado de session durable por tenant, sobre ``MemoryStore``.

    Claves internas (todas namespaced con ``session:`` para no colisionar con
    otros módulos que usan el mismo ``MemoryStore``):

    * ``session:<sid>:turns``      -> {"version", "turns": [turn, ...]}
    * ``session:<sid>:state``      -> {key: value}  (estado arbitrario)
    * ``session:<sid>:board_links``-> {"links": [board_task_id, ...]}
    * ``session_index:<tenant>``   -> {"sessions": [sid, ...]}  (índice)
    """

    def __init__(self, memory_store: MemoryStore) -> None:
        self.memory = memory_store

    # -- keys ---------------------------------------------------------------
    @staticmethod
    def _turns_key(session_id: str) -> str:
        return f"session:{session_id}:turns"

    @staticmethod
    def _state_key(session_id: str) -> str:
        return f"session:{session_id}:state"

    @staticmethod
    def _links_key(session_id: str) -> str:
        return f"session:{session_id}:board_links"

    @staticmethod
    def _index_key(tenant_id: Optional[str]) -> str:
        sentinel = tenant_id if tenant_id is not None else "__none__"
        return f"session_index:{sentinel}"

    # -- index --------------------------------------------------------------
    def _load_index(self, tenant_id: Optional[str]) -> List[str]:
        # El índice se guarda con session_id == tenant_id (determinista) para
        # no depender de una session concreta.
        sid = tenant_id if tenant_id is not None else "__none__"
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=sid, key=self._index_key(tenant_id)
        )
        if isinstance(payload, dict):
            return list(payload.get("sessions", []))
        return []

    def _register(self, *, tenant_id: Optional[str], session_id: str) -> None:
        sessions = self._load_index(tenant_id)
        if session_id not in sessions:
            sessions.append(session_id)
            sid = tenant_id if tenant_id is not None else "__none__"
            self.memory.set(
                tenant_id=tenant_id,
                session_id=sid,
                key=self._index_key(tenant_id),
                value={"sessions": sessions},
            )

    # -- turns --------------------------------------------------------------
    def append_turn(
        self, *, tenant_id: Optional[str], session_id: str, turn: Dict[str, Any]
    ) -> None:
        """Añade un turno al historial de la session y lo registra en el índice."""
        turn = dict(turn)
        turn.setdefault("ts", time.time())
        turns = self.history(tenant_id=tenant_id, session_id=session_id)
        turns.append(turn)
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id,
            key=self._turns_key(session_id),
            value={"version": SESSION_VERSION, "turns": turns},
        )
        self._register(tenant_id=tenant_id, session_id=session_id)

    def history(self, *, tenant_id: Optional[str], session_id: str) -> List[Dict[str, Any]]:
        """Recupera el historial de turnos (vacío si la session no existe)."""
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=session_id, key=self._turns_key(session_id)
        )
        if isinstance(payload, dict):
            return [dict(t) for t in payload.get("turns", [])]
        return []

    # -- arbitrary state ----------------------------------------------------
    def _load_state_blob(self, tenant_id: Optional[str], session_id: str) -> Dict[str, Any]:
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=session_id, key=self._state_key(session_id)
        )
        return dict(payload) if isinstance(payload, dict) else {}

    def save_state(
        self, *, tenant_id: Optional[str], session_id: str, key: str, value: Any
    ) -> None:
        state = self._load_state_blob(tenant_id, session_id)
        state[key] = value
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id,
            key=self._state_key(session_id),
            value=state,
        )
        self._register(tenant_id=tenant_id, session_id=session_id)

    def load_state(
        self, *, tenant_id: Optional[str], session_id: str, key: str, default: Any = None
    ) -> Any:
        state = self._load_state_blob(tenant_id, session_id)
        return state.get(key, default)

    # -- board integration --------------------------------------------------
    def link_board_task(
        self, *, tenant_id: Optional[str], session_id: str, board_task_id: str
    ) -> None:
        """Vincula una tarea del board (mismo tenant) con la session."""
        links = self.board_links(tenant_id=tenant_id, session_id=session_id)
        if board_task_id not in links:
            links.append(board_task_id)
            self.memory.set(
                tenant_id=tenant_id,
                session_id=session_id,
                key=self._links_key(session_id),
                value={"links": links},
            )

    def board_links(self, *, tenant_id: Optional[str], session_id: str) -> List[str]:
        payload = self.memory.get(
            tenant_id=tenant_id, session_id=session_id, key=self._links_key(session_id)
        )
        if isinstance(payload, dict):
            return list(payload.get("links", []))
        return []

    # -- listing ------------------------------------------------------------
    def list_sessions(self, *, tenant_id: Optional[str]) -> List[str]:
        """Lista los ids de session registrados para un tenant."""
        return self._load_index(tenant_id)

append_turn(*, tenant_id: Optional[str], session_id: str, turn: Dict[str, Any]) -> None

Añade un turno al historial de la session y lo registra en el índice.

Source code in src/ciel/orchestration/session.py

def append_turn(
    self, *, tenant_id: Optional[str], session_id: str, turn: Dict[str, Any]
) -> None:
    """Añade un turno al historial de la session y lo registra en el índice."""
    turn = dict(turn)
    turn.setdefault("ts", time.time())
    turns = self.history(tenant_id=tenant_id, session_id=session_id)
    turns.append(turn)
    self.memory.set(
        tenant_id=tenant_id,
        session_id=session_id,
        key=self._turns_key(session_id),
        value={"version": SESSION_VERSION, "turns": turns},
    )
    self._register(tenant_id=tenant_id, session_id=session_id)

history(*, tenant_id: Optional[str], session_id: str) -> List[Dict[str, Any]]

Recupera el historial de turnos (vacío si la session no existe).

Source code in src/ciel/orchestration/session.py

def history(self, *, tenant_id: Optional[str], session_id: str) -> List[Dict[str, Any]]:
    """Recupera el historial de turnos (vacío si la session no existe)."""
    payload = self.memory.get(
        tenant_id=tenant_id, session_id=session_id, key=self._turns_key(session_id)
    )
    if isinstance(payload, dict):
        return [dict(t) for t in payload.get("turns", [])]
    return []

link_board_task(*, tenant_id: Optional[str], session_id: str, board_task_id: str) -> None

Vincula una tarea del board (mismo tenant) con la session.

Source code in src/ciel/orchestration/session.py

def link_board_task(
    self, *, tenant_id: Optional[str], session_id: str, board_task_id: str
) -> None:
    """Vincula una tarea del board (mismo tenant) con la session."""
    links = self.board_links(tenant_id=tenant_id, session_id=session_id)
    if board_task_id not in links:
        links.append(board_task_id)
        self.memory.set(
            tenant_id=tenant_id,
            session_id=session_id,
            key=self._links_key(session_id),
            value={"links": links},
        )

list_sessions(*, tenant_id: Optional[str]) -> List[str]

Lista los ids de session registrados para un tenant.

Source code in src/ciel/orchestration/session.py

def list_sessions(self, *, tenant_id: Optional[str]) -> List[str]:
    """Lista los ids de session registrados para un tenant."""
    return self._load_index(tenant_id)

BoardTask

Source code in src/ciel/orchestration/board.py

class BoardTask:
    def __init__(
        self,
        id: str,
        title: str,
        status: str = "todo",
        assignee: Optional[str] = None,
        tenant_id: Optional[str] = None,
        metadata: Optional[Dict[str, str]] = None,
    ) -> None:
        self.id = id
        self.title = title
        self.status = status
        self.assignee = assignee
        self.tenant_id = tenant_id
        self.metadata = metadata or {}

KanbanBoard

In-memory kanban board, optionally backed by a durable SQLite store.

When path is None (the default) the board keeps tasks in a process local dict exactly as before -- this preserves the existing in-memory API used by the gateway and the CLI. When path is provided the board transparently reads and writes every task to a SQLite database opened in WAL mode, so state survives process restarts.

Source code in src/ciel/orchestration/board.py

class KanbanBoard:
    """In-memory kanban board, optionally backed by a durable SQLite store.

    When ``path`` is ``None`` (the default) the board keeps tasks in a process
    local ``dict`` exactly as before -- this preserves the existing in-memory
    API used by the gateway and the CLI. When ``path`` is provided the board
    transparently reads and writes every task to a SQLite database opened in
    WAL mode, so state survives process restarts.
    """

    def __init__(self, path: Optional[str | Path] = None) -> None:
        self._path = None if path is None else str(path)
        self._tasks: Dict[str, BoardTask] = {}
        self._conn: Optional[sqlite3.Connection] = None
        if self._path is not None:
            self._conn = sqlite3.connect(self._path)
            self._conn.execute("PRAGMA journal_mode=WAL")
            self._conn.execute(
                """
                CREATE TABLE IF NOT EXISTS board_tasks (
                    id TEXT PRIMARY KEY,
                    title TEXT NOT NULL,
                    status TEXT NOT NULL,
                    assignee TEXT,
                    tenant_id TEXT,
                    metadata_json TEXT
                )
                """
            )
            self._conn.commit()

    # -- helpers -------------------------------------------------------------
    def _row_to_task(self, row: sqlite3.Row) -> BoardTask:
        return BoardTask(
            id=row["id"],
            title=row["title"],
            status=row["status"],
            assignee=row["assignee"],
            tenant_id=row["tenant_id"],
            metadata=json.loads(row["metadata_json"]) if row["metadata_json"] else {},
        )

    def _task_to_row(self, task: BoardTask) -> Dict[str, Any]:
        return {
            "id": task.id,
            "title": task.title,
            "status": task.status,
            "assignee": task.assignee,
            "tenant_id": task.tenant_id,
            "metadata_json": json.dumps(task.metadata or {}),
        }

    # -- public API ----------------------------------------------------------
    def add_task(self, task: BoardTask) -> BoardTask:
        if self._conn is None:
            self._tasks[task.id] = task
            return task

        row = self._task_to_row(task)
        self._conn.execute(
            """
            INSERT INTO board_tasks (id, title, status, assignee, tenant_id, metadata_json)
            VALUES (:id, :title, :status, :assignee, :tenant_id, :metadata_json)
            ON CONFLICT(id) DO UPDATE SET
                title=excluded.title,
                status=excluded.status,
                assignee=excluded.assignee,
                tenant_id=excluded.tenant_id,
                metadata_json=excluded.metadata_json
            """,
            row,
        )
        self._conn.commit()
        return task

    def assign(self, task_id: str, assignee: str) -> Optional[BoardTask]:
        if self._conn is None:
            task = self._tasks.get(task_id)
            if not task:
                return None
            task.assignee = assignee
            task.status = "in_progress"
            return task

        cur = self._conn.execute(
            """
            UPDATE board_tasks
            SET assignee = ?, status = 'in_progress'
            WHERE id = ?
            """,
            (assignee, task_id),
        )
        if cur.rowcount == 0:
            return None
        self._conn.commit()
        return self.show(task_id)

    def move(self, task_id: str, status: str) -> Optional[BoardTask]:
        if self._conn is None:
            task = self._tasks.get(task_id)
            if not task:
                return None
            task.status = status
            return task

        cur = self._conn.execute(
            "UPDATE board_tasks SET status = ? WHERE id = ?",
            (status, task_id),
        )
        if cur.rowcount == 0:
            return None
        self._conn.commit()
        return self.show(task_id)

    def list_tasks(
        self,
        status: Optional[str] = None,
        assignee: Optional[str] = None,
        tenant_id: Optional[str] = None,
    ) -> List[BoardTask]:
        if self._conn is None:
            out: List[BoardTask] = []
            for task in self._tasks.values():
                if status and task.status != status:
                    continue
                if assignee and task.assignee != assignee:
                    continue
                if tenant_id and task.tenant_id != tenant_id:
                    continue
                out.append(task)
            return out

        clauses: List[str] = []
        params: List[Any] = []
        if status:
            clauses.append("status = ?")
            params.append(status)
        if assignee:
            clauses.append("assignee = ?")
            params.append(assignee)
        if tenant_id:
            clauses.append("tenant_id = ?")
            params.append(tenant_id)
        where = (" WHERE " + " AND ".join(clauses)) if clauses else ""
        self._conn.row_factory = sqlite3.Row
        rows = self._conn.execute(
            f"SELECT id, title, status, assignee, tenant_id, metadata_json FROM board_tasks{where}",
            params,
        ).fetchall()
        return [self._row_to_task(r) for r in rows]

    def show(self, task_id: str) -> Optional[BoardTask]:
        if self._conn is None:
            return self._tasks.get(task_id)

        self._conn.row_factory = sqlite3.Row
        row = self._conn.execute(
            "SELECT id, title, status, assignee, tenant_id, metadata_json FROM board_tasks WHERE id = ?",
            (task_id,),
        ).fetchone()
        if row is None:
            return None
        return self._row_to_task(row)

    def close(self) -> None:
        if self._conn is not None:
            # Flush the WAL back into the main database file so the
            # board.sqlite-wal / board.sqlite-shm side-car files are released
            # (important on Windows where they otherwise stay locked).
            try:
                self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
            except sqlite3.Error:
                pass
            self._conn.close()
            self._conn = None

    def __enter__(self) -> "KanbanBoard":
        return self

    def __exit__(self, *exc: Any) -> None:
        self.close()

Worker = Callable[[WorkerContext], Any] module-attribute

Supervisor

Source code in src/ciel/orchestration/supervisor.py

class Supervisor:
    def __init__(
        self,
        max_attempts: int = 2,
        timeout_s: float = 2.0,
        budget: Optional[Any] = None,
        rate_limiter: Optional[Any] = None,
        agent_counter: Optional[Any] = None,
        rate_limit: int = 0,
    ) -> None:
        self.max_attempts = max_attempts
        self.timeout_s = timeout_s
        self.budget = budget
        self.rate_limiter = rate_limiter
        self.agent_counter = agent_counter
        self.rate_limit = rate_limit
        self._results: Dict[str, WorkerResult] = {}

    async def run(
        self,
        step_id: str,
        worker: Worker,
        payload: Optional[Dict[str, Any]] = None,
        worker_id: str = "worker-1",
    ) -> WorkerResult:
        key = f"{step_id}:{worker_id}"
        if self.agent_counter is not None and self.budget is not None:
            exceeded = self.agent_counter.exceed(self.budget)
            if exceeded:
                result = WorkerResult(
                    worker_id=worker_id,
                    attempts=0,
                    failed=True,
                    error=f"budget rejection before run: {exceeded}",
                )
                self._results[key] = result
                return result
        if self.rate_limiter is not None and self.rate_limit > 0:
            rate_key = step_id or worker_id
            exceeded = self.rate_limiter.check(rate_key, self.rate_limit)
            if exceeded:
                result = WorkerResult(
                    worker_id=worker_id,
                    attempts=0,
                    failed=True,
                    error=f"rate limit rejection before run: {exceeded}",
                )
                self._results[key] = result
                return result
        attempt = 0
        start = time.perf_counter()
        while attempt < self.max_attempts:
            attempt += 1
            if self.agent_counter is not None:
                self.agent_counter.consume_tool(1)
            ctx = WorkerContext(step_id=step_id, worker_id=worker_id, payload=payload)
            try:
                output = await asyncio.wait_for(worker(ctx), timeout=self.timeout_s)
                result = WorkerResult(
                    worker_id=worker_id,
                    output=output,
                    attempts=attempt,
                    latency_ms=(time.perf_counter() - start) * 1000.0,
                )
                self._results[key] = result
                return result
            except Exception as exc:
                if attempt >= self.max_attempts:
                    result = WorkerResult(
                        worker_id=worker_id,
                        attempts=attempt,
                        latency_ms=(time.perf_counter() - start) * 1000.0,
                        failed=True,
                        error=str(exc),
                    )
                    self._results[key] = result
                    return result
        return self._results[key]

    def results(self) -> Dict[str, WorkerResult]:
        return dict(self._results)

WorkerContext dataclass

Source code in src/ciel/orchestration/supervisor.py

@dataclass
class WorkerContext:
    step_id: str
    worker_id: str
    payload: Optional[Dict[str, Any]] = None

WorkerResult dataclass

Source code in src/ciel/orchestration/supervisor.py

@dataclass
class WorkerResult:
    worker_id: str
    output: Any = None
    error: Optional[str] = None
    attempts: int = 0
    latency_ms: float = 0.0
    failed: bool = False