Construye un workflow de programación al estilo Ultracode con Agno y AgentOS

Resumen

• Este diseño usa orquestación solo con Agno. Workflow es el planificador, las instancias nativas de Agent cumplen los roles de razonamiento y AgentOS proporciona el entorno de ejecución y la API. Aún se necesita infraestructura externa de Git y sandbox.
• Usa un paso personalizado del workflow para distribuir dinámicamente el trabajo entre workers. El bloque Parallel de Agno es excelente cuando las ramas se conocen al definir el workflow; un número de workers generado por el planificador es más fácil de ejecutar con asyncio.gather() dentro de un paso personalizado.
• Guarda metadatos compactos del workflow en el estado de sesión de Agno, pero mantén los diffs grandes, logs y artefactos de pruebas en almacenamiento de objetos.
• No permitas que la salida del modelo determine qué se integra. Un administrador de espacios de trabajo debe obtener los archivos modificados desde Git, componer un único candidato, ejecutar pruebas deterministas y promover solo el paquete de información del candidato revisado mediante el mecanismo HITL del workflow.
• Usa un Loop acotado para las rondas de implementación y reparación. Detente después de un número fijo y reducido de rondas en lugar de dejar que un ciclo de tareas de Agno Team se ejecute indefinidamente.
• AgentOS puede persistir en PostgreSQL las sesiones, el estado de workflows pausados, las trazas y los requisitos HITL de los workflows. Una confirmación del workflow no es lo mismo que el registro independiente de AgentOS en /approvals creado por una herramienta decorada con @approval.
• Para producción, agrega un pool de workers de sandbox, un administrador de leases, un almacén inmutable de artefactos, credenciales de corta duración, autorización por tenant, presupuestos y limpieza.

Qué aprenderás aquí

• Cómo adaptar el patrón Ultracode a las primitivas de Agno
• Cuándo usar Workflow, Loop, Parallel y pasos de funciones personalizadas
• Cómo crear agentes tipados de planificación, trabajo, reducción y revisión
• Cómo ejecutar simultáneamente un conjunto dinámico de workers de programación
• Cómo integrar y probar un candidato inmutable antes de la revisión
• Cómo pausar antes del merge y reanudar mediante AgentOS
• Qué infraestructura proporciona Agno y cuál sigue siendo tu responsabilidad
• Cómo evaluar el workflow de Agno frente a un solo agente de programación

Esta es la Parte 2 de la serie Arquitectura de Ultracode. La Parte 1 deriva la arquitectura portable. La Parte 3 es una alternativa independiente con LangGraph. Este artículo supone que elegiste Agno e implementa el patrón de workflow de programación sin LangGraph ni otro motor externo de workflows.

Por qué Agno puede aproximarse al patrón

Agno expone actualmente tres capas relevantes:

Agno Teams también admite los modos de coordinación, enrutamiento, difusión y tareas. No usaría un Team como planificador principal aquí.

El workflow ya controla:

• el número de rondas
• la distribución del trabajo entre workers
• la integración del candidato
• los puntos de control de revisión
• la finalización de la reparación
• la revisión del merge

Agregar un líder de Team que cree y actualice de forma independiente otra lista de tareas daría al sistema dos planificadores. Usa un Workflow para controlar explícitamente las rondas de reparación. TeamMode.tasks también está limitado por max_iterations, cuyo valor predeterminado es 10, pero su líder controla un ciclo de tareas separado, dirigido por el modelo.

La arquitectura es:

AgentOS API
    |
    v
Agno Workflow
    |
    +--> Plan step --------------------> planner Agent
    |
    +--> Bounded Loop
    |      |
    |      +--> prepare round
    |      +--> dynamic asyncio fan-out
    |      |      +--> coding Agent in workspace A
    |      |      +--> coding Agent in workspace B
    |      |      +--> coding Agent in workspace C
    |      +--> deterministic candidate integration
    |      +--> deterministic test runner
    |      +--> reducer Agent
    |      +--> verifier + refuter Agents
    |      +--> stop when passed or exhausted
    |
    +--> top-level candidate output review
    |
    +--> finalize
           +--> passed + changed: promote exact reviewed SHA
           +--> passed + read-only: return report
           +--> failed: return failure outcome

Estructura del proyecto

agno-ultracode/
  app/
    schemas.py
    agents.py
    workspace.py
    workflow.py
    runtime.py
  evals/
    cases.py
  tests/
    test_plan_invariants.py
    test_candidate_binding.py
    test_resume.py
  pyproject.toml

Los ejemplos se verificaron el July 15, 2026 con Python 3.10+ y Agno 2.7.3.

uv add "agno[os,openai,postgres]==2.7.3" \
  "psycopg[binary,pool]" sqlalchemy pydantic

Paso 1: Define contratos que el modelo no pueda reescribir

# app/schemas.py
from typing import Literal

from pydantic import BaseModel, Field, model_validator


class CodingRequest(BaseModel):
    repository: str
    base_sha: str
    request: str
    success_criteria: list[str] = Field(min_length=1)


class WorkItem(BaseModel):
    id: str
    objective: str
    read_only: bool
    write_globs: list[str] = Field(default_factory=list)
    additional_suite_ids: list[str] = Field(default_factory=list)

    @model_validator(mode="after")
    def validate_permissions(self):
        if self.read_only and self.write_globs:
            raise ValueError("read-only work cannot declare writes")
        if not self.read_only and not self.write_globs:
            raise ValueError("writable work requires declared scopes")
        return self


class WorkflowPlan(BaseModel):
    substantive: bool
    reason: str
    success_criteria: list[str] = Field(min_length=1)
    tasks: list[WorkItem] = Field(min_length=1, max_length=4)

    @model_validator(mode="after")
    def validate_wave(self):
        expected = range(2, 5) if self.substantive else range(1, 2)
        if len(self.tasks) not in expected:
            raise ValueError("routine plans need 1 task; substantive plans need 2-4")
        ids = [task.id for task in self.tasks]
        if len(ids) != len(set(ids)):
            raise ValueError("task IDs must be unique")
        return self


class WorkerOutput(BaseModel):
    status: Literal["completed", "blocked"]
    summary: str
    unresolved_risks: list[str] = Field(default_factory=list)


class ReviewOutput(BaseModel):
    candidate_id: str
    candidate_sha: str
    test_report_digest: str
    evidence_digest: str
    candidate_passed: bool
    blocking_issues: list[str] = Field(default_factory=list)
    evidence: list[str] = Field(default_factory=list)

    @model_validator(mode="after")
    def validate_review(self):
        if self.candidate_passed and self.blocking_issues:
            raise ValueError("passing reviews cannot contain blocking issues")
        return self


class SynthesisOutput(BaseModel):
    candidate_id: str
    candidate_sha: str
    test_report_digest: str
    evidence_digest: str
    summary: str
    acceptance_evidence: list[str] = Field(default_factory=list)
    unresolved_risks: list[str] = Field(default_factory=list)

El planificador puede sugerir alcances. El servicio de espacios de trabajo debe expandirlos usando el repositorio y rechazar cualquier superposición de archivos concretos antes de que comiencen los workers.

Paso 2: Crea agentes nativos con responsabilidades específicas

# app/agents.py
import os

from agno.agent import Agent
from agno.models.openai import OpenAIResponses

from app.schemas import ReviewOutput, SynthesisOutput, WorkerOutput, WorkflowPlan


def model() -> OpenAIResponses:
    return OpenAIResponses(
        id=os.environ["AGENT_MODEL_ID"],
        retries=2,
        delay_between_retries=1,
        exponential_backoff=True,
    )


def planner(tools: list) -> Agent:
    return Agent(
        id="ultracode-planner",
        model=model(),
        tools=tools,
        output_schema=WorkflowPlan,
        instructions=[
            "Inspect the repository before planning.",
            "Create one task for routine work or 2-4 independent tasks.",
            "Declare write scopes and optional additional verification suite IDs.",
            "Do not weaken the supplied success criteria during repair.",
        ],
    )


def worker(task_id: str, tools: list) -> Agent:
    return Agent(
        id=f"ultracode-worker-{task_id}",
        model=model(),
        tools=tools,
        output_schema=WorkerOutput,
        instructions=[
            "Complete only the assigned task.",
            "Use only the provided workspace tools.",
            "Report blocked work honestly.",
        ],
    )


def reviewer(role: str) -> Agent:
    instruction = (
        "Verify the candidate against every success criterion."
        if role == "verifier"
        else "Try to falsify the candidate using a concrete counterexample."
    )
    return Agent(
        id=f"ultracode-{role}",
        model=model(),
        output_schema=ReviewOutput,
        instructions=[
            instruction,
            "Use immutable diff and test evidence.",
            "candidate_passed means the exact candidate is safe to approve.",
        ],
    )


def reducer() -> Agent:
    return Agent(
        id="ultracode-reducer",
        model=model(),
        output_schema=SynthesisOutput,
        instructions=[
            "Summarize the immutable candidate and deterministic test evidence.",
            "Do not hide failed tests, contradictions, or unresolved risks.",
        ],
    )

Usa valores estables de session_id de Agno para la correlación. Agrega una base de datos de Agno a los agentes solo si quieres persistir sus sesiones individuales; la sesión del workflow sigue siendo la fuente de verdad para la orquestación.

Paso 3: Define los límites del espacio de trabajo

Los toolkits Workspace y CodingTools de Agno restringen las operaciones a un directorio raíz y pueden requerir confirmación. Su documentación aclara que restringir rutas no equivale a aislar procesos.

El workflow debe depender de un servicio como este:

# app/workspace.py
from typing import Protocol

from pydantic import BaseModel

from app.schemas import CodingRequest, WorkItem, WorkerOutput


class WorkerRecord(BaseModel):
    task_id: str
    round: int
    workspace_id: str
    base_sha: str
    changed_files: list[str]
    artifact_uris: list[str]
    output: WorkerOutput


class Candidate(BaseModel):
    id: str
    sha: str
    base_sha: str
    original_base_sha: str
    round: int
    changed_files: list[str]  # Cumulative from original_base_sha.
    diff_uri: str  # Full diff from original_base_sha to sha.


class TestReport(BaseModel):
    candidate_id: str
    candidate_sha: str
    passed: bool
    exit_codes: dict[str, int]
    artifact_uri: str
    digest: str


class EvidenceManifest(BaseModel):
    candidate_id: str
    candidate_sha: str
    diff_uri: str
    test_artifact_uri: str
    test_report_digest: str
    diff_text: str
    test_log_text: str
    digest: str


class WorkspaceService(Protocol):
    async def planner_tools(self, request: CodingRequest, *, user_id: str) -> list: ...
    async def resolve_suite_ids(
        self,
        request: CodingRequest,
        *,
        user_id: str,
        proposed_suite_ids: list[str],
    ) -> list[str]:
        """Return required plus allowlisted optional suites from server policy."""
        ...
    async def create_worker(
        self,
        request: CodingRequest,
        task: WorkItem,
        *,
        user_id: str,
        round: int,
        idempotency_key: str,
        parent_candidate_id: str | None,
    ) -> tuple[str, list]: ...
    async def inspect_and_freeze(
        self,
        workspace_id: str,
        task: WorkItem,
        output: WorkerOutput,
        *,
        round: int,
    ) -> WorkerRecord: ...
    async def compose(
        self,
        records: list[WorkerRecord],
        *,
        parent_candidate_id: str | None,
    ) -> Candidate: ...
    async def test(self, candidate: Candidate, suite_ids: list[str]) -> TestReport: ...
    async def evidence(
        self,
        candidate: Candidate,
        report: TestReport,
    ) -> EvidenceManifest: ...
    async def cancel(self, workspace_id: str) -> None: ...
    async def promote(
        self,
        candidate: Candidate,
        *,
        expected_sha: str,
        idempotency_key: str,
    ) -> dict: ...

Este servicio, no un agente, debe:

• obtener los archivos modificados desde Git
• rechazar escrituras fuera de los patrones glob declarados
• resolver los IDs de suites obligatorias y opcionales según la política del repositorio o tenant
• componer los parches en un único espacio de trabajo candidato
• conservar original_base_sha y obtener los archivos modificados acumulados y el diff completo desde esa base en cada ronda de reparación
• detectar conflictos de merge
• ejecutar comandos y capturar sus códigos de salida reales
• calcular hashes de los artefactos inmutables de diffs y pruebas
• promover solo el hash del candidato aprobado

Para desarrollo local de confianza, el servicio puede usar worktrees de Git. Para repositorios no confiables, ejecuta las herramientas de Agno dentro de un contenedor, una VM, un espacio de trabajo de Daytona o un sandbox equivalente.

Paso 4: Implementa una ronda dinámica

Parallel de Agno es ideal para un conjunto conocido de ramas. Aquí el planificador decide el número de workers en tiempo de ejecución, así que un paso asíncrono personalizado distribuye el trabajo.

# app/workflow.py
import asyncio
import json

from agno.run import RunContext
from agno.workflow import Loop, Step, StepInput, StepOutput, Workflow
from agno.workflow.types import OnError, OnReject
from pydantic import BaseModel

from app.agents import planner, reducer, reviewer, worker
from app.schemas import (
    CodingRequest,
    ReviewOutput,
    SynthesisOutput,
    WorkflowPlan,
    WorkerOutput,
)
from app.workspace import (
    Candidate,
    EvidenceManifest,
    TestReport,
    WorkerRecord,
    WorkspaceService,
)


class CandidateEnvelope(BaseModel):
    passed: bool
    candidate_id: str
    candidate_sha: str
    original_base_sha: str
    changed_files: list[str]
    test_report_digest: str
    evidence_digest: str
    diff_uri: str
    test_artifact_uri: str
    test_exit_codes: dict[str, int]
    synthesis: dict


def build_workflow(db, workspaces: WorkspaceService):
    async def create_plan(
        step_input: StepInput,
        run_context: RunContext,
    ) -> StepOutput:
        request = CodingRequest.model_validate(step_input.input)
        user_id = run_context.user_id
        if not user_id:
            raise PermissionError("authenticated user_id is required")
        tools = await workspaces.planner_tools(request, user_id=user_id)
        baseline_suite_ids = await workspaces.resolve_suite_ids(
            request,
            user_id=user_id,
            proposed_suite_ids=[],
        )
        if not baseline_suite_ids:
            raise RuntimeError("repository policy returned no required suites")
        response = await planner(tools).arun(
            {
                "request": request.request,
                "immutable_success_criteria": request.success_criteria,
                "required_suite_ids": baseline_suite_ids,
            },
            user_id=user_id,
            session_id=f"{run_context.session_id}:planner",
        )
        plan = WorkflowPlan.model_validate(response.content).model_copy(
            update={"success_criteria": request.success_criteria}
        )
        run_context.session_state.update(
            {
                "request": request.model_dump(),
                "user_id": user_id,
                "success_criteria": request.success_criteria,
                "baseline_suite_ids": baseline_suite_ids,
                "plan": plan.model_dump(),
                "round": 0,
                "parent_candidate_id": None,
                "round_passed": False,
                "exhausted": False,
            }
        )
        return StepOutput(content=plan.model_dump_json())

    async def execute_round(
        step_input: StepInput,
        run_context: RunContext,
    ) -> StepOutput:
        state = run_context.session_state
        request = CodingRequest.model_validate(state["request"])
        plan = WorkflowPlan.model_validate(state["plan"])
        round_number = state["round"]
        user_id = state["user_id"]

        async def run_worker(task):
            workspace_id = None
            try:
                workspace_id, tools = await workspaces.create_worker(
                    request,
                    task,
                    user_id=user_id,
                    round=round_number,
                    idempotency_key=(
                        f"{run_context.session_id}:{round_number}:{task.id}:workspace"
                    ),
                    parent_candidate_id=state["parent_candidate_id"],
                )
                response = await worker(task.id, tools).arun(
                    {
                        "objective": task.objective,
                        "workspace_id": workspace_id,
                        "additional_suite_ids": task.additional_suite_ids,
                    },
                    user_id=user_id,
                    session_id=(
                        f"{run_context.session_id}:worker:{round_number}:{task.id}"
                    ),
                )
                output = WorkerOutput.model_validate(response.content)
                return await workspaces.inspect_and_freeze(
                    workspace_id,
                    task,
                    output,
                    round=round_number,
                )
            except asyncio.CancelledError:
                if workspace_id:
                    await asyncio.shield(workspaces.cancel(workspace_id))
                raise

        child_tasks = [
            asyncio.create_task(run_worker(task))
            for task in plan.tasks
        ]
        try:
            records = await asyncio.gather(*child_tasks)
        except BaseException:
            for child in child_tasks:
                child.cancel()
            await asyncio.gather(*child_tasks, return_exceptions=True)
            raise
        if any(record.output.status != "completed" for record in records):
            raise RuntimeError("a worker did not complete")

        candidate = await workspaces.compose(
            records,
            parent_candidate_id=state["parent_candidate_id"],
        )
        expected_base_sha = (
            request.base_sha
            if round_number == 0
            else Candidate.model_validate(state["candidate"]).sha
        )
        if (
            candidate.base_sha != expected_base_sha
            or candidate.original_base_sha != request.base_sha
            or any(record.base_sha != expected_base_sha for record in records)
        ):
            raise RuntimeError("candidate lineage does not match expected base")
        suite_ids = await workspaces.resolve_suite_ids(
            request,
            user_id=user_id,
            proposed_suite_ids=[
                suite_id
                for task in plan.tasks
                for suite_id in task.additional_suite_ids
            ],
        )
        if not set(state["baseline_suite_ids"]).issubset(suite_ids):
            raise RuntimeError("verification policy dropped a required suite")
        report = await workspaces.test(candidate, suite_ids)
        if (
            report.candidate_id != candidate.id
            or report.candidate_sha != candidate.sha
        ):
            raise RuntimeError("test report does not match candidate")

        evidence = EvidenceManifest.model_validate(
            await workspaces.evidence(candidate, report)
        )
        if (
            evidence.candidate_id != candidate.id
            or evidence.candidate_sha != candidate.sha
            or evidence.test_report_digest != report.digest
        ):
            raise RuntimeError("evidence is not bound to candidate and tests")
        synthesis_response = await reducer().arun(
            {
                "success_criteria": state["success_criteria"],
                "candidate": candidate.model_dump(),
                "test_report": report.model_dump(),
                "evidence": evidence.model_dump(),
                "worker_records": [record.model_dump() for record in records],
            },
            user_id=user_id,
            session_id=f"{run_context.session_id}:reducer:{round_number}",
        )
        synthesis = SynthesisOutput.model_validate(synthesis_response.content)
        if (
            synthesis.candidate_id != candidate.id
            or synthesis.candidate_sha != candidate.sha
            or synthesis.test_report_digest != report.digest
            or synthesis.evidence_digest != evidence.digest
        ):
            raise RuntimeError("synthesis is not bound to immutable evidence")
        review_payload = json.dumps(
            {
                "success_criteria": state["success_criteria"],
                "candidate": candidate.model_dump(),
                "test_report": report.model_dump(),
                "evidence": evidence.model_dump(),
                "synthesis": synthesis.model_dump(),
            }
        )
        verifier_result, refuter_result = await asyncio.gather(
            reviewer("verifier").arun(
                review_payload,
                user_id=user_id,
                session_id=f"{run_context.session_id}:verifier:{round_number}",
            ),
            reviewer("refuter").arun(
                review_payload,
                user_id=user_id,
                session_id=f"{run_context.session_id}:refuter:{round_number}",
            ),
        )
        verifier = ReviewOutput.model_validate(verifier_result.content)
        refuter = ReviewOutput.model_validate(refuter_result.content)
        for review in (verifier, refuter):
            if (
                review.candidate_id != candidate.id
                or review.candidate_sha != candidate.sha
                or review.test_report_digest != report.digest
                or review.evidence_digest != evidence.digest
            ):
                raise RuntimeError("review is not bound to immutable evidence")

        passed = (
            report.passed
            and verifier.candidate_passed
            and refuter.candidate_passed
        )
        state.update(
            {
                "candidate": candidate.model_dump(),
                "test_report": report.model_dump(),
                "reviews": {
                    "verifier": verifier.model_dump(),
                    "refuter": refuter.model_dump(),
                },
                "synthesis": synthesis.model_dump(),
                "evidence": evidence.model_dump(),
                "round_passed": passed,
            }
        )

        if not passed:
            if round_number >= 2:
                state["exhausted"] = True
            else:
                repair_response = await planner(
                    await workspaces.planner_tools(request, user_id=user_id)
                ).arun(
                    json.dumps(
                        {
                            "request": request.request,
                            "immutable_success_criteria": state["success_criteria"],
                            "candidate": candidate.model_dump(),
                            "test_report": report.model_dump(),
                            "evidence": evidence.model_dump(),
                            "synthesis": synthesis.model_dump(),
                            "reviews": state["reviews"],
                            "instruction": "Create one bounded repair wave.",
                        }
                    ),
                    user_id=user_id,
                    session_id=(
                        f"{run_context.session_id}:repair:{round_number + 1}"
                    ),
                )
                repair_plan = WorkflowPlan.model_validate(repair_response.content)
                state["plan"] = repair_plan.model_copy(
                    update={"success_criteria": state["success_criteria"]}
                ).model_dump()
                state["parent_candidate_id"] = candidate.id
                state["round"] = round_number + 1

        return StepOutput(
            content=json.dumps(
                {
                    "round": round_number,
                    "passed": passed,
                    "exhausted": state["exhausted"],
                    "candidate_sha": candidate.sha,
                }
            ),
            success=passed or not state["exhausted"],
        )

    def end_loop(outputs: list[StepOutput]) -> bool:
        result = json.loads(outputs[-1].content)
        return result["passed"] or result["exhausted"]

    def candidate_envelope(
        step_input: StepInput,
        run_context: RunContext,
    ) -> StepOutput:
        state = run_context.session_state
        candidate = Candidate.model_validate(state["candidate"])
        envelope = CandidateEnvelope(
            passed=state["round_passed"],
            candidate_id=candidate.id,
            candidate_sha=candidate.sha,
            original_base_sha=candidate.original_base_sha,
            changed_files=candidate.changed_files,
            test_report_digest=state["test_report"]["digest"],
            evidence_digest=state["evidence"]["digest"],
            diff_uri=state["evidence"]["diff_uri"],
            test_artifact_uri=state["evidence"]["test_artifact_uri"],
            test_exit_codes=state["test_report"]["exit_codes"],
            synthesis=state["synthesis"],
        )
        return StepOutput(content=envelope.model_dump_json())

    async def finalize(
        step_input: StepInput,
        run_context: RunContext,
    ) -> StepOutput:
        state = run_context.session_state
        envelope = CandidateEnvelope.model_validate_json(
            step_input.previous_step_content
        )
        candidate = Candidate.model_validate(state["candidate"])
        if (
            envelope.candidate_id != candidate.id
            or envelope.candidate_sha != candidate.sha
            or envelope.original_base_sha != candidate.original_base_sha
            or envelope.test_report_digest != state["test_report"]["digest"]
            or envelope.evidence_digest != state["evidence"]["digest"]
        ):
            raise RuntimeError("reviewed envelope does not match session candidate")
        if not envelope.passed:
            return StepOutput(
                content=json.dumps(
                    {
                        "status": "failed",
                        "round": state["round"],
                        "reviews": state.get("reviews"),
                    }
                )
            )
        if candidate.sha == candidate.original_base_sha:
            return StepOutput(
                content=json.dumps(
                    {
                        "status": "completed",
                        "kind": "report",
                        "candidate_sha": candidate.sha,
                        "synthesis": state["synthesis"],
                    }
                )
            )
        result = await workspaces.promote(
            candidate,
            expected_sha=envelope.candidate_sha,
            idempotency_key=(
                f"{run_context.session_id}:merge:{candidate.sha}"
            ),
        )
        return StepOutput(content=json.dumps(result))

    review_candidate_step = Step(
        name="Review Candidate Envelope",
        executor=candidate_envelope,
        requires_output_review=lambda output: (
            CandidateEnvelope.model_validate_json(output.content).passed
            and CandidateEnvelope.model_validate_json(output.content).candidate_sha
            != CandidateEnvelope.model_validate_json(output.content).original_base_sha
        ),
        output_review_message=(
            "Review the candidate SHA, deterministic tests, and evidence."
        ),
        on_reject=OnReject.cancel,
        max_retries=0,
        on_error=OnError.fail,
    )

    return Workflow(
        id="agno-ultracode",
        name="Agno Ultracode Approximation",
        db=db,
        input_schema=CodingRequest,
        session_state={},
        store_events=True,
        steps=[
            Step(
                name="Plan",
                executor=create_plan,
                max_retries=0,
                on_error=OnError.fail,
            ),
            Loop(
                name="Implementation and Review",
                steps=[
                    Step(
                        name="Execute Round",
                        executor=execute_round,
                        max_retries=0,
                        on_error=OnError.fail,
                    )
                ],
                end_condition=end_loop,
                max_iterations=3,
            ),
            review_candidate_step,
            Step(
                name="Finalize",
                executor=finalize,
                max_retries=0,
                on_error=OnError.fail,
            ),
        ],
    )

Este diseño tiene intencionalmente un solo planificador:

• Workflow controla el orden.
• Loop controla el número máximo de rondas.
• el paso personalizado controla los workers dinámicos.
• los agentes solo toman decisiones de razonamiento tipadas.
• el servicio de espacios de trabajo controla la información verificable del candidato.

Correcciones importantes para producción

El código es una estructura inicial, no una plataforma de sandbox lista para usar.

Antes del despliegue:

1. Valida que las tareas del planificador tengan IDs únicos y archivos concretos que no se superpongan.
2. Haz que la creación de espacios de trabajo sea idempotente por sesión, ronda y tarea.
3. Pon en cuarentena o restablece los espacios de trabajo cuando haya un tiempo de espera agotado o una cancelación.
4. Obtén los archivos modificados y los códigos de salida de las pruebas fuera del modelo.
5. Mantén la revisión del candidato como un paso de nivel superior del workflow; los pasos Condition anidados no imponen la confirmación previa a la ejecución de la misma manera.
6. Incluye el ID del candidato, SHA, digest de pruebas y digest de evidencia en el paquete de información revisado, y vuelve a compararlos antes de la promoción.
7. Asegúrate de que los planes de reparación conserven los comandos de referencia y los criterios de éxito originales.
8. Limpia los espacios de trabajo al finalizar con éxito, ante fallas o cancelaciones y al vencer el TTL.

El estado del workflow de Agno se persiste en los límites entre pasos cuando hay una base de datos configurada. Un paso personalizado de distribución dinámica que falla puede necesitar ejecutarse de nuevo, por lo que toda operación externa sigue necesitando una clave de idempotencia. Este diseño usa la revisión persistida de la salida del workflow, no el sistema independiente de registros de herramientas @approval de AgentOS.

Paso 5: Sírvelo con AgentOS

# app/runtime.py
import os

from agno.db.postgres import PostgresDb
from agno.os import AgentOS
from agno.os.config import AuthorizationConfig

from app.bootstrap import production_workspaces
from app.workflow import build_workflow


workflow_db = PostgresDb(
    db_url=os.environ["AGNO_DATABASE_URL"],
    session_table="ultracode_workflow_sessions",
)
agentos_db = PostgresDb(
    db_url=os.environ["AGENTOS_DATABASE_URL"],
)

workflow = build_workflow(workflow_db, production_workspaces)

agent_os = AgentOS(
    id="ultracode-agent-os",
    workflows=[workflow],
    db=agentos_db,
    tracing=True,
    authorization=True,
    authorization_config=AuthorizationConfig(
        verification_keys=[os.environ["JWT_VERIFICATION_KEY"]],
        algorithm="RS256",
        user_isolation=True,
    ),
)

app = agent_os.get_app()

Usa una URI postgresql+psycopg://... para PostgresDb de Agno.

AgentOS proporciona endpoints de workflows, ejecuciones almacenadas, streaming compatible con SSE, APIs de cancelación, continuación HITL y visibilidad de trazas. En este ejemplo están habilitados tanto la autorización JWT como el aislamiento de usuarios; la autorización por sí sola no restringe todas las lecturas de la base de datos por usuario.

Advertencia específica de la versión: en el tag de Agno 2.7.3, la ruta de respuesta SSE de un workflow pausado carga una sesión por session_id sin pasar el user_id correspondiente al ámbito del usuario. No expongas esa ruta de streaming pausado a clientes multi-tenant no confiables en esta versión fijada sin un gateway o parche que verifique la propiedad. Actualiza después de verificar la corrección, o deshabilita el stream afectado y usa endpoints autenticados de ejecución y estado que verifiquen la propiedad. Esta precaución se basa en workflow_response_streamer y el middleware user_scope de ese tag, no en una afirmación general de que todas las rutas de AgentOS carecen de restricciones de ámbito.

Iniciar, continuar y cancelar

run = await workflow.arun(
    input=CodingRequest(
        repository="github.com/acme/api",
        base_sha="abc123",
        request="Audit every route for missing authorization checks.",
        success_criteria=["Every route has an explicit authorization decision."],
    ),
    session_id="ultra-123",
    user_id="user-42",
)

if run.is_paused:
    # Trusted in-process example; production uses authenticated HITL handling.
    for requirement in run.steps_requiring_output_review:
        requirement.confirm()
    run = await workflow.acontinue_run(run_response=run)

En producción, gestiona las solicitudes HITL del workflow mediante rutas autenticadas de continuación de AgentOS o una interfaz de administración que verifique la propiedad de la ejecución. Las llamadas directas a la biblioteca anteriores omiten la autenticación de AgentOS y solo son adecuadas para administración confiable dentro del proceso.

Limitación importante: workflow.acancel_run() marca el workflow para una cancelación cooperativa, pero los Agents invocados manualmente dentro de este paso dinámico personalizado no se registran automáticamente como ejecutores de Steps hijos nativos. Una implementación de producción necesita un puente de cancelación que rastree los IDs de ejecuciones hijas y espacios de trabajo, llame a la API de cancelación de cada Agent, termine los grupos de procesos del sandbox y espere a que finalicen todas las tareas hijas.

Requisitos de infraestructura

Client / IDE / CI
       |
       v
AgentOS API + JWT authorization
       |
       v
Agno Workflow runtime
       |
       +--> workflow Postgres
       +--> trace Postgres / OTel exporter
       +--> persisted workflow HITL state
       +--> lease / concurrency controller
       +--> object storage
       |
       v
Sandbox worker pool
  |        |        |
Agno     Agno     Agno
worker   worker   worker
  |        |        |
isolated repository workspaces
       |
       v
Git provider + CI

PostgreSQL es suficiente para una cola pequeña y una tabla de leases. Agrega Redis, Valkey o una cola administrada solo cuando la señalización entre réplicas y una mayor concurrencia justifiquen otra dependencia operativa.

Límites de seguridad

Agno ofrece controles útiles, pero úsalos en la capa adecuada:

• Workspace restringe las herramientas de archivos a un directorio raíz y una política de confirmación.
• CodingTools puede restringir los comandos y directorios base cuando se configura para ello.
• AgentOS hace copias profundas de los objetos mutables del entorno de ejecución para cada solicitud.
• La autorización JWT es opcional.
• El aislamiento de datos por usuario es opcional.

Ninguno de esos controles convierte la ejecución arbitraria de comandos de shell del repositorio en un sandbox de procesos.

Para producción multi-tenant:

• usa contenedores efímeros o un aislamiento más robusto
• deniega el acceso a la red por defecto
• emite credenciales de corta duración para cada worker
• monta solo las rutas asignadas del repositorio y los artefactos
• limita CPU, memoria, disco, tiempo, salida y tokens
• termina el grupo de procesos al cancelar
• trata los archivos del repositorio como instrucciones no confiables

Semántica de fallas

La cancelación de Agno es cooperativa. Tu proveedor de espacios de trabajo debe implementar la terminación forzada del proceso.

Una falla esperada en las pruebas o un rechazo durante la revisión es un resultado del dominio y aun así puede finalizar con una ejecución del workflow técnicamente completada. Los clientes deben inspeccionar el contenido final (promoted, report, failed o cancelled) en lugar de interpretar el estado de la ejecución por sí solo como éxito del proceso.

Evaluación

Compara:

1. Un Agent de programación de Agno
2. Un Agno Team en modo de tareas
3. Este Workflow de Agno

Usa el mismo modelo, commit del repositorio, herramientas, tareas y presupuesto.

Prueba al menos:

• tareas rutinarias con un solo worker
• auditorías paralelas de solo lectura
• cambios en varios archivos sin superposición
• planes con escrituras superpuestas que deben rechazarse
• pruebas fallidas seguidas de una reparación
• rechazo del candidato durante la revisión
• reinicio del proceso y reanudación
• cancelación durante un comando de un worker

Las suites de evaluación de Agno pueden combinar comprobaciones de Agent-as-Judge y Reliability, y controlar el resultado de CI mediante un código de salida. Úsalas para evaluar el comportamiento de los agentes y luego agrega verificaciones deterministas del workflow para los hashes de candidatos, las llamadas a herramientas, la revisión HITL y la duplicación de efectos secundarios.

Qué no copia esta aproximación

No reproduce:

• la heurística privada de activación de Claude Code
• los workflows generados en JavaScript
• el ensamblaje de contexto de Claude Code
• el nivel de esfuerzo xhigh específico del proveedor
• su interfaz de workflows ni su política interna de convergencia

Sí recrea el patrón de ingeniería útil:

typed plan
+ bounded parallel workers
+ persistent session state
+ deterministic candidate and tests
+ role-conditioned verification
+ human review
+ resumable runtime

Recomendación final

Comienza con:

• un Workflow de Agno
• agentes de rol nativos de Agno
• PostgreSQL
• un adaptador de worktrees para repositorios de confianza
• no más de cuatro workers simultáneos
• dos rondas de reparación
• evidencia obligatoria del candidato, las pruebas y la revisión antes del merge

Pasa a sandboxes de contenedores o VMs antes de trabajar con repositorios no confiables o en entornos multi-tenant.

La regla de operación es:

Agno Workflow planifica. Los agentes razonan. La infraestructura de espacios de trabajo aporta las pruebas. AgentOS persiste y gobierna.

Lecturas relacionadas

• Arquitectura de Claude Code Ultracode para Cursor y agentes personalizados
• Construye un workflow de programación al estilo Ultracode con LangGraph
• Agno/AgentOS + FastAPI frente a LangChain/LangGraph + FastAPI
• Agentes de 12 factores en la práctica: LangChain/LangGraph y Agno/AgentOS
• Generación segura de código con agentes

Fuentes

• Documentación de Agno, “Construcción de workflows”, consultado el July 15, 2026.
• Documentación de Agno, “Patrones avanzados de workflows”, consultado el July 15, 2026.
• Documentación de Agno, “Workflow paralelo”, consultado el July 15, 2026.
• Documentación de Agno, “Referencia de pasos Loop”, consultado el July 15, 2026.
• Documentación de Agno, “Workflow con un paso de función personalizada”, consultado el July 15, 2026.
• Documentación de Agno, “Estado de sesión de workflows”, consultado el July 15, 2026.
• Documentación de Agno, “Esquema de entrada de workflows”, consultado el July 15, 2026.
• Documentación de Agno, “Intervención humana en workflows”, consultado el July 15, 2026.
• Documentación de Agno, “Revisión de la salida del workflow”, consultado el July 15, 2026.
• Documentación de Agno, “Aprobación”, consultado el July 15, 2026.
• Documentación de Agno, “Referencia de Workflow”, consultado el July 15, 2026.
• Documentación de Agno, “Toolkit Workspace”, consultado el July 15, 2026.
• Documentación de Agno, “CodingTools”, consultado el July 15, 2026.
• Documentación de Agno, “Seguridad y autenticación”, consultado el July 15, 2026.
• Documentación de Agno, “Aislamiento de datos por usuario”, consultado el July 15, 2026.
• Documentación de Agno, “Ejecución en segundo plano”, consultado el July 15, 2026.
• Documentación de Agno, “Observabilidad de agentes”, consultado el July 15, 2026.
• Documentación de Agno, “Introducción a las evaluaciones”, consultado el July 15, 2026.
• PyPI, agno 2.7.3, July 14, 2026.