ciel.cli — Interfaz de línea de comandos

Aplicación Typer que expone el comando ciel. El punto de entrada registrado en PyPI es ciel.cli.main:app.

ciel.cli

Ciel CLI application and command modules.

app = typer.Typer(name='ciel', help='Ciel Agent Framework CLI', no_args_is_help=True) module-attribute

REPO_ROOT = Path(__file__).resolve().parents[2] module-attribute

__version__ = _pkg_version('ciel') module-attribute

app = typer.Typer(name='ciel', help='Ciel Agent Framework CLI', no_args_is_help=True) module-attribute

console = Console() module-attribute

ChatMessage dataclass

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

@dataclass(frozen=True)
class ChatMessage:
    role: str
    content: ChatContent
    name: Optional[str] = None
    tool_call_id: Optional[str] = None
    tool_calls: Optional[list[dict[str, Any]]] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

    def text(self) -> str:
        """Extract plain text from content, tolerant to multimodal parts.

        - ``str`` content is returned verbatim.
        - ``list`` content concatenates the ``text`` of every part whose type
          is ``"text"`` but drops images/audio/video, so consumers (CLI,
          compression, ``AgentResponse.text``) see only readable text.
        """
        content = self.content
        if isinstance(content, str):
            return content
        if not isinstance(content, list):
            return ""
        parts: list[str] = []
        for part in content:
            if isinstance(part, dict) and part.get("type") == "text":
                text = part.get("text", "")
                if isinstance(text, str):
                    parts.append(text)
        return "".join(parts)

text() -> str

Extract plain text from content, tolerant to multimodal parts.

• str content is returned verbatim.
• list content concatenates the text of every part whose type is "text" but drops images/audio/video, so consumers (CLI, compression, AgentResponse.text) see only readable text.

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

def text(self) -> str:
    """Extract plain text from content, tolerant to multimodal parts.

    - ``str`` content is returned verbatim.
    - ``list`` content concatenates the ``text`` of every part whose type
      is ``"text"`` but drops images/audio/video, so consumers (CLI,
      compression, ``AgentResponse.text``) see only readable text.
    """
    content = self.content
    if isinstance(content, str):
        return content
    if not isinstance(content, list):
        return ""
    parts: list[str] = []
    for part in content:
        if isinstance(part, dict) and part.get("type") == "text":
            text = part.get("text", "")
            if isinstance(text, str):
                parts.append(text)
    return "".join(parts)

ChatRequest dataclass

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

@dataclass(frozen=True)
class ChatRequest:
    messages: Sequence[ChatMessage]
    tools: Sequence[ToolSpec] = ()
    model: Optional[str] = None
    temperature: Optional[float] = None
    max_tokens: Optional[int] = None
    extra: Dict[str, Any] = field(default_factory=dict)

DefaultAgentRuntime

Concrete runtime wiring provider + tool execution with tracing.

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

class DefaultAgentRuntime:
    """Concrete runtime wiring provider + tool execution with tracing."""

    def __init__(
        self,
        *,
        provider: ChatProvider,
        dispatcher: DefaultToolDispatcher,
        registry: Optional[ProviderRegistry] = None,
        audit_sink: Optional[Any] = None,
        agent: str = "default",
        approval_policy: Optional[Any] = None,
    ) -> None:
        self.provider = provider
        self.dispatcher = dispatcher
        self.registry = registry
        self.audit_sink = audit_sink or NullAuditSink()
        self.agent = agent
        self.approval_policy = approval_policy

    async def _emit(self, event: AuditEvent, *, tenant_id: Optional[str] = None) -> AuditEvent:
        normalized = propagate(event, tenant_id=tenant_id)
        await self.audit_sink.write(normalized)
        return normalized

    async def run_agent_loop(
        self,
        *,
        request: ChatRequest,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
        limit: int = 32,
    ) -> AgentRuntimeResult:
        """Run the agentic tool loop (multi-turn ReAct when tools are present).

        The loop issues up to ``limit`` model completions. After each completion
        that requests tool calls, the requested tools are dispatched and their
        results are appended as messages; the model is then called again. The
        loop stops when (a) the model returns no tool calls (``finish_reason``
        ``stop``), (b) ``limit`` turns are exhausted, or (c) there are no tools
        registered in the request.

        Backward compatibility: when ``limit <= 1`` or ``request.tools`` is
        empty, exactly one completion is produced and no tool messages are
        appended (the historical single-step behaviour). Every tool result from
        every turn is preserved in ``loop_results`` so the facade can surface
        the full ``tool_results`` list.
        """
        session_id = request.extra.get("session_id") or str(uuid.uuid4())
        await self._emit(
            AuditEvent(event="agent.loop.start", session_id=session_id, agent=self.agent, tenant_id=tenant_id),
            tenant_id=tenant_id,
        )

        messages: List[ChatMessage] = list(request.messages)
        loop_results: List[ToolLoopResult] = []
        finish_reason = "stop"
        has_tools = bool(request.tools)

        # Multi-turn only when tools exist and the caller allows more than one step.
        max_turns = max(1, limit) if (has_tools and limit > 1) else 1

        final_response = None
        for _ in range(max_turns):
            response = await self.provider.complete(
                ChatRequest(
                    messages=tuple(messages),
                    tools=request.tools,
                    model=request.model,
                    temperature=request.temperature,
                    max_tokens=request.max_tokens,
                    extra={**request.extra, "session_id": session_id, "tenant_id": tenant_id},
                )
            )
            final_response = response
            messages.append(response.choice.message)
            tool_calls = _extract_tool_calls(response)

            if not tool_calls:
                # Model is done; stop the loop.
                finish_reason = response.choice.finish_reason or "stop"
                break

            dispatch_results: List[ToolResult] = []
            for call in tool_calls:
                call_arguments = call.get("arguments") or {}
                decision = None
                if self.approval_policy is not None and call.get("name") not in {None, ""}:
                    try:
                        from ciel.security import ApprovalRequest, ApprovalPolicy
                        if isinstance(self.approval_policy, type):
                            policy = self.approval_policy()
                        else:
                            policy = self.approval_policy
                        if hasattr(policy, "evaluate"):
                            decision = policy.evaluate(
                                ApprovalRequest(
                                    request_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                                    actor=tenant_id or "unknown",
                                    tool=call.get("name", ""),
                                    arguments=call_arguments,
                                    risk="medium",
                                    tenant=tenant_id,
                                )
                            )
                    except Exception:
                        decision = None
                if decision is not None and not decision.approved:
                    dispatch_results.append(
                        ToolResult(
                            id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                            name=call.get("name", ""),
                            error=f"ApprovalDenied: {decision.note or 'denied'}",
                            metadata={"tenant_id": tenant_id, "approval_decision": decision.note or "denied"},
                        )
                    )
                    continue
                dispatch_results.append(
                    await self.dispatcher.dispatch(
                        tenant_id=tenant_id,
                        toolset=toolset or self.dispatcher.default_toolset or "default",
                        name=call.get("name", ""),
                        arguments=call_arguments,
                        tool_call_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                    )
                )

            tool_turn = ToolLoopResult(
                turn_id=str(uuid.uuid4()),
                messages=tuple(messages),
                tool_results=tuple(dispatch_results),
                finish_reason="tool_calls",
                tenant_id=tenant_id,
                metadata={"session_id": session_id, "toolset": toolset, "turn": len(loop_results) + 1},
            )
            loop_results.append(tool_turn)
            finish_reason = "tool_calls"
            await self._emit(
                AuditEvent(
                    event="agent.tool_calls.dispatched",
                    session_id=session_id,
                    agent=self.agent,
                    tool_call_id=dispatch_results[0].id if dispatch_results else None,
                    data={"tools": [r.name for r in dispatch_results], "turn": len(loop_results)},
                    tenant_id=tenant_id,
                ),
                tenant_id=tenant_id,
            )

            # Append tool results as tool messages so the next completion sees them.
            for res in dispatch_results:
                messages.append(
                    ChatMessage(
                        role="tool",
                        content=str(res.output) if res.error is None else (res.error or ""),
                        name=res.name,
                        tool_call_id=res.id,
                        metadata={"tenant_id": tenant_id, "tool_result": True, "error": res.error},
                    )
                )

        await self._emit(
            AuditEvent(event="agent.loop.end", session_id=session_id, agent=self.agent, tenant_id=tenant_id),
            tenant_id=tenant_id,
        )

        return AgentRuntimeResult(
            response=final_response,
            loop_results=tuple(loop_results),
            tenant_id=tenant_id,
            metadata={"session_id": session_id, "agent": self.agent},
        )

    async def stream_agent_loop(
        self,
        *,
        request: ChatRequest,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
        limit: int = 32,
    ) -> AsyncIterator[ToolLoopResult]:
        result = await self.run_agent_loop(
            request=request,
            tenant_id=tenant_id,
            toolset=toolset,
            limit=limit,
        )
        for turn in result.loop_results:
            yield turn

    async def stream_tokens(
        self,
        *,
        request: ChatRequest,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
    ) -> AsyncIterator[str]:
        """Stream incremental assistant tokens from the provider.

        Calls ``provider.stream`` (real SSE streaming) and re-emits the
        partial ``content`` of each incremental :class:`ChatResponse` as it
        arrives, so callers see the answer grow token by token.
        """
        chunks = await self.provider.stream(request=request)
        prior = ""
        for chunk in chunks:
            content = chunk.choice.message.text()
            if content != prior:
                yield content
                prior = content

run_agent_loop(*, request: ChatRequest, tenant_id: Optional[str] = None, toolset: Optional[str] = None, limit: int = 32) -> AgentRuntimeResult async

Run the agentic tool loop (multi-turn ReAct when tools are present).

The loop issues up to limit model completions. After each completion that requests tool calls, the requested tools are dispatched and their results are appended as messages; the model is then called again. The loop stops when (a) the model returns no tool calls (finish_reason stop), (b) limit turns are exhausted, or (c) there are no tools registered in the request.

Backward compatibility: when limit <= 1 or request.tools is empty, exactly one completion is produced and no tool messages are appended (the historical single-step behaviour). Every tool result from every turn is preserved in loop_results so the facade can surface the full tool_results list.

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

async def run_agent_loop(
    self,
    *,
    request: ChatRequest,
    tenant_id: Optional[str] = None,
    toolset: Optional[str] = None,
    limit: int = 32,
) -> AgentRuntimeResult:
    """Run the agentic tool loop (multi-turn ReAct when tools are present).

    The loop issues up to ``limit`` model completions. After each completion
    that requests tool calls, the requested tools are dispatched and their
    results are appended as messages; the model is then called again. The
    loop stops when (a) the model returns no tool calls (``finish_reason``
    ``stop``), (b) ``limit`` turns are exhausted, or (c) there are no tools
    registered in the request.

    Backward compatibility: when ``limit <= 1`` or ``request.tools`` is
    empty, exactly one completion is produced and no tool messages are
    appended (the historical single-step behaviour). Every tool result from
    every turn is preserved in ``loop_results`` so the facade can surface
    the full ``tool_results`` list.
    """
    session_id = request.extra.get("session_id") or str(uuid.uuid4())
    await self._emit(
        AuditEvent(event="agent.loop.start", session_id=session_id, agent=self.agent, tenant_id=tenant_id),
        tenant_id=tenant_id,
    )

    messages: List[ChatMessage] = list(request.messages)
    loop_results: List[ToolLoopResult] = []
    finish_reason = "stop"
    has_tools = bool(request.tools)

    # Multi-turn only when tools exist and the caller allows more than one step.
    max_turns = max(1, limit) if (has_tools and limit > 1) else 1

    final_response = None
    for _ in range(max_turns):
        response = await self.provider.complete(
            ChatRequest(
                messages=tuple(messages),
                tools=request.tools,
                model=request.model,
                temperature=request.temperature,
                max_tokens=request.max_tokens,
                extra={**request.extra, "session_id": session_id, "tenant_id": tenant_id},
            )
        )
        final_response = response
        messages.append(response.choice.message)
        tool_calls = _extract_tool_calls(response)

        if not tool_calls:
            # Model is done; stop the loop.
            finish_reason = response.choice.finish_reason or "stop"
            break

        dispatch_results: List[ToolResult] = []
        for call in tool_calls:
            call_arguments = call.get("arguments") or {}
            decision = None
            if self.approval_policy is not None and call.get("name") not in {None, ""}:
                try:
                    from ciel.security import ApprovalRequest, ApprovalPolicy
                    if isinstance(self.approval_policy, type):
                        policy = self.approval_policy()
                    else:
                        policy = self.approval_policy
                    if hasattr(policy, "evaluate"):
                        decision = policy.evaluate(
                            ApprovalRequest(
                                request_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                                actor=tenant_id or "unknown",
                                tool=call.get("name", ""),
                                arguments=call_arguments,
                                risk="medium",
                                tenant=tenant_id,
                            )
                        )
                except Exception:
                    decision = None
            if decision is not None and not decision.approved:
                dispatch_results.append(
                    ToolResult(
                        id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                        name=call.get("name", ""),
                        error=f"ApprovalDenied: {decision.note or 'denied'}",
                        metadata={"tenant_id": tenant_id, "approval_decision": decision.note or "denied"},
                    )
                )
                continue
            dispatch_results.append(
                await self.dispatcher.dispatch(
                    tenant_id=tenant_id,
                    toolset=toolset or self.dispatcher.default_toolset or "default",
                    name=call.get("name", ""),
                    arguments=call_arguments,
                    tool_call_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                )
            )

        tool_turn = ToolLoopResult(
            turn_id=str(uuid.uuid4()),
            messages=tuple(messages),
            tool_results=tuple(dispatch_results),
            finish_reason="tool_calls",
            tenant_id=tenant_id,
            metadata={"session_id": session_id, "toolset": toolset, "turn": len(loop_results) + 1},
        )
        loop_results.append(tool_turn)
        finish_reason = "tool_calls"
        await self._emit(
            AuditEvent(
                event="agent.tool_calls.dispatched",
                session_id=session_id,
                agent=self.agent,
                tool_call_id=dispatch_results[0].id if dispatch_results else None,
                data={"tools": [r.name for r in dispatch_results], "turn": len(loop_results)},
                tenant_id=tenant_id,
            ),
            tenant_id=tenant_id,
        )

        # Append tool results as tool messages so the next completion sees them.
        for res in dispatch_results:
            messages.append(
                ChatMessage(
                    role="tool",
                    content=str(res.output) if res.error is None else (res.error or ""),
                    name=res.name,
                    tool_call_id=res.id,
                    metadata={"tenant_id": tenant_id, "tool_result": True, "error": res.error},
                )
            )

    await self._emit(
        AuditEvent(event="agent.loop.end", session_id=session_id, agent=self.agent, tenant_id=tenant_id),
        tenant_id=tenant_id,
    )

    return AgentRuntimeResult(
        response=final_response,
        loop_results=tuple(loop_results),
        tenant_id=tenant_id,
        metadata={"session_id": session_id, "agent": self.agent},
    )

stream_tokens(*, request: ChatRequest, tenant_id: Optional[str] = None, toolset: Optional[str] = None) -> AsyncIterator[str] async

Stream incremental assistant tokens from the provider.

Calls provider.stream (real SSE streaming) and re-emits the partial content of each incremental :class:ChatResponse as it arrives, so callers see the answer grow token by token.

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

async def stream_tokens(
    self,
    *,
    request: ChatRequest,
    tenant_id: Optional[str] = None,
    toolset: Optional[str] = None,
) -> AsyncIterator[str]:
    """Stream incremental assistant tokens from the provider.

    Calls ``provider.stream`` (real SSE streaming) and re-emits the
    partial ``content`` of each incremental :class:`ChatResponse` as it
    arrives, so callers see the answer grow token by token.
    """
    chunks = await self.provider.stream(request=request)
    prior = ""
    for chunk in chunks:
        content = chunk.choice.message.text()
        if content != prior:
            yield content
            prior = content

DefaultToolDispatcher

Dispatch tool requests to a configured ToolProvider.

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

class DefaultToolDispatcher:
    """Dispatch tool requests to a configured ToolProvider."""

    provider: ToolProvider
    default_toolset: Optional[str]

    def __init__(self, provider: ToolProvider, default_toolset: Optional[str] = None) -> None:
        self.provider = provider
        self.default_toolset = default_toolset or getattr(provider.registry, "default_toolset", None)

    async def dispatch(
        self,
        *,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
        name: str,
        arguments: Dict[str, Any],
        tool_call_id: str,
    ) -> ToolResult:
        result = await self.provider.execute(
            tenant_id=tenant_id,
            toolset=toolset or self.default_toolset or "default",
            name=name,
            arguments=arguments,
            tool_call_id=tool_call_id,
        )
        result.metadata.setdefault("tenant_id", tenant_id)
        return result

    async def dispatch_all(
        self,
        *,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
        calls: Sequence[Dict[str, Any]],
    ):
        results: List[ToolResult] = []
        for call in calls:
            call_tenant_id = tenant_id or call.get("metadata", {}).get("tenant_id")
            results.append(
                await self.dispatch(
                    tenant_id=call_tenant_id,
                    toolset=toolset or self.default_toolset or "default",
                    name=call["name"],
                    arguments=call.get("arguments", {}),
                    tool_call_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                )
            )
        return results

ToolProvider dataclass

Concrete tool provider used by the runtime's tool dispatcher.

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

@dataclass(frozen=True)
class ToolProvider:
    """Concrete tool provider used by the runtime's tool dispatcher."""

    registry: ToolRegistry
    require_tenant_on_execution: bool = True

    async def tool_specs(self, tenant_id: Optional[str], toolset: str) -> Sequence[ToolSpec]:
        return tuple(getattr(self.registry, "_toolsets", {}).get(toolset, ToolsetSchema(name=toolset, description="")).tools)

    async def execute(
        self,
        *,
        tenant_id: Optional[str],
        toolset: Optional[str],
        name: str,
        arguments: Dict[str, Any],
        tool_call_id: str,
    ) -> ToolResult:
        target_toolset = toolset or (self.registry.default_toolset or "default")
        if self.require_tenant_on_execution and not tenant_id:
            raise TenantRequired(f"tenant_id is required to execute tool '{name}' in toolset '{target_toolset}'.")
        tool = self.registry.get_tool(toolset=target_toolset, name=name)
        if tool is None:
            return ToolResult(id=tool_call_id, name=name, error=f"unknown tool: {target_toolset}.{name}", metadata={"tenant_id": tenant_id})
        if getattr(tool, "required_tenant", False) and not tenant_id:
            raise TenantRequired(f"tool '{name}' requires tenant_id, but none was provided.")
        if tool.callable_ is None:
            output = {"arguments": arguments, "description": tool.spec.description}
            return ToolResult(id=tool_call_id, name=name, output=output, metadata={"tenant_id": tenant_id})
        # Official tool callable contract:
        #   callable_(arguments: dict, *, tool_call_id: str, tenant_id: str | None) -> ToolResult | dict | Any
        try:
            result = tool.callable_(arguments, tool_call_id=tool_call_id, tenant_id=tenant_id)
            if inspect.isawaitable(result):
                result = await result
        except Exception as exc:  # noqa: BLE001 — surface tool errors as ToolResult
            return ToolResult(id=tool_call_id, name=name, error=f"{type(exc).__name__}: {exc}", metadata={"tenant_id": tenant_id})
        if isinstance(result, ToolResult):
            if not result.metadata.get("tenant_id"):
                result.metadata["tenant_id"] = tenant_id
            return result
        return ToolResult(id=tool_call_id, name=name, output=result, metadata={"tenant_id": tenant_id})

ToolRegistry

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

class ToolRegistry:
    def __init__(self, *, default_toolset: Optional[str] = None) -> None:
        self._toolsets: Dict[str, ToolsetSchema] = {}
        self._tools: Dict[str, Dict[str, Tool]] = {}
        self._tenant_tools: Dict[str, Dict[str, Dict[str, Tool]]] = {}
        self.default_toolset = default_toolset

    def register_toolset(self, schema: ToolsetSchema) -> None:
        self._toolsets[schema.name] = schema
        self._tools.setdefault(schema.name, {})
        self._tenant_tools.setdefault(schema.name, {})
        for tool in schema.tools:
            self._tools[schema.name][tool.name] = Tool(spec=tool, required_tenant=schema.require_tenant)

    def register_tool(self, toolset, tool, *, tenant_id=None):
        if isinstance(toolset, ToolsetSchema):
            schema = toolset
            toolset = schema.name
            self._toolsets[schema.name] = schema
            self._tools.setdefault(schema.name, {})
            self._tenant_tools.setdefault(schema.name, {})
            for tool in schema.tools:
                self._tools[schema.name][tool.name] = Tool(spec=tool, required_tenant=schema.require_tenant)
            return
        schema = self._toolsets.get(toolset)
        require_tenant = schema.require_tenant if schema is not None else tool.required_tenant
        tool_obj = Tool(spec=tool.spec, callable_=tool.callable_, metadata=tool.metadata, required_tenant=tool.required_tenant or require_tenant)
        if schema is None:
            schema = ToolsetSchema(name=toolset, description="", require_tenant=require_tenant)
            self._toolsets[toolset] = schema
        self._tools.setdefault(toolset, {})
        self._tools[toolset][tool.spec.name] = tool_obj
        # Keep the schema's tool list in sync so get_toolset_schema/export_schema reflect registered tools.
        self._toolsets[toolset] = replace(schema, tools=tuple(t.spec for t in self._tools[toolset].values()))
        if tenant_id:
            self._tenant_tools.setdefault(toolset, {}).setdefault(tenant_id, {})[tool.spec.name] = tool_obj

    def get_toolset(self, name):
        return self._toolsets.get(name)

    def get_tool(self, toolset: str, name: str, tenant_id: Optional[str] = None):
        if tenant_id:
            tenant_tools = self._tenant_tools.get(toolset, {}).get(tenant_id)
            if tenant_tools is None:
                raise ValueError(f"tenant_id='{tenant_id}' has no mapped tools for toolset='{toolset}'")
            tool = tenant_tools.get(name)
            if tool is not None:
                return tool
        tools = self._tools.get(toolset)
        if not tools:
            return None
        return tools.get(name)

    def toolset_names(self) -> Sequence[str]:
        return tuple(self._toolsets.keys())

    def tool_names(self, toolset: str) -> Sequence[str]:
        return tuple(self._tools.get(toolset, {}).keys())

    def export_schema(self, toolset: str) -> Dict[str, Any]:
        schema = self._toolsets.get(toolset)
        if schema is None:
            raise KeyError(f"unknown toolset: {toolset}")
        return schema.to_json()

    async def lookup(self, *, tenant_id: Optional[str], toolset: str) -> Sequence[Tool]:
        schema = self._toolsets.get(toolset)
        if schema is None:
            return ()
        effective_tenant = schema.tenant_for(caller_tenant_id=tenant_id)
        if effective_tenant:
            tools = self._tenant_tools.get(toolset, {}).get(effective_tenant)
            if tools:
                return tuple(tools.values())
        return tuple(self._tools.get(toolset, {}).values())

_EchoProvider

Proveedor offline: devuelve echo:<prompt> sin red.

Source code in src/ciel/cli/main.py

class _EchoProvider:
    """Proveedor offline: devuelve ``echo:<prompt>`` sin red."""

    provider_name = "echo"

    async def complete(self, request: ChatRequest) -> object:
        from ciel.runtime import ChatChoice, ChatResponse

        prompt = request.messages[-1].content if request.messages else ""
        return ChatResponse(
            choice=ChatChoice(
                message=ChatMessage(role="assistant", content=f"echo:{prompt}"),
                finish_reason="stop",
            ),
            metadata={},
        )

    async def stream(self, request: ChatRequest) -> tuple:  # pragma: no cover - parity
        return (await self.complete(request),)

    async def models(self):  # pragma: no cover
        from ciel.providers import ModelInfo

        return [ModelInfo(id="echo", provider=self.provider_name)]

_build_runtime()

Construye un runtime para la CLI.

Usa el proveedor remoto si CIEL_PROVIDER_URL está configurado, o un proveedor echo offline en caso contrario (para smoke tests sin red).

Source code in src/ciel/cli/main.py

def _build_runtime():
    """Construye un runtime para la CLI.

    Usa el proveedor remoto si ``CIEL_PROVIDER_URL`` está configurado, o un
    proveedor ``echo`` offline en caso contrario (para smoke tests sin red).
    """
    from ciel.providers import OpenAICompatibleProvider

    provider_url = os.getenv("CIEL_PROVIDER_URL")
    if provider_url:
        provider: object = OpenAICompatibleProvider(
            base_url=provider_url,
            api_key=os.getenv("CIEL_API_KEY"),
            default_model=os.getenv("CIEL_MODEL"),
        )
        console.print(f"[dim]chat: usando proveedor remoto {provider_url}[/]")
    else:
        provider = _EchoProvider()
        console.print("[dim]chat: sin CIEL_PROVIDER_URL; usando proveedor echo offline[/]")

    registry = ToolRegistry(default_toolset="default")
    dispatcher = DefaultToolDispatcher(
        provider=ToolProvider(registry=registry, require_tenant_on_execution=False),
        default_toolset="default",
    )
    return DefaultAgentRuntime(provider=provider, dispatcher=dispatcher)

_callback(ctx: typer.Context, verbose: bool = typer.Option(False, '--verbose', '-v', help='Log extended')) -> None

Source code in src/ciel/cli/main.py

@app.callback()
def _callback(
    ctx: typer.Context,
    verbose: bool = typer.Option(False, "--verbose", "-v", help="Log extended"),
) -> None:
    ctx.obj = {"verbose": verbose}

_load_board_group()

Source code in src/ciel/cli/main.py

def _load_board_group():
    from ciel.cli.board import board_app as group
    return group

_load_chat_group()

Source code in src/ciel/cli/main.py

def _load_chat_group():
    from ciel.cli.chat import chat_app as group
    return group

_load_cost_group()

Source code in src/ciel/cli/main.py

def _load_cost_group():
    from ciel.cli.cost import cost_app as group
    return group

_load_eval_group()

Source code in src/ciel/cli/main.py

def _load_eval_group():
    from ciel.cli.evaluate import evaluate_app as group
    return group

_load_flow_group()

Source code in src/ciel/cli/main.py

def _load_flow_group():
    from ciel.cli.flow import flow_app as group
    return group

_load_graph_group()

Source code in src/ciel/cli/main.py

def _load_graph_group():
    from ciel.cli.graph import graph_app as group
    return group

_load_loop_group()

Source code in src/ciel/cli/main.py

def _load_loop_group():
    from ciel.cli.loop import loop_app as group
    return group

_load_rbac_group()

Source code in src/ciel/cli/main.py

def _load_rbac_group():
    from ciel.cli.rbac import rbac_app as group
    return group

_load_root_group()

Source code in src/ciel/cli/main.py

def _load_root_group():
    from ciel.cli.root import root_app as group
    return group

_load_skills_group()

Source code in src/ciel/cli/main.py

def _load_skills_group():
    from ciel.cli.skills_cli import skills_app as group
    return group

_load_studio_group()

Source code in src/ciel/cli/main.py

def _load_studio_group():
    from ciel.cli.studio_cli import studio_app as group
    return group

_load_swarm_group()

Source code in src/ciel/cli/main.py

def _load_swarm_group():
    from ciel.cli.swarm import swarm_app as group
    return group

_ok(value: str) -> str

Source code in src/ciel/cli/main.py

def _ok(value: str) -> str:
    return f"[green]{value}[/]"

_run_stream(runtime, request: ChatRequest, tenant: Optional[str]) -> None

Imprime tokens en tiempo real usando runtime.stream_tokens.

stream_tokens re-emite el contenido creciente del assistant, así que aquí solo se imprime el delta respecto al fragmento anterior para evitar duplicados. Si no hay fragmentos, se imprime un aviso.

Source code in src/ciel/cli/main.py

def _run_stream(runtime, request: ChatRequest, tenant: Optional[str]) -> None:
    """Imprime tokens en tiempo real usando ``runtime.stream_tokens``.

    ``stream_tokens`` re-emite el contenido *creciente* del assistant, así que
    aquí solo se imprime el delta respecto al fragmento anterior para evitar
    duplicados. Si no hay fragmentos, se imprime un aviso.
    """
    import anyio

    async def _consume() -> str:
        prior = ""
        last = ""
        async for token in runtime.stream_tokens(request=request, tenant_id=tenant):
            last = token
            delta = token[len(prior):]
            if delta:
                console.print(delta, end="")
            prior = token
        return last

    full = anyio.run(_consume)
    console.print()  # salto de línea final
    if not full:
        console.print("[yellow](sin contenido del proveedor)[/]")

chat(message: str = typer.Argument(..., help='Prompt a enviar al agente'), app_name: str = typer.Option('default', '--app', '-a', help='App name to chat with'), quiet: bool = typer.Option(False, '-q', '--quiet', help='Minimal output mode'), stream: bool = typer.Option(False, '--stream', help='Streaming en tiempo real (SSE por consola)'), tenant: Optional[str] = typer.Option(None, '--tenant', help='Tenant id (default $CIEL_TENANT)')) -> None

Envía un prompt al runtime y muestra la respuesta.

Con --stream se imprimen los fragmentos incrementalmente en tiempo real. Si el proveedor no implementa streaming, hace fallback a un único chunk con el texto completo.

Source code in src/ciel/cli/main.py

@app.command("chat")
def chat(
    message: str = typer.Argument(..., help="Prompt a enviar al agente"),
    app_name: str = typer.Option("default", "--app", "-a", help="App name to chat with"),
    quiet: bool = typer.Option(False, "-q", "--quiet", help="Minimal output mode"),
    stream: bool = typer.Option(False, "--stream", help="Streaming en tiempo real (SSE por consola)"),
    tenant: Optional[str] = typer.Option(None, "--tenant", help="Tenant id (default $CIEL_TENANT)"),
) -> None:
    """Envía un prompt al runtime y muestra la respuesta.

    Con ``--stream`` se imprimen los fragmentos incrementalmente en tiempo real.
    Si el proveedor no implementa streaming, hace fallback a un único chunk con
    el texto completo.
    """
    effective_tenant = tenant or os.getenv("CIEL_TENANT")
    runtime = _build_runtime()
    request = ChatRequest(
        messages=(ChatMessage(role="user", content=message),),
        tools=(),
    )

    if not quiet:
        console.print(f"[yellow]Chat with app:[/] {app_name}")

    if stream:
        _run_stream(runtime, request, effective_tenant)
        return

    result = asyncio.run(
        runtime.run_agent_loop(request=request, tenant_id=effective_tenant)
    )
    text = getattr(getattr(result.response, "choice", None), "message", None)
    text = getattr(text, "content", "") or ""
    if quiet:
        console.print(text)
    else:
        console.print(Panel.fit(text, title="Respuesta", border_style="blue"))

checkpoints(session_root: Path = typer.Option(..., '--session-root', exists=True, file_okay=False, help='Directory containing session state')) -> None

Source code in src/ciel/cli/main.py

@app.command("checkpoints")
def checkpoints(
    session_root: Path = typer.Option(..., "--session-root", exists=True, file_okay=False, help="Directory containing session state"),
) -> None:
    table = Table(title="Checkpoints")
    table.add_column("checkpoint_id")
    table.add_column("path")
    table.add_column("status")
    table.add_row("cp-demo", str(session_root / "cp-demo.json"), "pending inspection")
    console.print(table)

compression(target_dir: Path = typer.Option(..., '--target-dir', exists=True, file_okay=False, help='Project root for context loading'), max_chars: int = typer.Option(20000, '--max-chars', help='Context render budget in characters')) -> None

Source code in src/ciel/cli/main.py

@app.command("compression")
def compression(
    target_dir: Path = typer.Option(..., "--target-dir", exists=True, file_okay=False, help="Project root for context loading"),
    max_chars: int = typer.Option(20000, "--max-chars", help="Context render budget in characters"),
) -> None:
    from ciel.runtime.context import load_project_context
    from ciel.runtime.context_compression import compress_context
    from ciel.runtime import ChatMessage

    context = load_project_context(path=str(target_dir))
    rendered = context.render(max_chars=max_chars) if context.files else ""
    placeholder_message = ChatMessage(role="system", content=rendered)
    summarized, slice_ = compress_context([placeholder_message], max_chars=max_chars)
    table = Table(title="Context compression summary")
    table.add_column("limit")
    table.add_column("removed")
    table.add_column("kept")
    table.add_row(str(max_chars), str(slice_.removed), str(slice_.keep_head))
    console.print(table)
    for message in summarized:
        console.print(message.text())

doctor() -> None

Source code in src/ciel/cli/main.py

@app.command("doctor")
def doctor() -> None:
    table = Table(title="Environment")
    table.add_column("Item")
    table.add_column("Status")
    table.add_row("CLI version", _ok(f"{__version__}"))
    table.add_row("Python", _ok(f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"))
    table.add_row("Repo root", _ok(str(REPO_ROOT)))
    console.print(table)

info() -> None

Source code in src/ciel/cli/main.py

@app.command("info")
def info() -> None:
    panel = Panel.fit(
        f"[bold]Ciel[/bold] {__version__}\nRepo: {REPO_ROOT}",
        title="Framework info",
        border_style="blue",
    )
    console.print(panel)

init(path: Path = typer.Argument(Path('.'), help='Target directory (created if missing)'), force: bool = typer.Option(False, '--force', help='Overwrite existing files')) -> None

Scaffold a new Ciel project (pyproject + agent + ciel.yaml). Offline-safe.

Source code in src/ciel/cli/main.py

@app.command("init")
def init(
    path: Path = typer.Argument(Path("."), help="Target directory (created if missing)"),
    force: bool = typer.Option(False, "--force", help="Overwrite existing files"),
) -> None:
    """Scaffold a new Ciel project (pyproject + agent + ciel.yaml). Offline-safe."""
    from ciel.cli.scaffold import scaffold_project

    target = Path(path)
    created = scaffold_project(target, force=force)
    console.print(f"[green]Initialized Ciel project at[/] {target}")
    for rel in created:
        console.print(f"  [dim]+ {rel}[/]")

observe(otel_endpoint: str | None = typer.Option(None, '--otel-endpoint', help='OTLP collector endpoint')) -> None

Initialize a tracing session and report OpenTelemetry observability status.

Useful as a smoke check of the observability stack: prints whether OTel is available, the active exporter, and the number of spans emitted so far.

Source code in src/ciel/cli/main.py

@app.command("observe")
def observe(
    otel_endpoint: str | None = typer.Option(None, "--otel-endpoint", help="OTLP collector endpoint"),
) -> None:
    """Initialize a tracing session and report OpenTelemetry observability status.

    Useful as a smoke check of the observability stack: prints whether OTel is
    available, the active exporter, and the number of spans emitted so far.
    """
    from ciel.observability.otel import (
        OTEL_AVAILABLE,
        init_tracing,
        span_count,
    )

    if not OTEL_AVAILABLE:
        console.print(
            "[red]OpenTelemetry not available.[/] Install with: uv pip install 'ciel[observability]'"
        )
        raise typer.Exit(code=1)
    provider = init_tracing(service_name="ciel", otlp_endpoint=otel_endpoint)
    exporter_kind = "OTLP" if otel_endpoint else "in-memory"
    console.print(
        Panel.fit(
            f"[bold]Ciel observability[/bold]\n"
            f"OpenTelemetry: available\n"
            f"exporter: {exporter_kind}\n"
            f"spans emitted (this process): {span_count()}\n"
            f"[dim]Wire a collector (Tempo/Jaeger/OTel Collector) via --otel-endpoint in `ciel serve`.[/]",
            title="observe",
            border_style="blue",
        )
    )

run(agent: str = typer.Option(..., '--agent', '-a', help='Agent module or path'), input_file: Path | None = typer.Option(None, '--input', '-i', help='Input file')) -> None

Source code in src/ciel/cli/main.py

@app.command("run")
def run(
    agent: str = typer.Option(..., "--agent", "-a", help="Agent module or path"),
    input_file: Path | None = typer.Option(None, "--input", "-i", help="Input file"),
) -> None:
    console.print(f"[yellow]Running agent:[/] {agent}")
    if input_file:
        console.print(f"[yellow]Input:[/] {input_file}")
    console.print("[yellow]Not implemented yet.[/]")

serve(host: str = typer.Option('0.0.0.0', '--host', help='Bind host'), port: int = typer.Option(8080, '--port', '-p', help='Bind port'), tenant: str | None = typer.Option(None, '--tenant', help='Default tenant id (else $CIEL_TENANT)'), otel: bool = typer.Option(False, '--otel', help='Enable OpenTelemetry tracing (in-memory or OTLP endpoint)'), otel_endpoint: str | None = typer.Option(None, '--otel-endpoint', help='OTLP collector endpoint (else in-memory exporter)')) -> None

Run the composed Ciel gateway (control + MCP host + webhook) via uvicorn.

Source code in src/ciel/cli/main.py

@app.command("serve")
def serve(
    host: str = typer.Option("0.0.0.0", "--host", help="Bind host"),
    port: int = typer.Option(8080, "--port", "-p", help="Bind port"),
    tenant: str | None = typer.Option(None, "--tenant", help="Default tenant id (else $CIEL_TENANT)"),
    otel: bool = typer.Option(False, "--otel", help="Enable OpenTelemetry tracing (in-memory or OTLP endpoint)"),
    otel_endpoint: str | None = typer.Option(None, "--otel-endpoint", help="OTLP collector endpoint (else in-memory exporter)"),
) -> None:
    """Run the composed Ciel gateway (control + MCP host + webhook) via uvicorn."""
    import uvicorn

    from ciel.gateway.server import make_app
    from ciel.observability.otel import OTEL_AVAILABLE, init_tracing

    effective_tenant = tenant or os.getenv("CIEL_TENANT")
    if otel:
        init_tracing(service_name="ciel", otlp_endpoint=otel_endpoint)
        otel_status = (
            f"OTLP {otel_endpoint}" if otel_endpoint else "in-memory exporter"
        ) if OTEL_AVAILABLE else "UNAVAILABLE (install ciel[observability])"
    else:
        otel_status = "disabled"
    app = make_app(tenant_id=effective_tenant)
    console.print(
        Panel.fit(
            f"[bold]Ciel[/bold] {__version__} — gateway\n"
            f"host={host} port={port} default_tenant={effective_tenant or '(none — tenant required per request)'}\n"
            f"tracing={otel_status}\n"
            f"[dim]surfaces: control / , MCP /mcp , webhook /v1/messaging/webhook , Teams/Discord/WebUI[/]",
            title="serve",
            border_style="blue",
        )
    )
    uvicorn.run(app, host=host, port=port)

version() -> None

Source code in src/ciel/cli/main.py

@app.command("version")
def version() -> None:
    console.print(f"Ciel {__version__}")

_ = (tempfile, Path) module-attribute

console = Console() module-attribute

root_app = typer.Typer(name='root', help='Run the root agent (ADK sub_agents, offline-safe)') 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,
        )

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)

_build_demo_agent() -> RootAgent

Root agent de DEMOSTRACIÓN EN MEMORIA con 2 specialists + root handler.

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

def _build_demo_agent() -> RootAgent:
    """Root agent de DEMOSTRACIÓN EN MEMORIA con 2 specialists + root handler."""
    return (
        RootAgent(name="root")
        .add_specialist(Specialist("db", _db_handler, "consultas a base de datos"))
        .add_specialist(Specialist("net", _net_handler, "operaciones de red"))
        .set_router(_demo_router)
        .set_root_handler(_root_handler)
    )

_db_handler(state: RootState) -> str

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

def _db_handler(state: RootState) -> str:
    prior = len(state.history)
    return f"db-handled: {state.prompt} (turnos previos en session: {prior})"

_demo_router(prompt: str)

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

def _demo_router(prompt: str):
    p = prompt.lower()
    if any(k in p for k in ("sql", "base de datos", "tabla", "select", "insert", "update", "delete")):
        return "db"
    if any(k in p for k in ("http", "red", "fetch", "url", "api")):
        return "net"
    return None

_net_handler(state: RootState) -> str

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

def _net_handler(state: RootState) -> str:
    prior = len(state.history)
    return f"net-handled: {state.prompt} (turnos previos en session: {prior})"

_resolve_session_store(db: Optional[str], tenant: Optional[str])

Resuelve un SessionStore sobre MemoryStore (o None si no hay --db).

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

def _resolve_session_store(db: Optional[str], tenant: Optional[str]):
    """Resuelve un SessionStore sobre MemoryStore (o None si no hay --db)."""
    if not db:
        return None
    memory = MemoryStore(db)
    return SessionStore(memory)

_root_handler(state: RootState) -> str

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

def _root_handler(state: RootState) -> str:
    prior = len(state.history)
    return f"root-handled: {state.prompt} (turnos previos en session: {prior})"

route(prompt: str = typer.Argument(..., help='Petición a enrutar por el root agent'), db: Optional[str] = typer.Option(None, '--db', help='Ruta del MemoryStore SQLite para session state persistente'), session_id: Optional[str] = typer.Option(None, '--session-id', help='Id de session (mantiene historial entre turnos)'), tenant: Optional[str] = typer.Option(None, '--tenant', help='Tenant id (multitenancy nativo)')) -> None

Route a prompt through the demo root agent (offline, ADK sub_agents).

Con --db + --session-id el agente recuerda turnos previos entre invocaciones (session state persistente por tenant, estilo ADK). Sin red ni proveedor.

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

@root_app.command("route")
def route(
    prompt: str = typer.Argument(..., help="Petición a enrutar por el root agent"),
    db: Optional[str] = typer.Option(
        None, "--db", help="Ruta del MemoryStore SQLite para session state persistente"
    ),
    session_id: Optional[str] = typer.Option(
        None, "--session-id", help="Id de session (mantiene historial entre turnos)"
    ),
    tenant: Optional[str] = typer.Option(
        None, "--tenant", help="Tenant id (multitenancy nativo)"
    ),
) -> None:
    """Route a prompt through the demo root agent (offline, ADK sub_agents).

    Con ``--db`` + ``--session-id`` el agente recuerda turnos previos entre
    invocaciones (session state persistente por tenant, estilo ADK). Sin red ni
    proveedor.
    """
    effective_tenant = tenant or os.getenv("CIEL_TENANT")
    agent = _build_demo_agent()
    runner: RootRunner = agent.compile()

    session_store = _resolve_session_store(db, effective_tenant)
    sid = session_id or ("demo-session" if session_store is not None else None)

    try:
        state = asyncio.run(
            runner.route(
                prompt,
                session_id=sid,
                session_store=session_store,
                tenant_id=effective_tenant,
            )
        )
    except KeyboardInterrupt:
        raise typer.Exit(0)

    console.print(
        Panel.fit(
            f"prompt: {state.prompt}\n"
            f"route: {state.route or '(root handler)'}\n"
            f"handled_by_root: {state.handled_by_root}\n"
            f"result: {state.result}\n"
            f"turnos en session: {len(state.history)}",
            title="Root agent (offline demo)",
            border_style="blue",
        )
    )
    if sid:
        console.print(
            f"[dim]session_id={sid} tenant={effective_tenant or '(none)'}"
            f" (session state {'persistido en ' + db if session_store else 'en memoria'})[/dim]"
        )
    if not state.route and not state.handled_by_root:
        raise typer.Exit(1)

__all__ = ['chat_app', 'group'] module-attribute

chat_app = typer.Typer(name='chat', help='Run group chats (AutoGen-style, offline-safe)') module-attribute

console = Console() 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))

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

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"),
        )

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)

_build_demo_chat(max_rounds: int) -> GroupChat

Group chat de DEMOSTRACION EN MEMORIA con 3 agentes locales.

No usa red ni proveedor; cada responder opera sobre state.transcript (OFFLINE-SAFE). El revisor emite TERMINATE cuando ya existe propuesta, implementacion y pruebas -> el chat converge deterministicamente.

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

def _build_demo_chat(max_rounds: int) -> GroupChat:
    """Group chat de DEMOSTRACION EN MEMORIA con 3 agentes locales.

    No usa red ni proveedor; cada ``responder`` opera sobre ``state.transcript``
    (OFFLINE-SAFE). El revisor emite ``TERMINATE`` cuando ya existe propuesta,
    implementacion y pruebas -> el chat converge deterministicamente.
    """
    participants = [
        Agent(name="reviewer", responder=_reviewer, system_message="Revisor del diseno"),
        Agent(name="coder", responder=_coder, system_message="Implementador"),
        Agent(name="tester", responder=_tester, system_message="Validador"),
    ]
    return GroupChat(
        participants,
        max_rounds=max_rounds,
        terminate_keyword="TERMINATE",
    )

_coder(state: GroupChatState) -> str

Codificador: reacciona a la propuesta/ultimo mensaje.

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

def _coder(state: GroupChatState) -> str:
    """Codificador: reacciona a la propuesta/ultimo mensaje."""
    last = state.transcript[-1].content if state.transcript else ""
    return f"Implemento: {last}"

_print_transcript(state: GroupChatState, title: str = 'Group chat') -> None

Imprime con Rich el transcripto (role/content/round) y un Panel resumen.

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

def _print_transcript(state: GroupChatState, title: str = "Group chat") -> None:
    """Imprime con Rich el transcripto (role/content/round) y un Panel resumen."""
    table = Table(title=title)
    table.add_column("round")
    table.add_column("role")
    table.add_column("content", overflow="fold")
    for msg in state.transcript:
        table.add_row(str(msg.round), msg.role, msg.content)
    console.print(table)
    console.print(
        Panel.fit(
            f"rounds: {state.rounds}\n"
            f"terminated: {state.terminated}\n"
            f"terminator: {state.terminator or '(none)'}",
            title="Summary",
            border_style="blue",
        )
    )

_reviewer(state: GroupChatState) -> str

Revisor: propone en su primera ronda; cuando coder y tester ya reaccionaron a la propuesta, aprueba y emite TERMINATE (converge).

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

def _reviewer(state: GroupChatState) -> str:
    """Revisor: propone en su primera ronda; cuando coder y tester ya
    reaccionaron a la propuesta, aprueba y emite TERMINATE (converge)."""
    roles = [m.role for m in state.transcript]
    has_proposal = any("Propuesta:" in m.content for m in state.transcript)
    has_reviewer = "reviewer" in roles
    if not has_reviewer:
        return "Propuesta: disenar feature de login con JWT y refresh tokens."
    if has_proposal and "coder" in roles and "tester" in roles:
        return "Aprobado: propuesta + implementacion + pruebas correctas. TERMINATE"
    return "ok, continuemos revisando."

_tester(state: GroupChatState) -> str

Tester: valida la implementacion/ultimo mensaje.

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

def _tester(state: GroupChatState) -> str:
    """Tester: valida la implementacion/ultimo mensaje."""
    last = state.transcript[-1].content if state.transcript else ""
    return f"Testeo: {last}"

group(message: str = typer.Option('Tarea: disenar feature de login', '--message', '-m', help='Mensaje inicial del usuario que dispara el chat'), rounds: int = typer.Option(12, '--rounds', '-r', help='Numero maximo de rondas antes de forzar fin'), tenant: Optional[str] = typer.Option(None, '--tenant', help='Tenant id (para trazabilidad)')) -> None

Run an offline 3-agent group chat that converges with TERMINATE.

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

@chat_app.command("group")
def group(
    message: str = typer.Option(
        "Tarea: disenar feature de login",
        "--message",
        "-m",
        help="Mensaje inicial del usuario que dispara el chat",
    ),
    rounds: int = typer.Option(
        12, "--rounds", "-r", help="Numero maximo de rondas antes de forzar fin"
    ),
    tenant: Optional[str] = typer.Option(
        None, "--tenant", help="Tenant id (para trazabilidad)"
    ),
) -> None:
    """Run an offline 3-agent group chat that converges with TERMINATE."""
    chat = _build_demo_chat(max_rounds=rounds)
    manager = GroupChatManager(chat, supervisor=Supervisor(), tenant_id=tenant)

    async def _run() -> GroupChatState:
        return await manager.run(initial_message=message, initial_sender="user")

    try:
        state = asyncio.run(_run())
    except KeyboardInterrupt:
        raise typer.Exit(0)

    _print_transcript(state, title="Group chat (offline demo)")
    console.print(
        Panel.fit(
            "Offline demo: reviewer -> coder -> tester (round-robin).\n"
            "El revisor emite TERMINATE cuando hay propuesta + implementacion + pruebas.\n"
            "No provider, no network required.",
            title="Summary",
            border_style="blue",
        )
    )
    if not state.terminated:
        console.print("[yellow]chat finished without explicit TERMINATE (max rounds).[/]")

DEFAULT_DB_NAME = 'ciel_loop.sqlite3' module-attribute

__all__ = ['loop_app', 'run', 'resume'] module-attribute

console = Console() module-attribute

loop_app = typer.Typer(name='loop', help='Autonomous event-loop agent (offline-safe)') module-attribute

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

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)

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()

_demo_handler(task: Task)

Handler OFFLINE: devuelve echo del objetivo, sin red ni proveedor.

Source code in src/ciel/cli/loop.py

def _demo_handler(task: Task):
    """Handler OFFLINE: devuelve echo del objetivo, sin red ni proveedor."""

    def _run(t: Task):
        return {"echo": t.goal, "payload": dict(t.payload)}

    return _run

_print_task(task: Task, title: str = 'Task') -> None

Source code in src/ciel/cli/loop.py

def _print_task(task: Task, title: str = "Task") -> None:
    table = Table(title=title)
    table.add_column("id")
    table.add_column("goal")
    table.add_column("status")
    table.add_column("attempts")
    table.add_column("result")
    table.add_column("error")
    table.add_row(
        task.id[:8],
        task.goal,
        task.status,
        str(task.attempts),
        repr(task.result),
        task.error or "(none)",
    )
    console.print(table)

resolve_db_path(db_flag: Optional[Path]) -> str

Resuelve la ruta del loop SQLite.

Prioridad: --db > variable de entorno CIEL_LOOP_DB > archivo por defecto en el directorio actual.

Source code in src/ciel/cli/loop.py

def resolve_db_path(db_flag: Optional[Path]) -> str:
    """Resuelve la ruta del loop SQLite.

    Prioridad: ``--db`` > variable de entorno ``CIEL_LOOP_DB`` > archivo por
    defecto en el directorio actual.
    """
    if db_flag is not None:
        return str(db_flag)
    env = os.environ.get("CIEL_LOOP_DB")
    if env:
        return env
    return str(Path.cwd() / DEFAULT_DB_NAME)

resume(run_id: str = typer.Option(..., '--run-id', help='Run id to resume'), db: Path = typer.Option(..., '--db', help='Same SQLite db used in run'), tenant: Optional[str] = typer.Option(None, '--tenant', help='Tenant id'), session: Optional[str] = typer.Option(None, '--session-id', help='Session id')) -> None

Resume an interrupted autonomous loop from its last checkpoint (requires --db).

Source code in src/ciel/cli/loop.py

@loop_app.command("resume")
def resume(
    run_id: str = typer.Option(..., "--run-id", help="Run id to resume"),
    db: Path = typer.Option(..., "--db", help="Same SQLite db used in run"),
    tenant: Optional[str] = typer.Option(None, "--tenant", help="Tenant id"),
    session: Optional[str] = typer.Option(None, "--session-id", help="Session id"),
) -> None:
    """Resume an interrupted autonomous loop from its last checkpoint (requires --db)."""
    db_path = resolve_db_path(db)
    memory = MemoryStore(db_path)
    checkpointer = EventLoopCheckpointStore(memory)

    # Reconstruye el handler offline (determinista por objetivo persistido).
    saved = checkpointer.load(run_id=run_id, tenant_id=tenant, session_id=session or run_id)
    if saved is None:
        console.print(f"[red]no checkpoint found for run_id '{run_id}'[/]")
        memory.close()
        raise typer.Exit(1)
    task = Task.from_snapshot(saved["task"])
    loop = EventLoop(
        supervisor=Supervisor(),
        checkpointer=checkpointer,
        tenant_id=tenant,
        session_id=session or run_id,
        max_attempts=5,
    )

    async def _resume() -> Task:
        return await loop.resume(run_id=run_id, handler=_demo_handler(task))

    try:
        out = asyncio.run(_resume())
    except Exception as exc:
        console.print(f"[red]resume failed:[/] {exc}")
        raise typer.Exit(1)
    finally:
        memory.close()

    _print_task(out, title="Resumed task")
    console.print(
        Panel.fit(
            f"run_id: {run_id}\ncheckpointer: {db_path}\ntenant: {tenant or '(none)'}",
            title="Resume summary",
            border_style="blue",
        )
    )

run(goal: str = typer.Argument(..., help='Goal for the autonomous agent'), run_id: Optional[str] = typer.Option(None, '--run-id', help='Explicit run id (for later resume)'), db: Optional[Path] = typer.Option(None, '--db', help='SQLite db for the checkpointer (or CIEL_LOOP_DB)'), tenant: Optional[str] = typer.Option(None, '--tenant', help='Tenant id'), session: Optional[str] = typer.Option(None, '--session-id', help='Session id (for session state)')) -> None

Run an autonomous agent over a goal offline (no network, no provider).

With --db the run is persisted via a checkpointer so it can be resumed later with the resume command (e.g. after a process restart).

Source code in src/ciel/cli/loop.py

@loop_app.command("run")
def run(
    goal: str = typer.Argument(..., help="Goal for the autonomous agent"),
    run_id: Optional[str] = typer.Option(None, "--run-id", help="Explicit run id (for later resume)"),
    db: Optional[Path] = typer.Option(None, "--db", help="SQLite db for the checkpointer (or CIEL_LOOP_DB)"),
    tenant: Optional[str] = typer.Option(None, "--tenant", help="Tenant id"),
    session: Optional[str] = typer.Option(None, "--session-id", help="Session id (for session state)"),
) -> None:
    """Run an autonomous agent over a goal offline (no network, no provider).

    With --db the run is persisted via a checkpointer so it can be resumed
    later with the `resume` command (e.g. after a process restart).
    """
    checkpointer = None
    memory = None
    db_path = None
    if db is not None:
        db_path = resolve_db_path(db)
        memory = MemoryStore(db_path)
        checkpointer = EventLoopCheckpointStore(memory)

    task = Task(goal=goal)
    loop = EventLoop(
        supervisor=Supervisor(),
        checkpointer=checkpointer,
        tenant_id=tenant,
        session_id=session or run_id,
        max_attempts=5,
    )

    async def _run() -> Task:
        return await loop.run(task, _demo_handler(task), run_id=run_id)

    try:
        out = asyncio.run(_run())
    except Exception as exc:
        console.print(f"[red]loop failed:[/] {exc}")
        raise typer.Exit(1)
    finally:
        if memory is not None:
            memory.close()

    _print_task(out, title="Autonomous task")
    if db_path is not None:
        console.print(
            Panel.fit(
                f"run_id: {loop.run_id}\ncheckpointer: {db_path}\n"
                f"tenant: {tenant or '(none)'}\nsession: {session or loop.run_id}\n\n"
                "Resume after a restart with: ciel loop resume --run-id <run_id> --db <db> [--tenant <tenant>]",
                title="Run summary",
                border_style="blue",
            )
        )
    else:
        console.print(
            Panel.fit(
                "Offline demo (no --db): a single autonomous task runs with a local\n"
                "echo handler. No provider, no network required. Pass --db to enable resume.",
                title="Summary",
                border_style="blue",
            )
        )

DEFAULT_DB_NAME = 'ciel_graph.sqlite3' module-attribute

__all__ = ['graph_app', 'demo', 'run', 'resume'] module-attribute

console = Console() module-attribute

graph_app = typer.Typer(name='graph', help='Build and run explicit state graphs (offline-safe)') module-attribute

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)

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

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,
        )

_build_demo_graph() -> StateGraph

Grafo de DEMOSTRACIÓN EN MEMORIA: entry -> plan -> execute -> finish.

No usa red ni proveedor; cada nodo escribe en state.data y devuelve un payload. Pensado para smoke tests offline (igual que ciel chat con _EchoProvider).

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

def _build_demo_graph() -> StateGraph:
    """Grafo de DEMOSTRACIÓN EN MEMORIA: entry -> plan -> execute -> finish.

    No usa red ni proveedor; cada nodo escribe en ``state.data`` y devuelve un
    payload. Pensado para smoke tests offline (igual que ``ciel chat`` con
    ``_EchoProvider``).
    """
    graph = StateGraph(name="demo")

    async def entry(state_data: dict) -> dict:
        state_data.setdefault("started", True)
        return {"node": "entry"}

    async def plan(state_data: dict) -> dict:
        state_data["plan"] = "Define the task and decompose it into steps."
        return {"node": "plan", "plan": state_data["plan"]}

    async def execute(state_data: dict) -> dict:
        state_data["result"] = "Executed the plan and produced the result."
        return {"node": "execute", "result": state_data["result"]}

    async def finish(state_data: dict) -> dict:
        state_data["done"] = True
        return {"node": "finish"}

    graph.add_node("entry", entry)
    graph.add_node("plan", plan)
    graph.add_node("execute", execute)
    graph.add_node("finish", finish)
    graph.add_edge("entry", "plan")
    graph.add_edge("plan", "execute")
    graph.add_edge("execute", "finish")
    graph.set_entry_point("entry")
    graph.set_finish_point("finish")
    return graph

_build_spec_graph(spec: AgentSpec) -> StateGraph

Construye un grafo pipeline lineal a partir de un AgentSpec.

entry = steps[0].id, aristas en orden, finish = último paso. Cada nodo guarda en state.data[f"__out__{step.id}"] un dict con el id/kind/prompt del paso.

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

def _build_spec_graph(spec: AgentSpec) -> StateGraph:
    """Construye un grafo pipeline lineal a partir de un AgentSpec.

    ``entry`` = ``steps[0].id``, aristas en orden, ``finish`` = último paso.
    Cada nodo guarda en ``state.data[f"__out__{step.id}"]`` un dict con el
    id/kind/prompt del paso.
    """
    graph = StateGraph(name=spec.name or "spec")
    steps = list(spec.steps)
    if not steps:
        raise typer.BadParameter("AgentSpec must have at least one step")

    def _make_fn(step):
        async def _fn(state_data: dict) -> dict:
            payload = {"id": step.id, "kind": step.kind, "prompt": step.prompt}
            state_data[f"__out__{step.id}"] = payload
            return payload

        return _fn

    for step in steps:
        graph.add_node(step.id, _make_fn(step))
    for i in range(len(steps) - 1):
        graph.add_edge(steps[i].id, steps[i + 1].id)
    graph.set_entry_point(steps[0].id)
    graph.set_finish_point(steps[-1].id)
    return graph

_print_state(state: GraphState, title: str = 'Graph') -> None

Imprime con Rich el estado resultante del grafo.

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

def _print_state(state: GraphState, title: str = "Graph") -> None:
    """Imprime con Rich el estado resultante del grafo."""
    visited = " -> ".join(state.visited) if state.visited else "(none)"
    data_keys = ", ".join(state.data.keys()) if state.data else "(none)"
    table = Table(title=title)
    table.add_column("visited")
    table.add_column("data keys")
    table.add_column("current_node")
    table.add_column("last_output")
    table.add_row(
        visited,
        data_keys,
        str(state.current_node),
        repr(state.last_output),
    )
    console.print(table)
    console.print(
        Panel.fit(
            f"visited: {len(state.visited)} node(s)\n"
            f"data keys: {', '.join(state.data.keys()) or '(none)'}\n"
            f"current_node: {state.current_node}",
            title=title,
            border_style="blue",
        )
    )

demo() -> None

Run an in-memory demo graph offline (no network, no provider).

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

@graph_app.command("demo")
def demo() -> None:
    """Run an in-memory demo graph offline (no network, no provider)."""
    graph = _build_demo_graph()
    runner = graph.compile()  # Supervisor() por defecto, sin proveedor -> offline-safe

    async def _run() -> GraphState:
        return await runner.run(initial_data={})

    try:
        state = asyncio.run(_run())
    except KeyboardInterrupt:
        raise typer.Exit(0)

    _print_state(state, title="Demo graph")
    console.print(
        Panel.fit(
            "Offline demo: entry -> plan -> execute -> finish.\n"
            "No provider, no network required.",
            title="Summary",
            border_style="blue",
        )
    )

resolve_db_path(db_flag: Optional[Path]) -> str

Resuelve la ruta del grafo SQLite.

Prioridad: --db > variable de entorno CIEL_GRAPH_DB > archivo por defecto en el directorio actual.

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

def resolve_db_path(db_flag: Optional[Path]) -> str:
    """Resuelve la ruta del grafo SQLite.

    Prioridad: ``--db`` > variable de entorno ``CIEL_GRAPH_DB`` > archivo por
    defecto en el directorio actual.
    """
    if db_flag is not None:
        return str(db_flag)
    env = os.environ.get("CIEL_GRAPH_DB")
    if env:
        return env
    return str(Path.cwd() / DEFAULT_DB_NAME)

resume(run_id: str = typer.Option(..., '--run-id', help='Run id to resume'), db: Path = typer.Option(..., '--db', help='Same SQLite db used in run'), spec: Optional[Path] = typer.Option(None, '--spec', '-s', help='YAML AgentSpec used in run (else demo graph)'), tenant: Optional[str] = typer.Option(None, '--tenant', help='Tenant id')) -> None

Resume an interrupted graph from its last checkpoint.

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

@graph_app.command("resume")
def resume(
    run_id: str = typer.Option(..., "--run-id", help="Run id to resume"),
    db: Path = typer.Option(..., "--db", help="Same SQLite db used in run"),
    spec: Optional[Path] = typer.Option(
        None, "--spec", "-s", help="YAML AgentSpec used in run (else demo graph)"
    ),
    tenant: Optional[str] = typer.Option(None, "--tenant", help="Tenant id"),
) -> None:
    """Resume an interrupted graph from its last checkpoint."""
    db_path = resolve_db_path(db)
    memory = MemoryStore(db_path)
    checkpointer = GraphCheckpointStore(memory)

    if spec is not None:
        agent_spec = AgentSpec.from_yaml(Path(spec).read_text())
        graph = _build_spec_graph(agent_spec)
    else:
        graph = _build_demo_graph()

    runner = graph.compile(checkpointer=checkpointer, tenant_id=tenant)

    async def _resume() -> GraphState:
        return await runner.resume(run_id=run_id)

    try:
        state = asyncio.run(_resume())
    except Exception as exc:
        console.print(f"[red]resume failed:[/] {exc}")
        raise typer.Exit(1)
    finally:
        memory.close()

    _print_state(state, title="Resumed graph")

run(spec: Optional[Path] = typer.Option(None, '--spec', '-s', help='YAML AgentSpec (name, topology, steps[{id,kind,tool,prompt,depends_on}])'), run_id: Optional[str] = typer.Option(None, '--run-id', help='Explicit run id (for later resume)'), db: Optional[Path] = typer.Option(None, '--db', help='SQLite db for the checkpointer (or CIEL_GRAPH_DB)'), tenant: Optional[str] = typer.Option(None, '--tenant', help='Tenant id')) -> None

Run a graph. With --spec builds a linear pipeline; otherwise runs the demo.

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

@graph_app.command("run")
def run(
    spec: Optional[Path] = typer.Option(
        None, "--spec", "-s", help="YAML AgentSpec (name, topology, steps[{id,kind,tool,prompt,depends_on}])"
    ),
    run_id: Optional[str] = typer.Option(None, "--run-id", help="Explicit run id (for later resume)"),
    db: Optional[Path] = typer.Option(None, "--db", help="SQLite db for the checkpointer (or CIEL_GRAPH_DB)"),
    tenant: Optional[str] = typer.Option(None, "--tenant", help="Tenant id"),
) -> None:
    """Run a graph. With --spec builds a linear pipeline; otherwise runs the demo."""
    db_path = resolve_db_path(db)
    memory = MemoryStore(db_path)
    checkpointer = GraphCheckpointStore(memory)

    if spec is not None:
        agent_spec = AgentSpec.from_yaml(Path(spec).read_text())
        graph = _build_spec_graph(agent_spec)
        label = f"Spec graph: {agent_spec.name}"
    else:
        graph = _build_demo_graph()
        label = "Demo graph (no --spec)"

    runner = graph.compile(checkpointer=checkpointer, tenant_id=tenant)

    async def _run() -> GraphState:
        return await runner.run(initial_data={}, run_id=run_id)

    try:
        state = asyncio.run(_run())
    except KeyboardInterrupt:
        raise typer.Exit(0)
    finally:
        memory.close()

    _print_state(state, title=label)
    console.print(
        Panel.fit(
            f"run_id: {runner.run_id}\ncheckpointer: {db_path}\ntenant: {tenant or '(none)'}",
            title="Run summary",
            border_style="blue",
        )
    )

DEFAULT_DB_NAME = 'ciel_flow.sqlite3' module-attribute

__all__ = ['flow_app', 'run', 'resume'] module-attribute

console = Console() module-attribute

flow_app = typer.Typer(name='flow', help='Build and run event-driven flows (offline-safe)') 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

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"),
        )

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)

_build_demo_flow() -> Flow

Flow de DEMOSTRACIÓN EN MEMORIA (estilo CrewAI.Flows).

ingest -> transform -> router {a: branch_a, b: branch_b}

No usa red ni proveedor; cada paso escribe en state.data. Pensado para smoke tests offline.

Source code in src/ciel/cli/flow.py

def _build_demo_flow() -> Flow:
    """Flow de DEMOSTRACIÓN EN MEMORIA (estilo CrewAI.Flows).

    ingest -> transform -> router {a: branch_a, b: branch_b}

    No usa red ni proveedor; cada paso escribe en ``state.data``. Pensado para
    smoke tests offline.
    """
    flow = Flow(name="demo")

    def ingest(state: FlowState) -> str:
        state.data["items"] = [1, 2, 3]
        return "ingested"

    def transform(state: FlowState) -> str:
        state.data["count"] = len(state.data["items"])
        return "transformed"

    def decide(state: FlowState) -> str:
        return "a" if state.data["items"][0] == 1 else "b"

    def branch_a(state: FlowState) -> str:
        state.data["result"] = "A"
        return "done-a"

    def branch_b(state: FlowState) -> str:
        state.data["result"] = "B"
        return "done-b"

    flow.add_start(ingest)
    flow.add_listen("ingest", transform)
    flow.add_router("transform", decide, {"a": "branch_a", "b": "branch_b"})
    flow.add_branch(branch_a, step_id="branch_a")
    flow.add_branch(branch_b, step_id="branch_b")
    return flow

_print_state(state: FlowState, title: str = 'Flow') -> None

Imprime con Rich el estado resultante del flow.

Source code in src/ciel/cli/flow.py

def _print_state(state: FlowState, title: str = "Flow") -> None:
    """Imprime con Rich el estado resultante del flow."""
    completed = " -> ".join(state.completed) if state.completed else "(none)"
    data_keys = ", ".join(state.data.keys()) if state.data else "(none)"
    table = Table(title=title)
    table.add_column("completed")
    table.add_column("data keys")
    table.add_column("results")
    table.add_column("last_event")
    table.add_row(
        completed,
        data_keys,
        repr(state.results),
        str(state.last_event),
    )
    console.print(table)
    console.print(
        Panel.fit(
            f"completed: {len(state.completed)} step(s)\n"
            f"data keys: {', '.join(state.data.keys()) or '(none)'}\n"
            f"result: {state.data.get('result', '(none)')}\n"
            f"last_event: {state.last_event}",
            title=title,
            border_style="blue",
        )
    )

resolve_db_path(db_flag: Optional[Path]) -> str

Resuelve la ruta del grafo SQLite.

Prioridad: --db > variable de entorno CIEL_FLOW_DB > archivo por defecto en el directorio actual.

Source code in src/ciel/cli/flow.py

def resolve_db_path(db_flag: Optional[Path]) -> str:
    """Resuelve la ruta del grafo SQLite.

    Prioridad: ``--db`` > variable de entorno ``CIEL_FLOW_DB`` > archivo por
    defecto en el directorio actual.
    """
    if db_flag is not None:
        return str(db_flag)
    env = os.environ.get("CIEL_FLOW_DB")
    if env:
        return env
    return str(Path.cwd() / DEFAULT_DB_NAME)

resume(run_id: str = typer.Option(..., '--run-id', help='Run id to resume'), db: Path = typer.Option(..., '--db', help='Same SQLite db used in run'), tenant: Optional[str] = typer.Option(None, '--tenant', help='Tenant id')) -> None

Resume an interrupted flow from its last checkpoint (requires --db).

Source code in src/ciel/cli/flow.py

@flow_app.command("resume")
def resume(
    run_id: str = typer.Option(..., "--run-id", help="Run id to resume"),
    db: Path = typer.Option(..., "--db", help="Same SQLite db used in run"),
    tenant: Optional[str] = typer.Option(None, "--tenant", help="Tenant id"),
) -> None:
    """Resume an interrupted flow from its last checkpoint (requires --db)."""
    db_path = resolve_db_path(db)
    memory = MemoryStore(db_path)
    checkpointer = FlowCheckpointStore(memory)

    flow = _build_demo_flow()
    runner = flow.compile(
        supervisor=Supervisor(),
        checkpointer=checkpointer,
        tenant_id=tenant,
    )

    async def _resume() -> FlowState:
        return await runner.resume(run_id=run_id)

    try:
        state = asyncio.run(_resume())
    except Exception as exc:
        console.print(f"[red]resume failed:[/] {exc}")
        raise typer.Exit(1)
    finally:
        memory.close()

    _print_state(state, title="Resumed flow")
    console.print(
        Panel.fit(
            f"run_id: {run_id}\ncheckpointer: {db_path}\ntenant: {tenant or '(none)'}",
            title="Resume summary",
            border_style="blue",
        )
    )

run(run_id: Optional[str] = typer.Option(None, '--run-id', help='Explicit run id (for later resume)'), db: Optional[Path] = typer.Option(None, '--db', help='SQLite db for the checkpointer (or CIEL_FLOW_DB)'), tenant: Optional[str] = typer.Option(None, '--tenant', help='Tenant id')) -> None

Run an in-memory demo flow offline (no network, no provider).

With --db the run is persisted via a checkpointer so it can be resumed later with the resume command.

Source code in src/ciel/cli/flow.py

@flow_app.command("run")
def run(
    run_id: Optional[str] = typer.Option(None, "--run-id", help="Explicit run id (for later resume)"),
    db: Optional[Path] = typer.Option(None, "--db", help="SQLite db for the checkpointer (or CIEL_FLOW_DB)"),
    tenant: Optional[str] = typer.Option(None, "--tenant", help="Tenant id"),
) -> None:
    """Run an in-memory demo flow offline (no network, no provider).

    With --db the run is persisted via a checkpointer so it can be resumed
    later with the `resume` command.
    """
    checkpointer = None
    memory = None
    db_path = None
    if db is not None:
        db_path = resolve_db_path(db)
        memory = MemoryStore(db_path)
        checkpointer = FlowCheckpointStore(memory)

    flow = _build_demo_flow()
    runner = flow.compile(
        supervisor=Supervisor(),
        checkpointer=checkpointer,
        tenant_id=tenant,
    )

    async def _run() -> FlowState:
        return await runner.run(initial_data={}, run_id=run_id)

    try:
        state = asyncio.run(_run())
    except KeyboardInterrupt:
        raise typer.Exit(0)
    finally:
        if memory is not None:
            memory.close()

    _print_state(state, title="Demo flow")
    if db_path is not None:
        console.print(
            Panel.fit(
                f"run_id: {runner.run_id}\ncheckpointer: {db_path}\ntenant: {tenant or '(none)'}\n\n"
                "Resume with: ciel flow resume --run-id <run_id> --db <db> [--tenant <tenant>]",
                title="Run summary",
                border_style="blue",
            )
        )
    else:
        console.print(
            Panel.fit(
                "Offline demo (no --db): ingest -> transform -> router {a,b}.\n"
                "No provider, no network required. Pass --db to enable resume.",
                title="Summary",
                border_style="blue",
            )
        )

DEFAULT_DB_NAME = 'ciel_board.sqlite3' module-attribute

__all__ = ['board_app', 'add', 'list_tasks', 'show', 'assign'] module-attribute

board_app = typer.Typer(name='board', help='Manage the durable kanban board') module-attribute

console = Console() module-attribute

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()

add(title: str = typer.Argument(..., help='Task title'), task_id: Optional[str] = typer.Option(None, '--id', help='Task id'), assignee: Optional[str] = typer.Option(None, '--assignee', '-a', help='Assignee'), tenant_id: Optional[str] = typer.Option(None, '--tenant-id', help='Tenant id'), db: Optional[str] = typer.Option(None, '--db', help='Ruta del board SQLite (o CIEL_BOARD_DB)')) -> None

Source code in src/ciel/cli/board.py

@board_app.command("add")
def add(
    title: str = typer.Argument(..., help="Task title"),
    task_id: Optional[str] = typer.Option(None, "--id", help="Task id"),
    assignee: Optional[str] = typer.Option(None, "--assignee", "-a", help="Assignee"),
    tenant_id: Optional[str] = typer.Option(None, "--tenant-id", help="Tenant id"),
    db: Optional[str] = typer.Option(None, "--db", help="Ruta del board SQLite (o CIEL_BOARD_DB)"),
) -> None:
    with KanbanBoard(path=resolve_db_path(db)) as board:
        task = BoardTask(id=task_id or "", title=title, assignee=assignee, tenant_id=tenant_id)
        board.add_task(task)
        console.print(f"added {task.id}")

assign(task_id: str = typer.Argument(...), assignee: str = typer.Argument(...), db: Optional[str] = typer.Option(None, '--db', help='Ruta del board SQLite (o CIEL_BOARD_DB)')) -> None

Source code in src/ciel/cli/board.py

@board_app.command("assign")
def assign(
    task_id: str = typer.Argument(...),
    assignee: str = typer.Argument(...),
    db: Optional[str] = typer.Option(None, "--db", help="Ruta del board SQLite (o CIEL_BOARD_DB)"),
) -> None:
    with KanbanBoard(path=resolve_db_path(db)) as board:
        task = board.assign(task_id, assignee)
        if not task:
            console.print(f"not found {task_id}")
            raise typer.Exit(1)
        console.print(f"assigned {task.id} -> {task.assignee}")

list_tasks(status: Optional[str] = typer.Option(None, '--status', '-s'), assignee: Optional[str] = typer.Option(None, '--assignee', '-a'), tenant_id: Optional[str] = typer.Option(None, '--tenant-id'), db: Optional[str] = typer.Option(None, '--db', help='Ruta del board SQLite (o CIEL_BOARD_DB)')) -> None

Source code in src/ciel/cli/board.py

@board_app.command("list")
def list_tasks(
    status: Optional[str] = typer.Option(None, "--status", "-s"),
    assignee: Optional[str] = typer.Option(None, "--assignee", "-a"),
    tenant_id: Optional[str] = typer.Option(None, "--tenant-id"),
    db: Optional[str] = typer.Option(None, "--db", help="Ruta del board SQLite (o CIEL_BOARD_DB)"),
) -> None:
    with KanbanBoard(path=resolve_db_path(db)) as board:
        items = board.list_tasks(status=status, assignee=assignee, tenant_id=tenant_id)
        if not items:
            console.print("No tasks")
            raise typer.Exit(0)
        table = Table(title="Board")
        table.add_column("id")
        table.add_column("title")
        table.add_column("status")
        table.add_column("assignee")
        table.add_column("tenant_id")
        for item in items:
            table.add_row(item.id, item.title, item.status, item.assignee or "", item.tenant_id or "")
        console.print(table)

resolve_db_path(db_flag: Optional[str]) -> Optional[str]

Resuelve la ruta del board SQLite.

Prioridad: --db > variable de entorno CIEL_BOARD_DB > archivo por defecto en el directorio actual. Si nada está configurado se devuelve None (board en memoria, comportamiento legacy).

Source code in src/ciel/cli/board.py

def resolve_db_path(db_flag: Optional[str]) -> Optional[str]:
    """Resuelve la ruta del board SQLite.

    Prioridad: ``--db`` > variable de entorno ``CIEL_BOARD_DB`` > archivo por
    defecto en el directorio actual. Si nada está configurado se devuelve
    ``None`` (board en memoria, comportamiento legacy).
    """
    if db_flag:
        return db_flag
    env = os.environ.get("CIEL_BOARD_DB")
    if env:
        return env
    return str(Path.cwd() / DEFAULT_DB_NAME)

show(task_id: str = typer.Argument(...), db: Optional[str] = typer.Option(None, '--db', help='Ruta del board SQLite (o CIEL_BOARD_DB)')) -> None

Source code in src/ciel/cli/board.py

@board_app.command("show")
def show(
    task_id: str = typer.Argument(...),
    db: Optional[str] = typer.Option(None, "--db", help="Ruta del board SQLite (o CIEL_BOARD_DB)"),
) -> None:
    with KanbanBoard(path=resolve_db_path(db)) as board:
        task = board.show(task_id)
        if not task:
            console.print(f"not found {task_id}")
            raise typer.Exit(1)
        console.print(f"id: {task.id}")
        console.print(f"title: {task.title}")
        console.print(f"status: {task.status}")
        console.print(f"assignee: {task.assignee or ''}")
        console.print(f"tenant_id: {task.tenant_id or ''}")

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

console = Console() module-attribute

swarm_app = typer.Typer(name='swarm', help='Run agent swarms from an AgentSpec') module-attribute

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)

Budget dataclass

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

@dataclass
class Budget:
    max_tools: int = 16
    max_tokens: Optional[int] = None
    max_seconds: Optional[float] = None

RateLimiter

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

class RateLimiter:
    def __init__(self) -> None:
        self._calls: Dict[str, int] = {}

    def check(self, key: str, limit: int) -> Optional[str]:
        current = self._calls.get(key, 0) + 1
        self._calls[key] = current
        if current > limit:
            return "rate limit exceeded"
        return None

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)

TopologyEngine

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

class TopologyEngine:
    def __init__(
        self,
        agent_spec: Any,
        runner: Any,
        budget: Optional[Any] = None,
        rate_limiter: Optional[Any] = None,
        rate_limits: Optional[Dict[str, int]] = None,
        counter_for_step: Optional[Any] = None,
    ) -> None:
        self.spec = agent_spec
        self.runner = runner
        self.budget = budget
        self.rate_limiter = rate_limiter
        self.rate_limits = rate_limits or {}
        self.counter_for_step = counter_for_step
        self._executed: Dict[str, Any] = {}

    async def _execute(self, step) -> Any:
        run_fn = getattr(self.runner, "run", self.runner)
        if not callable(run_fn):
            raise TypeError("runner must be callable or async-callable")
        candidate = run_fn(step)
        if hasattr(candidate, "__await__"):
            return await candidate
        return candidate

    async def run(self) -> Any:
        topology = getattr(self.spec, "topology", "pipeline")
        if topology == "pipeline":
            return await self._pipeline()
        if topology == "fan-out":
            return await self._fan_out()
        if topology == "debate":
            return await self._debate()
        raise TopologyError(f"topology '{topology}' unsupported")

    async def _reject_if_budget_or_rate_exceeded(self, step) -> None:
        if self.budget is None:
            return
        counter = self.counter_for_step(step) if callable(self.counter_for_step) else None
        if counter is None:
            return
        exceeded = counter.exceed(self.budget)
        if exceeded:
            raise TopologyError(f"budget rejection on step '{step.id}': {exceeded}")
        if self.rate_limiter is not None:
            limit = self.rate_limits.get(step.id, 0)
            if limit > 0:
                exceeded = self.rate_limiter.check(step.id, limit)
                if exceeded:
                    raise TopologyError(f"rate limit rejection on step '{step.id}': {exceeded}")

    async def _pipeline(self) -> List[Any]:
        results: List[Any] = []
        for step in self.spec.steps:
            missing = [name for name in getattr(step, "depends_on", []) if name not in self._executed]
            if missing:
                raise TopologyError(f"missing dependencies: {sorted(missing)}")
            await self._reject_if_budget_or_rate_exceeded(step)
            results.append(await self._execute(step))
            self._executed[step.id] = True
        return results

    async def _fan_out(self) -> Dict[str, Any]:
        results: Dict[str, Any] = {}
        for step in self.spec.steps:
            await self._reject_if_budget_or_rate_exceeded(step)
            results[step.id] = await self._execute(step)
            self._executed[step.id] = True
        return results

    async def _debate(self) -> List[Any]:
        if not self.spec.steps:
            return []
        primary, *rest = self.spec.steps
        await self._reject_if_budget_or_rate_exceeded(primary)
        output = await self._execute(primary)
        transcript: List[Any] = [output]
        for step in rest:
            await self._reject_if_budget_or_rate_exceeded(step)
            output = await self._execute(step)
            transcript.append(output)
        return transcript

_AdHocContext dataclass

Source code in src/ciel/cli/swarm.py

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

_NoCounter

Source code in src/ciel/cli/swarm.py

class _NoCounter:
    def exceed(self, budget):
        return None

    def consume_tool(self, count: int = 1) -> None:
        return None

_SupervisorStepRunner

Source code in src/ciel/cli/swarm.py

class _SupervisorStepRunner:
    def __init__(self, supervisor: Supervisor) -> None:
        self.supervisor = supervisor

    async def run(self, step: Any) -> Any:
        result = await self.supervisor.run(
            step.id,
            self._worker_for(step.id),
            payload=None,
            worker_id=f"step-{step.id}",
        )
        return result.output

    def _worker_for(self, step_id: str) -> Worker:
        async def _execute(ctx: WorkerContext) -> Dict[str, Any]:
            return {"output": {"step_id": ctx.step_id, "worker_id": ctx.worker_id}}
        return _execute

swarm_run(spec: typer.FileText = typer.Option(..., '--spec', '-s', help='YAML AgentSpec file'), max_tools: int = typer.Option(8, '--max-tools', help='Max tool calls'), max_tokens: int | None = typer.Option(None, '--max-tokens', help='Max token count'), seconds: float = typer.Option(60.0, '--seconds', help='Max wall-clock seconds'), rate_limit: int = typer.Option(0, '--rate-limit', help='Per-step rate limit (0 disables)')) -> None

Source code in src/ciel/cli/swarm.py

@swarm_app.command("run")
def swarm_run(
    spec: typer.FileText = typer.Option(..., "--spec", "-s", help="YAML AgentSpec file"),
    max_tools: int = typer.Option(8, "--max-tools", help="Max tool calls"),
    max_tokens: int | None = typer.Option(None, "--max-tokens", help="Max token count"),
    seconds: float = typer.Option(60.0, "--seconds", help="Max wall-clock seconds"),
    rate_limit: int = typer.Option(0, "--rate-limit", help="Per-step rate limit (0 disables)"),
) -> None:
    agent_spec = AgentSpec.from_yaml(spec.read())
    budget = Budget(max_tools=max_tools, max_tokens=max_tokens, max_seconds=seconds)
    rate_limiter = RateLimiter() if rate_limit > 0 else None
    rate_limits = {step.id: rate_limit for step in agent_spec.steps} if rate_limiter is not None else {}

    supervisor = Supervisor(
        budget=budget,
        agent_counter=_NoCounter(),
        rate_limit=rate_limit,
        rate_limiter=rate_limiter,
    )
    runner = _SupervisorStepRunner(supervisor=supervisor)

    async def run_swarm() -> None:
        engine = TopologyEngine(
            agent_spec,
            runner=runner,
            budget=budget,
            rate_limiter=rate_limiter,
            rate_limits=rate_limits,
        )
        outputs = await engine.run()
        rows = outputs if isinstance(outputs, list) else [outputs]
        table = Table(title=f"Swarm: {agent_spec.name}")
        table.add_column("step")
        table.add_column("output")
        for row in rows:
            step_id = row.get("step_id") if isinstance(row, dict) else str(row)
            table.add_row(str(step_id), str(row))
        console.print(table)
        console.print(Panel.fit(f"max_tools={budget.max_tools}\nmax_tokens={budget.max_tokens}\nmax_seconds={budget.max_seconds}", title="Budget", border_style="blue"))

    try:
        asyncio.run(run_swarm())
    except KeyboardInterrupt:
        raise typer.Exit(0)

CLI ciel cost — cost governance (offline-safe, demo en memoria).

Comandos

ciel cost record --tenant T --model gpt-4o --in 1000 --out 500 [--price-in X --price-out Y] ciel cost status --tenant T ciel cost check --tenant T --model gpt-4o --in 1000 --out 500

console = Console() module-attribute

cost_app = typer.Typer(name='cost', help='Cost governance per tenant/model') module-attribute

BudgetExceededError

Bases: Exception

Se lanza cuando una operación superaría el presupuesto del tenant.

Source code in src/ciel/enterprise/cost.py

class BudgetExceededError(Exception):
    """Se lanza cuando una operación superaría el presupuesto del tenant."""

CostGovernor

Gobernador de costos por tenant con presupuesto, alertas y corte.

Source code in src/ciel/enterprise/cost.py

class CostGovernor:
    """Gobernador de costos por tenant con presupuesto, alertas y corte."""

    def __init__(
        self,
        *,
        tenant_id: Optional[str] = None,
        budgets: Optional[dict] = None,
        models: Optional[dict] = None,
        alert_threshold: float = 0.8,
    ) -> None:
        # ``budgets``: dict[tenant_id o "*"] -> float ($ límite)
        # ``models``: dict[model] -> ModelCost
        self.tenant_id = tenant_id
        self.budgets: dict = dict(budgets or {})
        self.models: dict = dict(models or {})
        self.alert_threshold = alert_threshold
        # gasto acumulado por tenant (estado en memoria)
        self._spent: dict[str, float] = {}

    # -- cálculo -----------------------------------------------------------
    def estimate(
        self, model: str, input_tokens: int, output_tokens: int
    ) -> float:
        """Costo estimado de una llamada (en $)."""
        if model not in self.models:
            raise CostError(f"modelo desconocido: {model!r}")
        cost = self.models[model]
        est_in = (input_tokens / 1000.0) * cost.per_1k_input
        est_out = (output_tokens / 1000.0) * cost.per_1k_output
        return est_in + est_out

    # -- registro ----------------------------------------------------------
    def record(
        self,
        tenant_id: str,
        model: str,
        input_tokens: int,
        output_tokens: int,
    ) -> float:
        """Acumula el gasto del tenant y devuelve el gasto actual."""
        amount = self.estimate(model, input_tokens, output_tokens)
        self._spent[tenant_id] = self._spent.get(tenant_id, 0.0) + amount
        return self._spent[tenant_id]

    def spent(self, tenant_id: str) -> float:
        """Gasto acumulado del tenant."""
        return self._spent.get(tenant_id, 0.0)

    # -- presupuesto -------------------------------------------------------
    def budget_of(self, tenant_id: str) -> float:
        """Presupuesto efectivo: el del tenant o el global "*"."""
        if tenant_id in self.budgets:
            return float(self.budgets[tenant_id])
        if "*" in self.budgets:
            return float(self.budgets["*"])
        return 0.0

    def remaining(self, tenant_id: str) -> float:
        """Presupuesto restante del tenant."""
        return self.budget_of(tenant_id) - self.spent(tenant_id)

    # -- corte / alertas ---------------------------------------------------
    def allowed(
        self,
        tenant_id: str,
        model: str,
        input_tokens: int,
        output_tokens: int,
    ) -> bool:
        """False si gasto actual + estimado supera el presupuesto."""
        if model not in self.models:
            raise CostError(f"modelo desconocido: {model!r}")
        projected = self.spent(tenant_id) + self.estimate(
            model, input_tokens, output_tokens
        )
        return projected <= self.budget_of(tenant_id)

    def check_budget(
        self,
        tenant_id: str,
        model: str,
        input_tokens: int,
        output_tokens: int,
    ) -> None:
        """Lanza ``BudgetExceededError`` si la operación no está permitida."""
        if not self.allowed(tenant_id, model, input_tokens, output_tokens):
            raise BudgetExceededError(
                f"presupuesto excedido para tenant {tenant_id!r}"
            )

    def alerted(self, tenant_id: str) -> bool:
        """True si el gasto cruzó el umbral de alerta (alert_threshold)."""
        budget = self.budget_of(tenant_id)
        if budget <= 0:
            return False
        return self.spent(tenant_id) >= self.alert_threshold * budget

alerted(tenant_id: str) -> bool

True si el gasto cruzó el umbral de alerta (alert_threshold).

Source code in src/ciel/enterprise/cost.py

def alerted(self, tenant_id: str) -> bool:
    """True si el gasto cruzó el umbral de alerta (alert_threshold)."""
    budget = self.budget_of(tenant_id)
    if budget <= 0:
        return False
    return self.spent(tenant_id) >= self.alert_threshold * budget

allowed(tenant_id: str, model: str, input_tokens: int, output_tokens: int) -> bool

False si gasto actual + estimado supera el presupuesto.

Source code in src/ciel/enterprise/cost.py

def allowed(
    self,
    tenant_id: str,
    model: str,
    input_tokens: int,
    output_tokens: int,
) -> bool:
    """False si gasto actual + estimado supera el presupuesto."""
    if model not in self.models:
        raise CostError(f"modelo desconocido: {model!r}")
    projected = self.spent(tenant_id) + self.estimate(
        model, input_tokens, output_tokens
    )
    return projected <= self.budget_of(tenant_id)

budget_of(tenant_id: str) -> float

Presupuesto efectivo: el del tenant o el global "*".

Source code in src/ciel/enterprise/cost.py

def budget_of(self, tenant_id: str) -> float:
    """Presupuesto efectivo: el del tenant o el global "*"."""
    if tenant_id in self.budgets:
        return float(self.budgets[tenant_id])
    if "*" in self.budgets:
        return float(self.budgets["*"])
    return 0.0

check_budget(tenant_id: str, model: str, input_tokens: int, output_tokens: int) -> None

Lanza BudgetExceededError si la operación no está permitida.

Source code in src/ciel/enterprise/cost.py

def check_budget(
    self,
    tenant_id: str,
    model: str,
    input_tokens: int,
    output_tokens: int,
) -> None:
    """Lanza ``BudgetExceededError`` si la operación no está permitida."""
    if not self.allowed(tenant_id, model, input_tokens, output_tokens):
        raise BudgetExceededError(
            f"presupuesto excedido para tenant {tenant_id!r}"
        )

estimate(model: str, input_tokens: int, output_tokens: int) -> float

Costo estimado de una llamada (en $).

Source code in src/ciel/enterprise/cost.py

def estimate(
    self, model: str, input_tokens: int, output_tokens: int
) -> float:
    """Costo estimado de una llamada (en $)."""
    if model not in self.models:
        raise CostError(f"modelo desconocido: {model!r}")
    cost = self.models[model]
    est_in = (input_tokens / 1000.0) * cost.per_1k_input
    est_out = (output_tokens / 1000.0) * cost.per_1k_output
    return est_in + est_out

record(tenant_id: str, model: str, input_tokens: int, output_tokens: int) -> float

Acumula el gasto del tenant y devuelve el gasto actual.

Source code in src/ciel/enterprise/cost.py

def record(
    self,
    tenant_id: str,
    model: str,
    input_tokens: int,
    output_tokens: int,
) -> float:
    """Acumula el gasto del tenant y devuelve el gasto actual."""
    amount = self.estimate(model, input_tokens, output_tokens)
    self._spent[tenant_id] = self._spent.get(tenant_id, 0.0) + amount
    return self._spent[tenant_id]

remaining(tenant_id: str) -> float

Presupuesto restante del tenant.

Source code in src/ciel/enterprise/cost.py

def remaining(self, tenant_id: str) -> float:
    """Presupuesto restante del tenant."""
    return self.budget_of(tenant_id) - self.spent(tenant_id)

spent(tenant_id: str) -> float

Gasto acumulado del tenant.

Source code in src/ciel/enterprise/cost.py

def spent(self, tenant_id: str) -> float:
    """Gasto acumulado del tenant."""
    return self._spent.get(tenant_id, 0.0)

ModelCost dataclass

Source code in src/ciel/enterprise/cost.py

@dataclass(frozen=True)
class ModelCost:
    per_1k_input: float
    per_1k_output: float

_governor() -> CostGovernor

Source code in src/ciel/cli/cost.py

def _governor() -> CostGovernor:
    return CostGovernor(
        budgets={"*": 10.0},
        models={
            "gpt-4o": ModelCost(per_1k_input=0.005, per_1k_output=0.015),
            "echo": ModelCost(per_1k_input=0.0, per_1k_output=0.0),
        },
    )

check(tenant: str = typer.Option(..., '--tenant', help='Tenant id'), model: str = typer.Option(..., '--model', help='Model id'), input_tokens: int = typer.Option(..., '--in', help='Input tokens'), output_tokens: int = typer.Option(..., '--out', help='Output tokens'), price_in: float = typer.Option(0.005, '--price-in', help='$/1k input'), price_out: float = typer.Option(0.015, '--price-out', help='$/1k output')) -> None

Check whether a planned usage is within budget (exit 1 if denied).

Source code in src/ciel/cli/cost.py

@cost_app.command("check")
def check(
    tenant: str = typer.Option(..., "--tenant", help="Tenant id"),
    model: str = typer.Option(..., "--model", help="Model id"),
    input_tokens: int = typer.Option(..., "--in", help="Input tokens"),
    output_tokens: int = typer.Option(..., "--out", help="Output tokens"),
    price_in: float = typer.Option(0.005, "--price-in", help="$/1k input"),
    price_out: float = typer.Option(0.015, "--price-out", help="$/1k output"),
) -> None:
    """Check whether a planned usage is within budget (exit 1 if denied)."""
    gov = _governor()
    gov.models[model] = ModelCost(per_1k_input=price_in, per_1k_output=price_out)
    try:
        gov.check_budget(tenant, model, input_tokens, output_tokens)
    except BudgetExceededError as exc:
        console.print(f"[red]DENY[/] {exc}")
        raise typer.Exit(code=1)
    console.print(
        f"[green]ALLOW[/] tenant={tenant} model={model} "
        f"estimated=${gov.estimate(model, input_tokens, output_tokens):.4f}"
    )

record(tenant: str = typer.Option(..., '--tenant', help='Tenant id'), model: str = typer.Option(..., '--model', help='Model id'), input_tokens: int = typer.Option(..., '--in', help='Input tokens'), output_tokens: int = typer.Option(..., '--out', help='Output tokens'), price_in: float = typer.Option(0.005, '--price-in', help='$/1k input'), price_out: float = typer.Option(0.015, '--price-out', help='$/1k output')) -> None

Record a usage and print the running spend for the tenant.

Source code in src/ciel/cli/cost.py

@cost_app.command("record")
def record(
    tenant: str = typer.Option(..., "--tenant", help="Tenant id"),
    model: str = typer.Option(..., "--model", help="Model id"),
    input_tokens: int = typer.Option(..., "--in", help="Input tokens"),
    output_tokens: int = typer.Option(..., "--out", help="Output tokens"),
    price_in: float = typer.Option(0.005, "--price-in", help="$/1k input"),
    price_out: float = typer.Option(0.015, "--price-out", help="$/1k output"),
) -> None:
    """Record a usage and print the running spend for the tenant."""
    gov = _governor()
    gov.models[model] = ModelCost(per_1k_input=price_in, per_1k_output=price_out)
    spent = gov.record(tenant, model, input_tokens, output_tokens)
    console.print(
        f"[green]recorded[/] tenant={tenant} model={model} "
        f"spent=${spent:.4f} remaining=${gov.remaining(tenant):.4f}"
    )

status(tenant: str = typer.Option(..., '--tenant', help='Tenant id')) -> None

Show current spend / budget / remaining for a tenant.

Source code in src/ciel/cli/cost.py

@cost_app.command("status")
def status(tenant: str = typer.Option(..., "--tenant", help="Tenant id")) -> None:
    """Show current spend / budget / remaining for a tenant."""
    gov = _governor()
    table = Table(title=f"Cost status — {tenant}")
    table.add_column("metric")
    table.add_column("value")
    table.add_row("spent", f"${gov.spent(tenant):.4f}")
    table.add_row("budget", f"${gov.budget_of(tenant):.4f}")
    table.add_row("remaining", f"${gov.remaining(tenant):.4f}")
    table.add_row("alerted", str(gov.alerted(tenant)))
    console.print(table)

CLI ciel rbac — gestión de roles y permisos (offline-safe, demo en memoria).

Comandos

ciel rbac list-roles # imprime roles y permisos por defecto ciel rbac assign --subject X --role admin [--tenant T] ciel rbac check --subject X --action agent:run [--tenant T]

DEFAULT_ROLES: dict[str, Role] = {'admin': Role('admin', frozenset({'agent:*', 'tools:*', 'admin:*', 'board:*', 'cost:*', 'approve:*'})), 'operator': Role('operator', frozenset({'agent:run', 'tools:exec', 'board:write'})), 'viewer': Role('viewer', frozenset({'agent:read', 'board:read'}))} module-attribute

console = Console() module-attribute

rbac_app = typer.Typer(name='rbac', help='RBAC roles and permissions') module-attribute

RBACEngine

Source code in src/ciel/enterprise/rbac.py

class RBACEngine:
    def __init__(self, *, tenant_id=None, roles=None, assignments=None):
        self.tenant_id = tenant_id
        # roles: mezcla DEFAULT_ROLES con los dados
        self.roles: dict[str, Role] = dict(DEFAULT_ROLES)
        if roles:
            self.roles.update(roles)
        # assignments: dict[(tenant_id or "*", subject)] -> role
        self._assignments: dict[tuple[str, str], str] = {}
        if assignments:
            if isinstance(assignments, dict):
                for key, role in assignments.items():
                    self._assignments[key] = role
            else:
                for a in assignments:
                    key = (a.tenant_id or "*", a.subject)
                    self._assignments[key] = a.role

    def assign(self, subject: str, role_name: str, *, tenant_id=None) -> None:
        self._assignments[(tenant_id or "*", subject)] = role_name

    def revoke(self, subject: str, *, tenant_id=None) -> None:
        self._assignments.pop((tenant_id or "*", subject), None)

    def role_of(self, subject: str, *, tenant_id=None) -> Optional[str]:
        # asignación específica de tenant > asignación global ("*") > None
        if tenant_id is not None:
            role = self._assignments.get((tenant_id, subject))
            if role is not None:
                return role
        role = self._assignments.get(("*", subject))
        if role is not None:
            return role
        return None

    def has_permission(self, subject: str, action: str, *, tenant_id=None) -> bool:
        role_name = self.role_of(subject, tenant_id=tenant_id)
        if role_name is None:
            return False
        role = self.roles.get(role_name)
        if role is None:
            return False
        return _permission_matches(role.permissions, action)

    def check(self, subject: str, action: str, *, tenant_id=None) -> None:
        if not self.has_permission(subject, action, tenant_id=tenant_id):
            raise RBACError(
                f"subject '{subject}' no autorizado para '{action}'"
                f" (tenant={tenant_id})"
            )

    def list_roles(self) -> list[str]:
        return sorted(self.roles.keys())

    def snapshot(self) -> dict:
        return {
            "tenant_id": self.tenant_id,
            "roles": {
                name: sorted(role.permissions) for name, role in self.roles.items()
            },
            "assignments": {
                f"{k[0]}\x1f{k[1]}": v for k, v in self._assignments.items()
            },
        }

    @classmethod
    def from_snapshot(cls, data: dict) -> "RBACEngine":
        roles = {
            name: Role(name, frozenset(perms))
            for name, perms in data.get("roles", {}).items()
        }
        assignments: dict[tuple[str, str], str] = {}
        for key, role in data.get("assignments", {}).items():
            tid, subject = key.split("\x1f", 1)
            assignments[(tid, subject)] = role
        return cls(
            tenant_id=data.get("tenant_id"),
            roles=roles,
            assignments=assignments,
        )

RBACError

Bases: Exception

Source code in src/ciel/enterprise/rbac.py

class RBACError(Exception):
    ...

assign(subject: str = typer.Option(..., '--subject', help='Subject (user/service)'), role: str = typer.Option(..., '--role', help='Role name'), tenant: str | None = typer.Option(None, '--tenant', help='Tenant id')) -> None

Assign a role to a subject (demo in-memory engine).

Source code in src/ciel/cli/rbac.py

@rbac_app.command("assign")
def assign(
    subject: str = typer.Option(..., "--subject", help="Subject (user/service)"),
    role: str = typer.Option(..., "--role", help="Role name"),
    tenant: str | None = typer.Option(None, "--tenant", help="Tenant id"),
) -> None:
    """Assign a role to a subject (demo in-memory engine)."""
    engine = RBACEngine()
    try:
        engine.assign(subject, role, tenant_id=tenant)
    except RBACError as exc:
        console.print(f"[red]error:[/] {exc}")
        raise typer.Exit(code=1)
    effective = tenant or "*"
    console.print(f"[green]assigned[/] role={role} subject={subject} tenant={effective}")

check(subject: str = typer.Option(..., '--subject', help='Subject (user/service)'), action: str = typer.Option(..., '--action', help='Action, e.g. agent:run'), tenant: str | None = typer.Option(None, '--tenant', help='Tenant id')) -> None

Check whether a subject may perform an action (demo in-memory engine).

Source code in src/ciel/cli/rbac.py

@rbac_app.command("check")
def check(
    subject: str = typer.Option(..., "--subject", help="Subject (user/service)"),
    action: str = typer.Option(..., "--action", help="Action, e.g. agent:run"),
    tenant: str | None = typer.Option(None, "--tenant", help="Tenant id"),
) -> None:
    """Check whether a subject may perform an action (demo in-memory engine)."""
    engine = RBACEngine()
    allowed = engine.has_permission(subject, action, tenant_id=tenant)
    role = engine.role_of(subject, tenant_id=tenant)
    if allowed:
        console.print(f"[green]ALLOW[/] {subject} -> {action} (role={role})")
    else:
        console.print(f"[red]DENY[/] {subject} -> {action} (role={role})")
        raise typer.Exit(code=1)

list_roles() -> None

List built-in roles and their permissions.

Source code in src/ciel/cli/rbac.py

@rbac_app.command("list-roles")
def list_roles() -> None:
    """List built-in roles and their permissions."""
    table = Table(title="RBAC roles")
    table.add_column("role")
    table.add_column("permissions")
    for name in sorted(DEFAULT_ROLES):
        role = DEFAULT_ROLES[name]
        table.add_row(name, ", ".join(sorted(role.permissions)))
    console.print(table)

Project scaffolding for ciel init.

Generates a minimal, offline-runnable Ciel project: a pyproject with a ciel.plugins entry point example, a starter agent module and a ciel.yaml. Idempotent: refuses to overwrite existing files unless force=True.

_AGENT = '"""Starter Ciel agent (offline-safe).\n\nRun with: uv run python {module}.py\n"""\nfrom __future__ import annotations\n\nfrom ciel.providers import OpenAICompatibleProvider\nfrom ciel.runtime import (\n ChatMessage,\n ChatRequest,\n DefaultAgentRuntime,\n DefaultToolDispatcher,\n ToolProvider,\n ToolRegistry,\n)\nfrom ciel.runtime.tools import ToolSpec, Tool, ToolsetSchema\n\n\ndef _echo(arguments, *, tool_call_id="", tenant_id=None):\n from ciel.runtime.tools import ToolResult\n return ToolResult(id=tool_call_id, name="echo", output={"echo": str(arguments.get("text", ""))})\n\n\ndef build_runtime():\n # Offline echo provider; swap for OpenAICompatibleProvider(base_url=..., api_key=...) with a real key.\n registry = ToolRegistry(default_toolset="default")\n registry.register_tool(\n "default",\n Tool(spec=ToolSpec(name="echo", description="Echo text", parameters={"text": {"type": "string"}}), callable_=_echo),\n )\n dispatcher = DefaultToolDispatcher(\n provider=ToolProvider(registry=registry, require_tenant_on_execution=False),\n default_toolset="default",\n )\n return DefaultAgentRuntime(provider=_EchoProvider(), dispatcher=dispatcher)\n\n\nclass _EchoProvider:\n provider_name = "echo"\n\n async def complete(self, request):\n from ciel.runtime import ChatChoice, ChatResponse\n prompt = request.messages[-1].content if request.messages else ""\n return ChatResponse(\n choice=ChatChoice(message=ChatMessage(role="assistant", content=f"echo: {prompt}"), finish_reason="stop"),\n metadata={},\n )\n\n async def stream(self, request):\n return (await self.complete(request),)\n\n async def models(self):\n from ciel.providers import ModelInfo\n return [ModelInfo(id="echo", provider="echo")]\n\n\nif __name__ == "__main__":\n import asyncio\n rt = build_runtime()\n result = asyncio.run(rt.run_agent_loop(request=ChatRequest(messages=[ChatMessage(role="user", content="hello")]), tenant_id="default"))\n print(getattr(getattr(result.response, "choice", None), "message", None).text())\n' module-attribute

_CIEL_YAML = 'project: {name}\ndefault_tenant: default\nproviders:\n - name: echo\n base_url: http://localhost:8000/v1\ntoolsets:\n - name: default\n description: default toolset\n' module-attribute

_PYPROJECT = '[project]\nname = "{name}"\nversion = "0.1.0"\ndescription = "A Ciel Agent Framework project"\nrequires-python = ">=3.11"\ndependencies = ["mana-ciel"]\n\n[project.entry-points."ciel.plugins"]\n# Register your own providers/tools/agents here, e.g.:\n# my_provider = my_project.plugins:register_provider\n\n[build-system]\nrequires = ["setuptools>=68"]\nbuild-backend = "setuptools.build_meta"\n\n[tool.setuptools]\npy-modules = ["{module}"]\n' module-attribute

scaffold_project(target: Path, *, force: bool = False) -> List[str]

Source code in src/ciel/cli/scaffold.py

def scaffold_project(target: Path, *, force: bool = False) -> List[str]:
    target.mkdir(parents=True, exist_ok=True)
    name = target.resolve().name or "my-ciel-project"
    safe = name.replace("-", "_").replace(" ", "_") or "my_ciel_project"
    module = f"{safe}_agent"

    created: List[str] = []

    def write(rel: str, content: str) -> None:
        p = target / rel
        if p.exists() and not force:
            return
        p.write_text(content, encoding="utf-8")
        created.append(rel)

    write("pyproject.toml", _PYPROJECT.replace("{name}", name).replace("{module}", module))
    write(f"{module}.py", _AGENT.replace("{module}", module))
    write("ciel.yaml", _CIEL_YAML.replace("{name}", name))
    return created

CLI ciel skills — dynamic skill library management (offline-safe).

Comandos

ciel skills list ciel skills create --name N --description D --code-file F ciel skills verify --name N --test-cases FILE ciel skills remove --name N

Todo opera sobre un :class:SkillLibrary en memoria/offline (sin red ni API keys), usando las primitivas de :mod:ciel.runtime.skills_lib (SkillLibrary, SkillVerifier). Se exponen funciones de lógica pura (list_registered, create_skill, verify_skill, remove_skill) que reciben la librería explícitamente, de modo que los tests pueden inyectar su propia instancia y compartir estado entre llamadas.

console = Console() module-attribute

skills_app = typer.Typer(name='skills', help='Dynamic skill library management (offline)') module-attribute

Skill dataclass

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

@dataclass
class Skill:
    name: str
    description: str
    content: str
    category: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)
    sha256: Optional[str] = None

SkillLibrary

Writable, in-memory skill store that wraps a :class:SkillRegistry.

The registry remains the source of truth for disk-loaded skills; the library layer adds creation, registration, update (with version bump) and removal of skills that live only in memory. Tenant isolation is supported via the optional tenant_id key on each stored :class:Skill.

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

class SkillLibrary:
    """Writable, in-memory skill store that wraps a :class:`SkillRegistry`.

    The registry remains the source of truth for disk-loaded skills; the
    library layer adds creation, registration, update (with version bump) and
    removal of skills that live only in memory. Tenant isolation is supported
    via the optional ``tenant_id`` key on each stored :class:`Skill`.
    """

    def __init__(self, registry: Optional[SkillRegistry] = None) -> None:
        self.registry = registry or SkillRegistry()
        # name -> list of Skill (newest last). We keep a list so update() can
        # preserve previous_versions for the evolution tree.
        self._skills: Dict[str, List[Skill]] = {}

    # -- backed by the passive registry (backward-compatible) -----------------

    def load_from_disk(self) -> List[Skill]:
        """Discover skills from the registry roots and index them in-memory."""
        found = self.registry.discover()
        for skill in found:
            self._skills.setdefault(skill.name, []).append(skill)
        return found

    # -- writable store --------------------------------------------------------

    def _sha256(self, content: str) -> str:
        import hashlib

        return hashlib.sha256(content.encode("utf-8")).hexdigest()

    def create_from_code(
        self,
        *,
        name: str,
        description: str,
        code: str,
        category: Optional[str] = None,
        tenant_id: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> Skill:
        """Compile ``code`` (syntax check) and store it as a new skill.

        Raises :class:`SkillError` if the code does not compile. The skill is
        NOT executed here — execution belongs to :class:`SkillVerifier`.
        """
        try:
            compile(code, f"<skill:{name}>", "exec")
        except SyntaxError as exc:  # noqa: BLE001
            raise SkillError(f"skill '{name}' has invalid syntax: {exc}") from exc

        content = code
        skill = Skill(
            name=name,
            description=description,
            content=content,
            category=category,
            metadata={
                **(metadata or {}),
                **({"tenant_id": tenant_id} if tenant_id else {}),
                "sha256": self._sha256(content),
            },
            sha256=self._sha256(content),
        )
        self._skills.setdefault(name, []).append(skill)
        return skill

    def register(self, skill: Skill) -> Skill:
        """Register an already-built :class:`Skill` (e.g. disk-loaded)."""
        if skill.sha256 is None:
            skill.sha256 = self._sha256(skill.content)
        self._skills.setdefault(skill.name, []).append(skill)
        return skill

    def get(self, name: str) -> Optional[Skill]:
        versions = self._skills.get(name)
        if not versions:
            # Fall back to the disk registry (backward-compat lookup).
            return self.registry.get(name)
        return versions[-1]

    def list_skills(self, *, category: Optional[str] = None, tenant_id: Optional[str] = None) -> List[Skill]:
        out: List[Skill] = []
        for versions in self._skills.values():
            latest = versions[-1]
            if category is not None and latest.category != category:
                continue
            if tenant_id is not None and latest.metadata.get("tenant_id") != tenant_id:
                continue
            out.append(latest)
        # Also include any registry-only skills not yet indexed in memory.
        for skill in self.registry.list_skills(category=category):
            if skill.name not in self._skills:
                if tenant_id is None or skill.metadata.get("tenant_id") == tenant_id:
                    out.append(skill)
        return out

    def history(self, name: str) -> List[Skill]:
        """Return every stored version of ``name`` (oldest first)."""
        return list(self._skills.get(name, []))

    def remove(self, name: str) -> bool:
        if name in self._skills:
            del self._skills[name]
            return True
        return False

    def update(
        self,
        *,
        name: str,
        description: Optional[str] = None,
        code: Optional[str] = None,
        category: Optional[str] = None,
        bump: str = "patch",
    ) -> Skill:
        """Create a new version of an existing skill, preserving history.

        ``bump`` is one of ``major``/``minor``/``patch`` (semantic) and is
        recorded in ``metadata.version``. The previous version is preserved in
        ``history(name)``.
        """
        versions = self._skills.get(name)
        if not versions:
            disk = self.registry.get(name)
            if disk is None:
                raise SkillError(f"cannot update unknown skill '{name}'")
            versions = [disk]
            self._skills[name] = versions

        previous = versions[-1]
        if code is None:
            code = previous.content
        else:
            try:
                compile(code, f"<skill:{name}>", "exec")
            except SyntaxError as exc:  # noqa: BLE001
                raise SkillError(f"skill '{name}' update has invalid syntax: {exc}") from exc

        new_version = self._next_version(previous.metadata.get("version"), bump)
        updated = Skill(
            name=name,
            description=description if description is not None else previous.description,
            content=code,
            category=category if category is not None else previous.category,
            metadata={
                **previous.metadata,
                "version": new_version,
                "previous_version": previous.metadata.get("version"),
                "sha256": self._sha256(code),
            },
            sha256=self._sha256(code),
        )
        versions.append(updated)
        return updated

    @staticmethod
    def _next_version(current: Optional[str], bump: str) -> str:
        major, minor, patch = (0, 0, 0)
        if current:
            parts = (current.split(".") + ["0", "0", "0"])[:3]
            try:
                major, minor, patch = (int(p) for p in parts)
            except ValueError:
                major, minor, patch = (0, 0, 0)
        if bump == "major":
            major, minor, patch = major + 1, 0, 0
        elif bump == "minor":
            minor, patch = minor + 1, 0
        else:
            patch += 1
        return f"{major}.{minor}.{patch}"

create_from_code(*, name: str, description: str, code: str, category: Optional[str] = None, tenant_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> Skill

Compile code (syntax check) and store it as a new skill.

Raises :class:SkillError if the code does not compile. The skill is NOT executed here — execution belongs to :class:SkillVerifier.

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

def create_from_code(
    self,
    *,
    name: str,
    description: str,
    code: str,
    category: Optional[str] = None,
    tenant_id: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
) -> Skill:
    """Compile ``code`` (syntax check) and store it as a new skill.

    Raises :class:`SkillError` if the code does not compile. The skill is
    NOT executed here — execution belongs to :class:`SkillVerifier`.
    """
    try:
        compile(code, f"<skill:{name}>", "exec")
    except SyntaxError as exc:  # noqa: BLE001
        raise SkillError(f"skill '{name}' has invalid syntax: {exc}") from exc

    content = code
    skill = Skill(
        name=name,
        description=description,
        content=content,
        category=category,
        metadata={
            **(metadata or {}),
            **({"tenant_id": tenant_id} if tenant_id else {}),
            "sha256": self._sha256(content),
        },
        sha256=self._sha256(content),
    )
    self._skills.setdefault(name, []).append(skill)
    return skill

history(name: str) -> List[Skill]

Return every stored version of name (oldest first).

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

def history(self, name: str) -> List[Skill]:
    """Return every stored version of ``name`` (oldest first)."""
    return list(self._skills.get(name, []))

load_from_disk() -> List[Skill]

Discover skills from the registry roots and index them in-memory.

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

def load_from_disk(self) -> List[Skill]:
    """Discover skills from the registry roots and index them in-memory."""
    found = self.registry.discover()
    for skill in found:
        self._skills.setdefault(skill.name, []).append(skill)
    return found

register(skill: Skill) -> Skill

Register an already-built :class:Skill (e.g. disk-loaded).

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

def register(self, skill: Skill) -> Skill:
    """Register an already-built :class:`Skill` (e.g. disk-loaded)."""
    if skill.sha256 is None:
        skill.sha256 = self._sha256(skill.content)
    self._skills.setdefault(skill.name, []).append(skill)
    return skill

update(*, name: str, description: Optional[str] = None, code: Optional[str] = None, category: Optional[str] = None, bump: str = 'patch') -> Skill

Create a new version of an existing skill, preserving history.

bump is one of major/minor/patch (semantic) and is recorded in metadata.version. The previous version is preserved in history(name).

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

def update(
    self,
    *,
    name: str,
    description: Optional[str] = None,
    code: Optional[str] = None,
    category: Optional[str] = None,
    bump: str = "patch",
) -> Skill:
    """Create a new version of an existing skill, preserving history.

    ``bump`` is one of ``major``/``minor``/``patch`` (semantic) and is
    recorded in ``metadata.version``. The previous version is preserved in
    ``history(name)``.
    """
    versions = self._skills.get(name)
    if not versions:
        disk = self.registry.get(name)
        if disk is None:
            raise SkillError(f"cannot update unknown skill '{name}'")
        versions = [disk]
        self._skills[name] = versions

    previous = versions[-1]
    if code is None:
        code = previous.content
    else:
        try:
            compile(code, f"<skill:{name}>", "exec")
        except SyntaxError as exc:  # noqa: BLE001
            raise SkillError(f"skill '{name}' update has invalid syntax: {exc}") from exc

    new_version = self._next_version(previous.metadata.get("version"), bump)
    updated = Skill(
        name=name,
        description=description if description is not None else previous.description,
        content=code,
        category=category if category is not None else previous.category,
        metadata={
            **previous.metadata,
            "version": new_version,
            "previous_version": previous.metadata.get("version"),
            "sha256": self._sha256(code),
        },
        sha256=self._sha256(code),
    )
    versions.append(updated)
    return updated

SkillVerificationResult dataclass

Outcome of :meth:SkillVerifier.verify.

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

@dataclass
class SkillVerificationResult:
    """Outcome of :meth:`SkillVerifier.verify`."""

    passed: bool
    skill: str
    attempts: int = 0
    error: Optional[str] = None
    traceback: Optional[str] = None
    expected: Optional[Any] = None
    got: Optional[Any] = None

SkillVerifier

Offline verifier: syntax check + executable test cases.

A test case is a dict {"call": {...}, "expect": <value>}. The verifier executes the skill code in an isolated namespace, looks up a callable named after the skill (or the first callable defined), invokes it with call arguments and compares the result to expect.

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

class SkillVerifier:
    """Offline verifier: syntax check + executable test cases.

    A test case is a dict ``{"call": {...}, "expect": <value>}``. The verifier
    executes the skill code in an isolated namespace, looks up a callable named
    after the skill (or the first callable defined), invokes it with ``call``
    arguments and compares the result to ``expect``.
    """

    def __init__(self, library: Optional[SkillLibrary] = None) -> None:
        self.library = library

    def _resolve_callable(self, skill: Skill, namespace: Dict[str, Any]) -> Any:
        if skill.name in namespace and callable(namespace[skill.name]):
            return namespace[skill.name]
        callables = [v for v in namespace.values() if callable(v) and not v.__module__ == "builtins"]
        if not callables:
            raise SkillVerificationError(f"skill '{skill.name}' defines no callable to invoke")
        return callables[0]

    def verify(self, skill: Skill, *, test_cases: Sequence[Dict[str, Any]]) -> SkillVerificationResult:
        # 1) Syntax validation first.
        try:
            compile(skill.content, f"<skill:{skill.name}>", "exec")
        except SyntaxError as exc:  # noqa: BLE001
            return SkillVerificationResult(
                passed=False,
                skill=skill.name,
                attempts=0,
                error=f"syntax error: {exc}",
                traceback=str(exc),
            )

        namespace: Dict[str, Any] = {}
        try:
            exec(skill.content, namespace)  # noqa: S102 — offline, trusted-by-construction
        except Exception as exc:  # noqa: BLE001
            return SkillVerificationResult(
                passed=False,
                skill=skill.name,
                attempts=0,
                error=f"load error: {type(exc).__name__}: {exc}",
                traceback=repr(exc),
            )

        fn = self._resolve_callable(skill, namespace)

        last_traceback: Optional[str] = None
        for attempt, case in enumerate(test_cases, start=1):
            call_args = case.get("call", {}) or {}
            expected = case.get("expect")
            try:
                got = fn(**call_args)
            except Exception as exc:  # noqa: BLE001
                last_traceback = repr(exc)
                return SkillVerificationResult(
                    passed=False,
                    skill=skill.name,
                    attempts=attempt,
                    error=f"case {attempt} raised {type(exc).__name__}: {exc}",
                    traceback=last_traceback,
                    expected=expected,
                )
            if got != expected:
                return SkillVerificationResult(
                    passed=False,
                    skill=skill.name,
                    attempts=attempt,
                    error=f"case {attempt}: expected {expected!r}, got {got!r}",
                    expected=expected,
                    got=got,
                )
        return SkillVerificationResult(
            passed=True,
            skill=skill.name,
            attempts=len(test_cases),
        )

    def verify_by_name(self, name: str, *, test_cases: Sequence[Dict[str, Any]]) -> SkillVerificationResult:
        if self.library is None:
            raise SkillVerificationError("SkillVerifier was built without a library; pass the skill directly")
        skill = self.library.get(name)
        if skill is None:
            raise SkillVerificationError(f"unknown skill '{name}'")
        return self.verify(skill, test_cases=test_cases)

_library() -> SkillLibrary

Librería en memoria por defecto para la CLI (offline, fresh por proceso).

Source code in src/ciel/cli/skills_cli.py

def _library() -> SkillLibrary:
    """Librería en memoria por defecto para la CLI (offline, fresh por proceso)."""
    return SkillLibrary()

_load_test_cases(path: str) -> List[Dict[str, Any]]

Carga casos de prueba desde un archivo JSON (lista de dicts).

Source code in src/ciel/cli/skills_cli.py

def _load_test_cases(path: str) -> List[Dict[str, Any]]:
    """Carga casos de prueba desde un archivo JSON (lista de dicts)."""
    data = json.loads(Path(path).read_text(encoding="utf-8"))
    if not isinstance(data, list):
        raise typer.BadParameter("test-cases file must contain a JSON list")
    return data

create_cmd(name: str = typer.Option(..., '--name', help='Skill name'), description: str = typer.Option(..., '--description', help='Skill description'), code_file: Path = typer.Option(..., '--code-file', exists=True, dir_okay=False, help='Path to a .py file with the skill code'), category: str | None = typer.Option(None, '--category', help='Optional category')) -> None

Crea un skill desde un archivo de código y lo registra en memoria.

Source code in src/ciel/cli/skills_cli.py

@skills_app.command("create")
def create_cmd(
    name: str = typer.Option(..., "--name", help="Skill name"),
    description: str = typer.Option(..., "--description", help="Skill description"),
    code_file: Path = typer.Option(
        ..., "--code-file", exists=True, dir_okay=False, help="Path to a .py file with the skill code"
    ),
    category: str | None = typer.Option(None, "--category", help="Optional category"),
) -> None:
    """Crea un skill desde un archivo de código y lo registra en memoria."""
    lib = _library()
    try:
        skill = create_skill(
            lib,
            name=name,
            description=description,
            code_file=str(code_file),
            category=category,
        )
    except Exception as exc:  # SyntaxError de compile() se propaga como SkillError
        console.print(f"[red]create failed[/] {exc}")
        raise typer.Exit(code=1) from exc
    console.print(
        f"[green]created[/] name={skill.name} sha={skill.sha256[:12] if skill.sha256 else '(none)'} "
        f"category={skill.category or '(none)'}"
    )

create_skill(lib: SkillLibrary, *, name: str, description: str, code_file: str, category: str | None = None) -> Skill

Lee code_file y registra un skill vía SkillLibrary.create_from_code.

Source code in src/ciel/cli/skills_cli.py

def create_skill(
    lib: SkillLibrary,
    *,
    name: str,
    description: str,
    code_file: str,
    category: str | None = None,
) -> Skill:
    """Lee ``code_file`` y registra un skill vía ``SkillLibrary.create_from_code``."""
    path = Path(code_file)
    code = path.read_text(encoding="utf-8")
    return lib.create_from_code(
        name=name,
        description=description,
        code=code,
        category=category,
    )

list_cmd() -> None

Lista los skills registrados en la librería en memoria.

Source code in src/ciel/cli/skills_cli.py

@skills_app.command("list")
def list_cmd() -> None:
    """Lista los skills registrados en la librería en memoria."""
    lib = _library()
    skills = list_registered(lib)
    table = Table(title="Registered skills")
    table.add_column("name")
    table.add_column("category")
    table.add_column("description")
    if not skills:
        console.print("[dim](no skills registered)[/]")
        return
    for skill in skills:
        table.add_row(
            skill.name,
            skill.category or "(none)",
            (skill.description or "").strip(),
        )
    console.print(table)

list_registered(lib: SkillLibrary) -> List[Skill]

Devuelve los skills registrados (última versión de cada nombre).

Source code in src/ciel/cli/skills_cli.py

def list_registered(lib: SkillLibrary) -> List[Skill]:
    """Devuelve los skills registrados (última versión de cada nombre)."""
    return lib.list_skills()

remove_cmd(name: str = typer.Option(..., '--name', help='Skill name to remove')) -> None

Elimina un skill de la librería en memoria.

Source code in src/ciel/cli/skills_cli.py

@skills_app.command("remove")
def remove_cmd(
    name: str = typer.Option(..., "--name", help="Skill name to remove"),
) -> None:
    """Elimina un skill de la librería en memoria."""
    lib = _library()
    removed = remove_skill(lib, name=name)
    if removed:
        console.print(f"[green]removed[/] name={name}")
    else:
        console.print(f"[yellow]not found[/] name={name}")
        raise typer.Exit(code=1)

remove_skill(lib: SkillLibrary, *, name: str) -> bool

Elimina name de la librería en memoria. Devuelve True si existía.

Source code in src/ciel/cli/skills_cli.py

def remove_skill(lib: SkillLibrary, *, name: str) -> bool:
    """Elimina ``name`` de la librería en memoria. Devuelve True si existía."""
    return lib.remove(name)

verify_cmd(name: str = typer.Option(..., '--name', help='Skill name to verify'), test_cases_file: Path = typer.Option(..., '--test-cases', exists=True, dir_okay=False, help='JSON file: list of {"call": {}, "expect": value}')) -> None

Verifica un skill contra casos de prueba (offline, sin ejecución de red).

Source code in src/ciel/cli/skills_cli.py

@skills_app.command("verify")
def verify_cmd(
    name: str = typer.Option(..., "--name", help="Skill name to verify"),
    test_cases_file: Path = typer.Option(
        ..., "--test-cases", exists=True, dir_okay=False, help="JSON file: list of {\"call\": {}, \"expect\": value}"
    ),
) -> None:
    """Verifica un skill contra casos de prueba (offline, sin ejecución de red)."""
    lib = _library()
    try:
        cases = _load_test_cases(str(test_cases_file))
    except Exception as exc:
        console.print(f"[red]invalid test-cases file[/] {exc}")
        raise typer.Exit(code=1) from exc

    result = verify_skill(lib, name=name, test_cases=cases)
    if result.passed:
        console.print(f"[green]PASS[/] skill={result.skill} cases={result.attempts}")
    else:
        console.print(
            f"[red]FAIL[/] skill={result.skill} "
            f"(attempt {result.attempts}): {result.error}"
        )
        raise typer.Exit(code=1)

verify_skill(lib: SkillLibrary, *, name: str, test_cases: Sequence[Dict[str, Any]]) -> SkillVerificationResult

Verifica name contra test_cases usando :class:SkillVerifier.

Source code in src/ciel/cli/skills_cli.py

def verify_skill(
    lib: SkillLibrary,
    *,
    name: str,
    test_cases: Sequence[Dict[str, Any]],
) -> SkillVerificationResult:
    """Verifica ``name`` contra ``test_cases`` usando :class:`SkillVerifier`."""
    verifier = SkillVerifier(library=lib)
    return verifier.verify_by_name(name, test_cases=test_cases)

---

ciel skills — gestión de la Skill Library (offline)

Subcomando Typer registrado como ciel skills. Toda la operación es offline-safe (sin red ni API keys): opera sobre una SkillLibrary en memoria usando ciel.runtime.skills_lib.

ciel skills list
ciel skills create --name add --description "Suma dos enteros" --code-file add.py
ciel skills verify --name add --test-cases cases.json
ciel skills remove --name add

ciel skills list

Lista los skills registrados en la librería en memoria (última versión de cada nombre), en una tabla Rich con columnas name / category / description. Si no hay ninguno, imprime (no skills registered).

ciel skills create

Crea un skill desde un archivo de código Python y lo registra.

Opción	Requerida	Descripción
--name	sí	Nombre del skill.
--description	sí	Descripción del skill.
--code-file	sí	Ruta a un .py con el código fuente (debe compilar; si no, falla con mensaje y exit 1).
--category	no	Categoría opcional.

Al registrar, imprime el sha256 de los 12 primeros caracteres y la categoría.

ciel skills verify

Verifica un skill contra casos de prueba, offline.

Opción	Requerida	Descripción
--name	sí	Nombre del skill a verificar.
--test-cases	sí	Archivo JSON: una lista de {"call": {...}, "expect": <valor>}.

El verificador ejecuta el código en un namespace aislado, invoca la callable (nombrada como el skill o la primera callable definida) con call y compara el resultado con expect. Imprime PASS/FAIL; en fallo hace exit 1.

[
  {"call": {"a": 2, "b": 3}, "expect": 5},
  {"call": {"a": 0, "b": 0}, "expect": 0}
]

ciel skills remove

Elimina un skill de la librería en memoria. --name requerido. Imprime removed si existía o not found + exit 1 si no.

Nota: la CLI usa una librería en memoria fresh por proceso, por lo que create/verify/remove no persisten entre invocaciones separadas. Para flujos persistentes usa la API de ciel.runtime.skills_lib directamente (ver docs/guide/skills.md).