Guía técnica para comprender y reconstruir los 23 componentes de la arquitectura de Claude Code, desde el ciclo básico de un agente hasta los sistemas multiagente, la gestión de contexto y la ejecución asíncrona.

A principios de 2026, Claude Code superó los 1.000 millones de dólares en ingresos anualizados en apenas seis meses desde su lanzamiento.

No consiguió ese resultado gracias a mejores prompts. Lo consiguió porque Anthropic construyó la infraestructura adecuada alrededor del modelo correcto: un ciclo de ejecución de agentes con transmisión en tiempo real, un sistema de herramientas gobernado por permisos y una capa de gestión del contexto que permite mantener al modelo enfocado durante sesiones de trabajo prolongadas.

Esta infraestructura, conocida como harness, puede reproducirse. Y eso es precisamente lo que vamos a construir en este artículo.

La arquitectura de Claude Code

Claude Code está construido sobre cinco componentes fundamentales que trabajan de manera coordinada:

Un ciclo principal de ejecución de un solo hilo: dirige al modelo a través de ciclos de percepción, razonamiento y ejecución de herramientas, incorporando los resultados al contexto hasta completar la tarea.

Un registro de herramientas con tipado definido: relaciona nombres de herramientas con sus funciones de ejecución, como Bash, Read, Write, Grep o Glob. Cada herramienta cuenta con un esquema de entrada estricto que limita lo que el modelo puede solicitar y lo que la infraestructura puede ejecutar.

Una capa de gestión del contexto: combina la carga de habilidades bajo demanda, la compresión del historial de conversación en tres niveles y la memoria persistente en disco para mantener un razonamiento coherente durante sesiones que superan la ventana de contexto del modelo.

Un sistema de permisos basado en reglas: utiliza tres niveles de evaluación —denegar siempre, permitir siempre y solicitar aprobación del usuario— respaldados por un bus de eventos que permite observar e interceptar las llamadas a herramientas.

Una capa de coordinación multiagente: admite aislamiento del contexto de subagentes, delegación asíncrona, protocolos de comunicación controlados mediante máquinas de estados finitos y aislamiento mediante Git Worktrees para ejecutar tareas en paralelo sin conflictos entre archivos.

Y dentro de estos cinco componentes existen numerosos mecanismos adicionales que explican el funcionamiento de Claude Code.

En este artículo vamos a construir y probar cada componente individualmente para comprender cómo funcionan juntos y qué ventajas ofrecen frente a otras arquitecturas de agentes de IA.

El código del proyecto está disponible en GitHub:

Claude Code From Scratch: 23 componentes de la arquitectura de Claude Code

Estructura del proyecto

El repositorio está organizado de manera que cada componente puede ejecutarse y probarse por separado.

claude-code-from-scratch/
│
├── core.py
├── 01_perception_action_loop.py
├── s02_tool_use.py
├── s03_todo_write.py
├── s04_subagent.py
...
├── s10_team_protocols.py
├── s11_autonomous_agents.py
├── s12_worktree_task_isolation.py
│
├── s13_streaming.py
...
├── s21_mcp_runtime.py
│
├── s22_production_mailbox.py
├── s23_worktree_advanced.py
...
└── skills/

El archivo core.py contiene los fundamentos compartidos, como el cliente del modelo, las herramientas, el sistema de ejecución y los permisos.

Los demás archivos implementan mecanismos específicos. Esta separación permite ejecutar y probar individualmente cada componente sin duplicar innecesariamente la lógica principal.

¿Qué es Harness Engineering?

Harness Engineering es la disciplina que consiste en construir el entorno que rodea a un modelo de inteligencia artificial, en lugar de modificar el modelo en sí.

El modelo razona y toma decisiones. El harness ejecuta esas decisiones, establece restricciones y conecta el modelo con las herramientas y los sistemas externos.

Una infraestructura bien diseñada proporciona al modelo exactamente las herramientas que necesita, sin añadir capacidades innecesarias, y controla las acciones que puede realizar.

Este enfoque puede resumirse en cuatro principios fundamentales.

1. El modelo es la única fuente de decisiones. La infraestructura no introduce una lógica propia para interpretar y decidir qué hacer a partir de las respuestas del modelo. Su responsabilidad consiste en ejecutar las acciones solicitadas.

2. Las herramientas son la interfaz entre el modelo y el mundo exterior. Todas las operaciones, desde leer un archivo hasta iniciar un subagente, se realizan mediante llamadas a herramientas con esquemas definidos y entradas validadas.

3. El contexto es un recurso administrado. La información que recibe el modelo en cada interacción se selecciona, comprime e incorpora deliberadamente. No se acumula todo el historial sin control.

4. Los permisos son declarativos, no procedimentales. Las acciones permitidas, bloqueadas o sujetas a autorización se definen mediante configuraciones, en lugar de distribuir esas reglas entre múltiples condiciones dentro del código.

¿Cómo utiliza Claude Code el Harness Engineering?

Claude Code no es simplemente un framework para agentes. Es una infraestructura de ejecución diseñada para que un modelo pueda trabajar con herramientas de manera controlada.

Anthropic no construyó una lógica específica para decidir cuándo leer archivos o ejecutar pruebas. Proporcionó a Claude las herramientas necesarias para realizar esas operaciones y dejó que el modelo decidiera cuándo utilizarlas.

Su arquitectura aplica los principios de Harness Engineering de varias maneras.

El ciclo principal de ejecución es genérico y no conserva un estado propio específico de cada tarea. Funciona de la misma manera cuando el usuario solicita corregir un error de una línea o realizar una refactorización compleja, porque la inteligencia específica de cada tarea reside en el modelo.

El registro de herramientas funciona como punto de extensión. Añadir una capacidad significa registrar una nueva herramienta con su nombre, descripción y esquema de entrada.

El contexto se administra activamente. Según la arquitectura descrita, cuando el uso de la ventana se aproxima al 92 %, las interacciones anteriores pueden resumirse y almacenarse en disco para mantener la memoria de trabajo enfocada.

Por último, el sistema de permisos funciona como una capa previa a la ejecución: las herramientas deben pasar por una evaluación de reglas antes de ejecutarse.

Esto permite que la seguridad no dependa exclusivamente de cómo se comporte el modelo, sino también de la arquitectura que controla sus acciones.

Fase 1. El ciclo principal del agente

El ciclo de ejecución es el elemento arquitectónico fundamental sobre el que se construye todo lo demás.

Antes de las herramientas, los permisos o la coordinación multiagente, existe un ciclo que llama al modelo, identifica lo que quiere hacer, ejecuta la acción correspondiente y devuelve el resultado.

Cada componente de esta fase incorpora un mecanismo adicional sin modificar ese principio básico.

1. El ciclo mínimo con while

El principio más importante de cualquier sistema de agentes es el ciclo de percepción, acción y observación.

El agente recibe una tarea e intenta resolverla utilizando una herramienta. Después observa el resultado y decide si debe continuar o detenerse.

Estas decisiones están dirigidas por el modelo, no por el código externo.

No estamos hablando de un mecanismo convencional de reintentos ni de una estrategia de respaldo. Este ciclo constituye el núcleo operativo del sistema.

El mismo mecanismo puede utilizarse para corregir un pequeño error de programación o modificar una base de código completa.

Inicializar el cliente de Anthropic

Para construir nuestra implementación inicial, primero debemos configurar el cliente y el modelo.

from anthropic import Anthropic# Clave de API necesaria# Obtenla en: https://console.anthropic.com/ANTHROPIC_API_KEY = "sk-ant-xxx"# Identificador del modeloMODEL_ID = "claude-sonnet-4-6"# Inicializar el clienteclient = Anthropic(    base_url="https://api.anthropic.com",    api_key=ANTHROPIC_API_KEY)MODEL = MODEL_ID

En este ejemplo utilizaremos un modelo de Anthropic porque estamos reconstruyendo la arquitectura de Claude Code.

Sin embargo, también es posible utilizar LiteLLM para conectar modelos de diferentes proveedores. El repositorio está planteado de manera flexible para permitir esa sustitución.

Definir el prompt de sistema

El prompt de sistema establece el comportamiento general del agente.

DEFAULT_SYSTEM = (    f"You are a coding agent at {os.getcwd()}. "    "Use tools to solve tasks. Act, don't explain.")

Aunque el prompt de sistema no siempre requiere una configuración compleja, resulta importante para establecer cómo debe abordar las tareas el modelo.

Definir las herramientas iniciales

Claude Code se construye alrededor de herramientas. Por eso necesitamos ofrecer al agente una forma de interactuar con su entorno.

BASIC_TOOLS = [    {        "name": "bash",        "description": "Run a shell command.",        "input_schema": {            "type": "object",            "properties": {                "command": {"type": "string"}            },            "required": ["command"]        }    }]

Aquí definimos una herramienta llamada bash que permite ejecutar comandos de terminal.

La herramienta incluye un nombre, una descripción y un esquema que especifica el formato esperado de los parámetros.

Crear el registro de ejecución

BASIC_DISPATCH: dict = {    "bash": lambda inp: run_bash(inp["command"])}

El mecanismo de despacho conecta las llamadas del modelo con funciones reales.

En lugar de incorporar la lógica de cada herramienta al ciclo principal, utilizamos un diccionario que relaciona nombres con sus correspondientes funciones.

Esta separación facilita incorporar nuevas herramientas sin modificar el funcionamiento del agente.

Ejecutar las llamadas a herramientas

def dispatch_tools(    response_content: list,    dispatch: dict) -> List[Dict[str, Any]]:    results = []    for block in response_content:        if block.type != "tool_use":            continue        tool_name = block.name        tool_input = block.input        tool_use_id = block.id        handler = dispatch.get(tool_name)        first_val = (            str(list(tool_input.values())[0])[:80]            if tool_input else ""        )        print(            f"\033[33m[{tool_name}] "            f"{first_val}...\033[0m"        )        if handler:            try:                output = handler(tool_input)            except Exception as e:                output = (                    f"Error during tool execution: {e}"                )        else:            output = (                f"Error: Unknown tool '{tool_name}'"            )        print(str(output)[:300])        results.append({            "type": "tool_result",            "tool_use_id": tool_use_id,            "content": str(output),        })    return results

La función dispatch_tools recibe la respuesta del modelo, identifica las llamadas a herramientas, las ejecuta y recopila los resultados.

También registra las operaciones para mostrar qué herramientas está utilizando el agente y cuáles son sus resultados.

Construir el ciclo principal

def agent_loop(    messages: List[Dict[str, Any]],    dispatch: Dict) -> None:    while True:        print("\n\033[36m> Thinking...\033[0m")        response = client.messages.create(            model=MODEL,            system=DEFAULT_SYSTEM,            messages=messages,            tools=BASIC_TOOLS,            max_tokens=8000,        )        messages.append({            "role": "assistant",            "content": response.content        })        if response.stop_reason != "tool_use":            break        results = dispatch_tools(            response.content,            dispatch        )        messages.append({            "role": "user",            "content": results        })

Nuestro agente utiliza un ciclo while que continúa ejecutándose hasta que el modelo genera una respuesta final.

El funcionamiento es sencillo: se envía el historial al modelo, se recibe su respuesta, se comprueba si solicita herramientas y, en ese caso, se ejecutan para devolver los resultados al contexto.

El proceso se repite hasta que el modelo considera que ha terminado.

Probar el agente

Para comprobar su funcionamiento, creamos un historial de conversación y aceptamos instrucciones desde la terminal.

history: List[Dict[str, Any]] = []while True:    try:        query = input("\033[36ms01 >> \033[0m")    except (EOFError, KeyboardInterrupt):        print("\nExiting.")        break    if query.strip().lower() in ("q", "exit", ""):        break    history.append({        "role": "user",        "content": query    })    agent_loop(history, BASIC_DISPATCH)    last_message = history[-1]    print("\n\033[32mFinal Answer:\033[0m")    for block in last_message.get("content", []):        if block.type == "text":            print(block.text)    print()

Probemos con la siguiente instrucción:

s01 >> Enumera todos los archivos Python
       del directorio actual.

El agente intenta ejecutar:

ls *.py

El sistema devuelve un error porque no encuentra el comando.

A continuación, el modelo cambia de estrategia y utiliza:

find . -name "*.py" -type f

Esta vez obtiene los resultados:

./main.py
./utils.py
./agent.py

El ejemplo demuestra el ciclo de percepción, acción y observación.

El agente intentó ejecutar una operación, recibió un error y corrigió su procedimiento sin necesidad de incorporar instrucciones específicas para ese fallo.

2. El patrón de despacho de herramientas

Uno de los componentes más interesantes de la arquitectura de Claude Code es su registro de herramientas.

El artículo original describe una implementación con 18 herramientas registradas, entre ellas Bash, Read, Write, Edit, Glob, Grep, WebFetch, AskUserQuestion y TodoWrite.

Lo importante no es la cantidad de herramientas, sino que agregar una nueva no requiere modificar el ciclo principal.

El registro funciona como un diccionario que relaciona lo que el modelo quiere hacer con la función que realiza esa operación.

EXTENDED_DISPATCH: dict = {    "bash": lambda inp: run_bash(        inp["command"]    ),    "read": lambda inp: run_read(        inp["path"],        inp.get("start_line"),        inp.get("end_line")    ),    "write": lambda inp: run_write(        inp["path"],        inp["content"]    ),    "grep": lambda inp: run_grep(        inp["pattern"],        inp.get("path", ".")    ),    "glob": lambda inp: run_glob(        inp["pattern"]    ),    "revert": lambda inp: run_revert(        inp["path"]    ),}

Cada clave corresponde al nombre exacto que utilizará el modelo.

Cada valor es la función Python encargada de ejecutar la operación.

Cuando Claude necesita leer un archivo, el sistema realiza una búsqueda en el diccionario y ejecuta el controlador correspondiente.

No necesitamos una jerarquía compleja de clases ni múltiples condiciones para elegir las herramientas.

La operación fundamental puede reducirse a:

output = handler(tool_input)

La importancia de las descripciones

Las definiciones de herramientas son tan importantes como sus implementaciones.

El modelo utiliza la descripción para determinar qué herramienta debe llamar.

EXTENDED_TOOLS = BASIC_TOOLS + [    {        "name": "read",        "description": (            "Read a file and return numbered lines. "            "Use when you need to inspect file content "            "or reference specific line numbers. "            "Returns up to 50,000 characters. "            "Use start_line/end_line for large files."        ),        "input_schema": {            "type": "object",            "properties": {                "path": {"type": "string"},                "start_line": {"type": "integer"},                "end_line": {"type": "integer"}            },            "required": ["path"]        }    },    {        "name": "write",        "description": (            "Write content to a file. "            "Automatically snapshots the previous "            "content so you can revert. "            "Creates parent directories if needed."        ),        "input_schema": {            "type": "object",            "properties": {                "path": {"type": "string"},                "content": {"type": "string"}            },            "required": ["path", "content"]        }    },    {        "name": "grep",        "description": (            "Search for a regex pattern across files. "            "Returns file paths and line numbers of matches."        ),        "input_schema": {            "type": "object",            "properties": {                "pattern": {"type": "string"},                "path": {"type": "string"},                "recursive": {"type": "boolean"}            },            "required": ["pattern"]        }    },    {        "name": "glob",        "description": (            "Find files matching a glob pattern, "            "e.g. '**/*.py'. Returns sorted list "            "of matching paths."        ),        "input_schema": {            "type": "object",            "properties": {                "pattern": {"type": "string"}            },            "required": ["pattern"]        }    },    {        "name": "revert",        "description": (            "Restore a file to its state "            "before the last write call."        ),        "input_schema": {            "type": "object",            "properties": {                "path": {"type": "string"}            },            "required": ["path"]        }    }]

Una descripción mal redactada puede hacer que el modelo seleccione la herramienta incorrecta.

Por ejemplo, si grep se describe simplemente como una herramienta para buscar archivos y bash como una herramienta para ejecutar comandos, el modelo puede utilizar Bash para todas las búsquedas.

Las descripciones específicas permiten que el modelo relacione con mayor precisión la intención del usuario con las capacidades disponibles.

Controlar errores en los ejecutores

Las funciones que ejecutan herramientas deben mantener un contrato consistente: reciben parámetros, devuelven resultados y gestionan los errores sin interrumpir el ciclo principal.

def run_read(    path: str,    start_line: int = None,    end_line: int = None) -> str:    try:        with open(            path,            "r",            encoding="utf-8",            errors="replace"        ) as f:            lines = f.readlines()        s = (start_line or 1) - 1        e = end_line or len(lines)        numbered = "".join(            f"{s+1+i:4d}\t{l}"            for i, l in enumerate(lines[s:e])        )        return numbered[:50000] or "(empty file)"    except FileNotFoundError:        return f"Error: file not found: {path}"    except Exception as e:        return f"Error reading {path}: {e}"

La función devuelve líneas numeradas deliberadamente.

Esto permite al modelo identificar secciones concretas de un archivo y realizar modificaciones precisas.

Por ejemplo, puede utilizar las referencias de una lectura anterior para identificar qué líneas necesita reemplazar.

También podemos definir una función de búsqueda:

def run_grep(    pattern: str,    path: str = ".",    recursive: bool = True) -> str:    try:        flags = ["-r"] if recursive else []        r = subprocess.run(            ["grep", "-n", *flags, pattern, path],            capture_output=True,            text=True,            timeout=30        )        return (            (r.stdout + r.stderr).strip()            or "(no matches)"        )[:10000]    except Exception as e:        return f"Error: {e}"

Prueba del sistema de herramientas

Solicitamos al agente:

Encuentra todos los comentarios TODO del proyecto
e indica en qué archivos aparecen.

El modelo utiliza grep:

[grep] TODO

./agents/s03_todo_write.py:45:
# TODO: add priority levels to tasks

./agents/s09_agent_teams.py:112:
# TODO: replace JSONL mailboxes with Redis

El resultado identifica dos comentarios TODO en dos archivos diferentes.

El modelo eligió grep en lugar de bash porque la descripción de la herramienta permitía comprender claramente su función.

Esta es la finalidad del registro de ejecución: la decisión sobre qué herramienta utilizar pertenece al modelo, mientras que el harness proporciona capacidades bien definidas y se encarga de ejecutarlas.

3. Planificación con TodoWrite antes de ejecutar tareas

Uno de los comportamientos más interesantes que muestran los registros de ejecución de Claude Code es lo que ocurre antes de que el modelo empiece a trabajar en una tarea compleja.

En lugar de comenzar inmediatamente a modificar archivos, puede utilizar TodoWrite para establecer un plan de trabajo.

Primero se define el plan y después se ejecutan sus pasos.

Este comportamiento responde a un problema habitual de los agentes de IA: cuando trabajan en tareas de varios pasos sin una planificación explícita, pueden desviarse del objetivo original.

El agente empieza ejecutando una operación, encuentra un resultado intermedio interesante, decide investigarlo y termina trabajando en algo relacionado, pero diferente de lo que solicitó el usuario.

TodoWrite proporciona un mecanismo de seguimiento que permite mantener el objetivo visible durante toda la ejecución.

En la arquitectura descrita, el estado actual del plan se incorpora nuevamente al contexto después de las llamadas a herramientas. De esta manera, el modelo dispone de una referencia continua sobre las tareas pendientes y las completadas.

Implementación de TodoWrite

TODO_FILE = Path(".agent_todo.json")def todo_write(tasks: list) -> str:    data = [        {            "id": i,            "task": t,            "status": "pending"        }        for i, t in enumerate(tasks)    ]    TODO_FILE.write_text(        json.dumps(data, indent=2)    )    return "Plan written:\n" + "\n".join(        f"  [{i}] {t}"        for i, t in enumerate(tasks)    )def todo_read() -> str:    if not TODO_FILE.exists():        return "(no plan)"    data = json.loads(TODO_FILE.read_text())    return "\n".join(        f"  [{t['id']}] "        f"[{t['status']:12s}] {t['task']}"        for t in data    )def todo_update(index: int, status: str) -> str:

Estas tres herramientas funcionan conjuntamente:

todo_write registra el plan completo al inicio.

todo_update modifica el estado de cada tarea a medida que avanza el agente.

todo_read permite consultar el progreso en cualquier momento.

Juntas crean una memoria de trabajo externa que mantiene un registro del proceso de ejecución.

Cada paso tiene un estado persistente, por lo que el sistema puede comprobar qué actividades siguen pendientes.

Hacer obligatoria la planificación

Actualizamos el prompt de sistema para indicar al modelo que debe planificar antes de trabajar en tareas complejas.

SYSTEM = (    f"You are a coding agent at {os.getcwd()}. "    "Before working on any multi-step task, "    "ALWAYS call todo_write first "    "to write your complete plan. "    "Execute each step in order. "    "Call todo_update after completing each step.")

La palabra ALWAYS es importante en este ejemplo.

Sin una instrucción firme, el modelo puede decidir planificar en determinadas ocasiones y omitir ese paso en otras.

Con una instrucción explícita, el comportamiento tiende a ser más consistente.

Ejemplo de ejecución

Solicitamos:

Refactoriza la función run_bash para incorporar
un tiempo de espera configurable y un límite
en la longitud de la salida.

Después, comprueba que no se haya roto nada.

El agente comienza registrando cinco tareas:

[todo_write]

[0] Leer la implementación actual de run_bash
[1] Agregar un timeout configurable de 120 segundos
[2] Limitar la salida a 50.000 caracteres
[3] Guardar la implementación actualizada
[4] Ejecutar las pruebas

Después ejecuta los pasos previstos.

[read] core.py (lines 45-72)
[todo_update] 0 → done

[write] core.py
[todo_update] 1 → done
[todo_update] 2 → done
[todo_update] 3 → done

[bash] python -m pytest tests/test_agents_smoke.py -q

44 passed in 2.1s

[todo_update] 4 → done

El resultado final confirma que la refactorización terminó correctamente, se añadió el tiempo de espera configurable y la limitación de salida, y las 44 pruebas se completaron sin errores.

El modelo definió cinco pasos antes de modificar un solo archivo, los ejecutó y verificó los resultados.

La clave no es que el modelo sea naturalmente más cuidadoso, sino que la infraestructura le proporciona una estructura para trabajar de manera organizada.

4. Aislamiento del contexto mediante subagentes

Los registros de ejecución de Claude Code muestran otro comportamiento interesante cuando el sistema necesita explorar grandes bases de código.

En lugar de incorporar todos los archivos directamente a la conversación principal, el agente puede delegar la exploración en varios subagentes.

Cada subagente trabaja con su propio contexto y devuelve un resumen de los resultados.

La conversación principal no necesita recibir todas las lecturas de archivos, búsquedas y operaciones intermedias realizadas para obtener esas conclusiones.

Este patrón se conoce como aislamiento del contexto de subagentes.

Permite trabajar con repositorios grandes sin llenar la ventana de contexto principal con información secundaria.

La implementación consiste en proporcionar a cada subagente una lista messages[] completamente independiente.

El único estado que se comparte entre el agente principal y el subagente es el resultado final que devuelve este último.

Definir el comportamiento del subagente

SUBAGENT_SYSTEM = (    f"You are a subagent working on "    f"a specific subtask at {os.getcwd()}. "    "Complete your task thoroughly "    "and summarise your findings clearly. "    "Your output will be the only thing "    "the parent agent sees.")

Crear un subagente

def spawn_subagent(prompt: str) -> str:    """Run a fresh agent loop with isolated context."""    print(        f"\033[35m  [subagent spawned] "        f"{prompt[:60]}\033[0m"    )    sub_messages = [        {            "role": "user",            "content": prompt        }    ]    while True:        resp = client.messages.create(            model=MODEL,            system=SUBAGENT_SYSTEM,            messages=sub_messages,            tools=EXTENDED_TOOLS,            max_tokens=8000,        )        sub_messages.append({            "role": "assistant",            "content": resp.content        })        if resp.stop_reason != "tool_use":            break        results = dispatch_tools(            resp.content,            EXTENDED_DISPATCH        )

El subagente ejecuta el mismo ciclo básico que el agente principal.

También dispone de las mismas herramientas.

La diferencia es que su historial comienza desde cero y su prompt de sistema se concentra en una tarea específica.

Cuando termina, los resultados intermedios se descartan y únicamente se devuelve el resumen final.

Registrar la herramienta

Para permitir que el modelo decida cuándo crear un subagente, debemos registrarlo como una herramienta.

SUBAGENT_TOOLS = EXTENDED_TOOLS + [    {        "name": "spawn_subagent",        "description": (            "Spawn an isolated subagent "            "to handle a subtask in a fresh context. "            "Use for exploration tasks, risky operations, "            "or any work that would generate large "            "amounts of intermediate output "            "that the main conversation does not need. "            "Returns only the subagent's final summary."        ),        "input_schema": {            "type": "object",            "properties": {                "prompt": {                    "type": "string"                }            },            "required": ["prompt"]        }    }]

Actualizamos el registro de ejecución:

SUBAGENT_DISPATCH = {    **EXTENDED_DISPATCH,    "spawn_subagent": lambda inp: spawn_subagent(        inp["prompt"]    ),}

Prueba con un repositorio completo

Solicitamos:

Analiza toda la base de código y proporciona
un resumen completo de su arquitectura.

El agente principal crea un subagente para explorar los archivos Python.

El subagente encuentra 25 archivos y comienza a analizarlos.

[subagent] [glob] agents/**/*.py
           → 25 files found

[subagent] [read] agents/core.py
[subagent] [read] agents/s01_agent_loop.py
[subagent] [read] agents/s02_tool_use.py

... 22 lecturas adicionales

Después devuelve una conclusión:

core.py exporta el cliente, las herramientas
y los registros de ejecución.

Cada archivo de sesión importa la base
compartida y añade un mecanismo específico.

La conversación principal registra una sola llamada a herramienta, aunque el subagente realizó 25 operaciones.

Sin aislamiento, el contexto principal habría acumulado el contenido de todos esos archivos.

La ventaja es que el agente principal puede seguir razonando sobre la arquitectura general, en lugar de dedicar parte de su contexto a información de bajo nivel.

Fase 2. Gestión del conocimiento y del contexto

En esta fase incorporamos la infraestructura que permite al agente ir más allá de una única sesión.

El objetivo es cargar conocimientos especializados únicamente cuando se necesitan, comprimir el historial antes de que afecte a la calidad del razonamiento y conservar información persistente en disco.

Estos mecanismos permiten que el agente retome tareas después de reinicios y mantenga un contexto de trabajo útil durante sesiones prolongadas.

5. Carga de habilidades bajo demanda

Uno de los errores más costosos en Harness Engineering es introducir en el prompt de sistema toda la información que el modelo podría necesitar.

Imaginemos un prompt que contiene instrucciones sobre procesamiento de PDF, revisión de código, despliegues y auditorías de seguridad.

Ese prompt podría consumir miles de tokens en cada llamada a la API, aunque la mayor parte del contenido no tenga relación con la tarea actual.

Claude Code aborda este problema mediante un patrón conocido como divulgación progresiva, o progressive disclosure.

El prompt de sistema contiene únicamente descripciones breves de las habilidades disponibles.

Cuando el modelo identifica que necesita una habilidad especializada, llama a una función como load_skill() para incorporar las instrucciones completas al contexto.

De esta forma, el coste de contexto se produce únicamente cuando el conocimiento resulta relevante.

Es posible tener cientos de habilidades instaladas sin cargar todas sus instrucciones en cada interacción.

Estructura de las habilidades

skills/
├── agent-builder/
│   └── SKILL.md
├── code-review/
│   └── SKILL.md
└── pdf/
    └── SKILL.md

Una habilidad incluye metadatos que permiten identificarla y un conjunto de instrucciones que indican cómo realizar una tarea.

Por ejemplo:

---
name: code-review
description: Use when asked to review code,
             audit a file for bugs,
             check quality, or suggest improvements.
---

# Code Review Skill

## Review process

1. Read the full file before commenting
2. Categorise issues: BUG / SECURITY / PERF / STYLE / SUGGEST
3. Cite file and line number for every finding
4. End with a deploy-readiness summary

El mecanismo de descubrimiento explora el directorio de habilidades al iniciar el sistema.

Lee los metadatos de cada archivo SKILL.md y construye un registro ligero que posteriormente se incorpora al prompt de sistema.

Descubrir las habilidades disponibles

SKILLS_DIR = Path(__file__).parent.parent / "skills"def discover_skills() -> dict[str, str]:    skills = {}    for skill_dir in sorted(SKILLS_DIR.iterdir()):        skill_md = skill_dir / "SKILL.md"        if skill_dir.is_dir() and skill_md.exists():            lines = skill_md.read_text(                encoding="utf-8"            ).splitlines()            desc = ""            in_frontmatter = False            for line in lines:                if line.strip() == "---":                    in_frontmatter = not in_frontmatter                    continue                if (                    not in_frontmatter                    and line.strip()                    and not line.startswith("#")                ):                    desc = line.strip()[:100]                    break            skills[skill_dir.name] = (                desc or "No description."            )    return skills

Cargar una habilidad

def load_skill(name: str) -> str:    skill_path = SKILLS_DIR / name / "SKILL.md"    if not skill_path.exists():        return (            f"Error: skill '{name}' not found. "            "Use list_skills to see available."        )    return (        f"=== SKILL: {name} ===\n"        f"{skill_path.read_text(encoding='utf-8')}"    )

El prompt de sistema referencia las habilidades sin cargar todo su contenido.

_skill_list = "\n".join(    f"  - {n}: {d}"    for n, d in discover_skills().items())SYSTEM = (    f"You are a coding agent at {os.getcwd()}.\n"    "When a task requires specialised knowledge, "    "call load_skill(name) to get full instructions "    "before starting. Do NOT guess.\n\n"    f"Available skills:\n{_skill_list}")

Prueba del sistema de habilidades

Solicitamos:

Revisa agents/core.py para identificar errores,
problemas de seguridad y calidad del código.

El agente comienza cargando la habilidad correspondiente.

[load_skill] code-review

Después lee el archivo y realiza búsquedas concretas.

[read] agents/core.py

[grep] def run_bash
[grep] subprocess.run

El análisis identifica tres problemas.

Seguridad: la utilización de shell=True con entradas no validadas puede permitir la inyección de comandos.

Rendimiento: la función run_grep crea un proceso nuevo para cada búsqueda, sin aprovechar resultados anteriores.

Estilo: run_read puede devolver una cadena vacía para determinados archivos, en lugar de ofrecer una respuesta descriptiva.

El resultado concluye que existe un problema de seguridad que debe resolverse antes del despliegue.

La habilidad no aumenta por sí sola la capacidad del modelo. Su función es proporcionar un procedimiento consistente para revisar código y presentar resultados con una estructura definida.

6. Compresión del contexto en tres niveles

Todas las sesiones prolongadas encuentran el mismo problema: la ventana de contexto termina llenándose.

Las salidas de herramientas, resultados intermedios e interacciones anteriores ocupan espacio que podría utilizarse para información relevante de la tarea actual.

El artículo describe un mecanismo de compresión de Claude Code que se activa automáticamente cuando se alcanza aproximadamente el 92 % de utilización de la ventana de contexto.

En lugar de descartar completamente el historial, el sistema lo resume para conservar la información importante mientras reduce el consumo de tokens.

El resumen también puede escribirse en un archivo Markdown persistente.

Funcionamiento de la compresión

La implementación utiliza tres niveles.

Primero conserva literalmente los mensajes recientes porque contienen el contexto activo.

Después comprime las interacciones anteriores en un resumen generado mediante una llamada al modelo.

Finalmente guarda ese resumen en .agent_memory.md, permitiendo que una nueva sesión recupere información importante.

Configuración inicial

COMPRESS_THRESHOLD = 40_000KEEP_RECENT = 6MEMORY_FILE = Path(".agent_memory.md")

En este ejemplo, el umbral se expresa en caracteres y equivale aproximadamente a 10.000 tokens estimados.

Se conservan los últimos seis mensajes sin modificaciones.

Estimar el tamaño del contexto

def _estimate_size(messages: list) -> int:    total = 0    for msg in messages:        content = msg.get("content", "")        if isinstance(content, str):            total += len(content)        elif isinstance(content, list):            for block in content:                if isinstance(block, dict):                    total += len(                        str(                            block.get("text", "")                            or block.get("content", "")                        )                    )                elif hasattr(block, "text"):                    total += len(block.text or "")    return total

Generar un resumen

def _summarize(messages: list) -> str:    text = "\n\n".join(        f"[{m['role']}]: " + (            m["content"]            if isinstance(m["content"], str)            else " ".join(                b.get("text", "")                if isinstance(b, dict)                else getattr(b, "text", "")                for b in (                    m["content"]                    if isinstance(m["content"], list)                    else []                )            )        )        for m in messages    )    resp = client.messages.create(        model=MODEL,        system=(            "Summarise this conversation. "            "Keep all important decisions, "            "code changes, file paths, and context. "            "Be concise but complete."        ),        messages=[            {                "role": "user",                "content": f"Summarise:\n\n{text[:20000]}"            }        ],        max_tokens=2000,    )

La función de resumen mantiene las decisiones importantes, las modificaciones de código, las rutas de archivos y el contexto necesario para continuar.

Activar la compresión cuando sea necesaria

def maybe_compress(messages: list) -> bool:    if _estimate_size(messages) < COMPRESS_THRESHOLD:        return False    if len(messages) <= KEEP_RECENT:        return False    old_messages = messages[:-KEEP_RECENT]    recent = messages[-KEEP_RECENT:]    print(        "[compress] context large - "        "summarising older messages..."    )    summary = _summarize(old_messages)    MEMORY_FILE.write_text(        f"# Agent Memory\n"        f"Generated: {datetime.now().isoformat()}\n\n"        f"{summary}\n"    )    messages.clear()    messages.append({        "role": "user",        "content": (            "[Context summary from previous conversation]"            f"\n\n{summary}"        )    })    messages.append({        "role": "assistant",        "content": (            "Understood. I have the full context "            "from our previous work."        )    })    messages.extend(recent)    return True

La compresión no se activa por tiempo transcurrido, sino por el tamaño estimado del contexto.

Recuperar memoria al iniciar una sesión

history = []if MEMORY_FILE.exists():    mem = MEMORY_FILE.read_text()    print(        f"[memory] loaded from {MEMORY_FILE}"    )    history = [        {            "role": "user",            "content": f"[Loaded memory]\n\n{mem}"        },        {            "role": "assistant",            "content": "Memory loaded. Ready to continue."        }    ]

Resultado de la prueba

El agente recibe una tarea para agregar gestión de errores a las funciones de core.py y crear pruebas.

Durante la ejecución lee archivos, escribe nuevas implementaciones y ejecuta pruebas.

Después de acumular suficiente historial, se activa la compresión.

[compress] context large -
summarising older messages...

[compress] 18 messages → 1 summary

saved to .agent_memory.md

Los 18 mensajes anteriores se convierten en un único bloque de contexto.

Cuando comienza otra sesión, ese resumen se recupera desde el archivo persistente.

El agente puede continuar trabajando sin necesitar que todas las interacciones anteriores se vuelvan a introducir en cada llamada a la API.

7. Grafo persistente de dependencias entre tareas

La compresión del contexto y el seguimiento de tareas resuelven problemas diferentes.

La compresión determina qué información conserva el modelo.

El grafo de tareas determina qué trabajos debe realizar el agente, cuáles ha completado y qué dependencias existen entre ellos.

A diferencia de un plan limitado a una sesión, el grafo permite mantener el estado del proyecto después de cerrar procesos, reiniciar sesiones o incluso reiniciar el equipo.

Cada tarea dispone de un identificador, una descripción, un estado, una prioridad y una lista de tareas que deben completarse previamente.

El grafo se almacena en .agent_tasks.json.

Configurar el almacenamiento

TASKS_FILE = Path(".agent_tasks.json")_TASKS_LOCK = threading.Lock()def _load() -> list:    if TASKS_FILE.exists():        return json.loads(TASKS_FILE.read_text())    return []def _save(tasks: list):    TASKS_FILE.write_text(        json.dumps(tasks, indent=2)    )

Crear tareas con dependencias

def run_task_create(    description: str,    depends_on: list = None,    priority: str = "medium") -> str:    with _TASKS_LOCK:        tasks = _load()        task = {            "id": uuid.uuid4().hex[:8],            "description": description,            "status": "pending",            "priority": priority,            "depends_on": depends_on or [],            "result": "",        }        tasks.append(task)        _save(tasks)    return (        f"created task {task['id']}: "        f"{description}"    )

Seleccionar la siguiente tarea disponible

def run_task_next() -> str:    tasks = _load()    done_ids = {        t["id"]        for t in tasks        if t["status"] == "done"    }    priorities = {        "high": 0,        "medium": 1,        "low": 2    }    for t in sorted(        tasks,        key=lambda x: priorities[x["priority"]]    ):        if t["status"] != "pending":            continue        if all(            dep in done_ids            for dep in t.get("depends_on", [])        ):            return (                f"[{t['id']}] "                f"[{t['priority']}] "                f"{t['description']}"            )    return "(no unblocked tasks available)"

Actualizar el estado de una tarea

def run_task_update(    task_id: str,    status: str,    result: str = "") -> str:    with _TASKS_LOCK:        tasks = _load()        for t in tasks:            if t["id"].startswith(task_id):                t["status"] = status                if result:                    t["result"] = result                _save(tasks)                return (                    f"task {t['id']} → {status}"                )    return f"Error: task '{task_id}' not found"

El bloqueo mediante threading.Lock() es importante porque varios agentes podrían intentar modificar el grafo simultáneamente.

Sin un mecanismo de sincronización, dos agentes podrían leer el mismo estado, modificarlo de manera independiente y sobrescribir accidentalmente los cambios del otro.

El bloqueo protege las operaciones concurrentes de lectura y escritura dentro del mismo proceso.

Ejemplo: migrar las herramientas a ejecución asíncrona

Solicitamos al agente:

Necesitamos migrar toda la base de código
para utilizar implementaciones asíncronas.

Divide el trabajo en tareas, identifica
las dependencias y comienza la ejecución.

El agente crea el siguiente conjunto de tareas:

Auditar todas las funciones síncronas de core.py.

Crear versiones asíncronas de run_bash, run_read y run_write.

Crear versiones asíncronas de run_grep y run_glob.

Actualizar el registro ASYNC_DISPATCH.

Modificar s18_parallel_tools.py.

Ejecutar todas las pruebas.

Las tareas de implementación dependen de la auditoría inicial.

La actualización del registro depende de que las implementaciones asíncronas estén disponibles.

Las pruebas finales dependen de que las modificaciones anteriores hayan terminado.

El agente ejecuta los trabajos respetando este orden.

[task_next]
[a3f2c891] Audit sync functions

[task_update] a3f2c891 → done

[task_next]
[b7d1e445] Write async bash/read/write

[task_update] b7d1e445 → done

[task_next]
[c9f3a112] Write async grep/glob

[task_update] c9f3a112 → done

[task_next]
[d2e8b334] Update ASYNC_DISPATCH

[task_update] d2e8b334 → done

[bash] python -m pytest tests/ -q

44 passed in 2.3s

El resultado confirma que las seis tareas se completaron correctamente y las 44 pruebas finalizaron sin errores.

Si el proceso se hubiera detenido después de la tercera tarea, al reiniciarse podría recuperar el grafo y continuar desde el siguiente trabajo pendiente.

Esa persistencia es la principal diferencia entre un plan de una sola sesión y un sistema de gestión de proyectos basado en dependencias.

Fase 3. Ejecución asíncrona y equipos multiagente

La tercera fase aborda las limitaciones de un único agente y una sola ventana de contexto.

Cuando una tarea requiere ejecutar procesos lentos, analizar varios componentes o modificar múltiples archivos, un solo hilo de ejecución puede convertirse en un cuello de botella.

Esta fase incorpora ejecución en segundo plano, agentes especializados persistentes, protocolos de comunicación, asignación autónoma de tareas y aislamiento de archivos mediante Git Worktrees.

8. Ejecución de tareas en segundo plano con notificaciones

Cuando Claude ejecuta una batería de pruebas, compila un proyecto o realiza una migración de base de datos, no siempre necesita permanecer inactivo hasta recibir el resultado.

Puede iniciar una operación en segundo plano, continuar trabajando y recibir una notificación cuando termine.

Sin esta capacidad, un agente queda limitado por la duración de su llamada a herramienta más lenta.

Por ejemplo, si ejecutar las pruebas requiere 45 segundos, un sistema bloqueante debe esperar ese tiempo antes de realizar cualquier otra operación.

La ejecución en segundo plano permite separar el trabajo operativo del ciclo de razonamiento del agente.

Crear una cola de notificaciones

import threadingimport queueimport subprocess_BG_QUEUE: queue.Queue = queue.Queue()

Ejecutar comandos en segundo plano

def run_bash_background(    command: str,    label: str = "") -> str:    label = label or command[:40]    def _run():        print(f"[bg] started: {label}")        try:            r = subprocess.run(                command,                shell=True,                capture_output=True,                text=True,                timeout=300,                cwd=os.getcwd()            )            output = (                (r.stdout + r.stderr).strip()[:2000]                or "(no output)"            )            status = "completed"        except subprocess.TimeoutExpired:            output = (                "Error: operation timed out after 300s"            )            status = "timed out"        except Exception as e:            output = f"Error: {e}"            status = "failed"        _BG_QUEUE.put(            f"[Background task '{label}' "            f"{status}]\n{output}"        )        print(f"[bg] {label} {status}")    thread = threading.Thread(        target=_run,        daemon=True    )    thread.start()    return (        f"Background task started: '{label}'. "        "You will be notified when it completes."    )

La implementación utiliza un hilo por operación y una cola compartida para almacenar las notificaciones.

Cuando una tarea termina, introduce su resultado en la cola.

El ciclo principal puede consultar esa cola e incorporar los resultados al historial de conversación.

Procesar las notificaciones

def agent_loop_with_bg(messages: list):    stream_loop(        messages,        BG_TOOLS,        BG_DISPATCH,        system=SYSTEM    )    while not _BG_QUEUE.empty():        notification = _BG_QUEUE.get_nowait()        print("[bg] notification injected")        messages.append({            "role": "user",            "content": notification        })        stream_loop(            messages,            BG_TOOLS,            BG_DISPATCH,            system=SYSTEM        )

El agente dispone ahora de una herramienta denominada run_bash_background().

Puede utilizarla cuando necesita iniciar una operación prolongada sin bloquear el hilo principal.

Ejemplo de ejecución simultánea

Solicitamos:

Ejecuta todas las pruebas en segundo plano.

Mientras se ejecutan, añade docstrings
a las funciones de core.py que no tengan
documentación.

El agente inicia las pruebas:

[bash_background]
python -m pytest tests/ -v --tb=short

Background task started: 'full test suite'

Mientras se ejecutan, lee el archivo, identifica las funciones sin documentación y añade los docstrings correspondientes.

Posteriormente recibe la notificación:

[bg] full test suite completed

44 passed in 2.1s

El resultado indica que se documentaron ocho funciones y que todas las pruebas se completaron correctamente.

La ejecución concurrente permite que el tiempo total se aproxime a la duración de la operación más lenta, en lugar de sumar obligatoriamente los tiempos de ambas.

9. Agentes persistentes con buzones JSONL

Los subagentes convencionales suelen crearse para realizar una tarea específica y desaparecer cuando terminan.

Sin embargo, los proyectos de ingeniería también se benefician de contar con especialistas que mantengan su contexto durante varias tareas.

Por ejemplo, podemos utilizar un agente especializado en explorar bases de código y otro dedicado a escribir y modificar archivos.

Los agentes persistentes conservan información sobre el proyecto durante las diferentes actividades que reciben.

Arquitectura de comunicación

Cada integrante del equipo ejecuta su propio ciclo de agente en un hilo de segundo plano.

Dispone de una especialización y de un archivo JSONL que funciona como bandeja de entrada.

El agente principal envía tareas escribiendo mensajes en el buzón del especialista.

El especialista lee el mensaje, ejecuta el trabajo y devuelve el resultado mediante el buzón del agente principal.

Esta comunicación es asíncrona: el agente principal puede continuar trabajando mientras sus colaboradores ejecutan otras tareas.

Configurar los buzones

MAILBOX_DIR = Path(".mailboxes")MAILBOX_DIR.mkdir(exist_ok=True)TEAMMATES = {    "explorer": (        f"You are an explorer agent specialising "        f"in reading and understanding codebases "        f"at {os.getcwd()}. "        "You excel at mapping architecture, "        "finding patterns, and understanding "        "how components connect. "        "Use read, glob, grep."    ),    "writer": (        f"You are a writer agent specialising "        f"in creating and editing code "        f"at {os.getcwd()}. "        "You excel at implementing changes cleanly, "        "following existing patterns, and writing tests. "        "Use write, read, bash."    )}

Enviar y recibir mensajes

def _inbox(name: str) -> Path:    return MAILBOX_DIR / f"{name}.jsonl"def _send(to: str, frm: str, body: str):    with open(_inbox(to), "a") as f:        f.write(            json.dumps({                "from": frm,                "body": body            }) + "\n"        )def _receive(name: str) -> list[dict]:    path = _inbox(name)    if not path.exists():        return []    messages = [        json.loads(line)        for line in path.read_text().splitlines()        if line.strip()    ]    path.write_text("")    return messages

El sistema de buzones es deliberadamente sencillo.

Cada mensaje incluye el remitente y su contenido, y se almacena como una línea JSON.

Cuando el agente recibe los mensajes, vacía el archivo para evitar procesarlos nuevamente.

Ejecutar un agente persistente

def _run_teammate(    name: str,    system: str,    stop_event: threading.Event):    print(f"[{name}] ready and listening")    while not stop_event.is_set():        for msg in _receive(name):            print(                f"[{name}] received: "                f"{msg['body'][:60]}"            )            sub_messages = [                {                    "role": "user",                    "content": msg["body"]                }            ]            while True:                resp = client.messages.create(                    model=MODEL,                    system=system,                    messages=sub_messages,                    tools=EXTENDED_TOOLS,                    max_tokens=4000                )                sub_messages.append({                    "role": "assistant",                    "content": resp.content

El agente comprueba periódicamente su buzón y procesa las instrucciones recibidas.

Después devuelve el resultado al remitente.

Prueba de colaboración multiagente

La tarea consiste en analizar la arquitectura de comunicación entre agentes y añadir registros estructurados a las funciones correspondientes.

El agente principal delega la exploración al especialista explorer.

Este identifica cinco funciones relacionadas con la comunicación en dos archivos.

Después, el especialista writer utiliza esa información para añadir registros con fecha, remitente, destinatario, longitud del mensaje y estado de entrega.

Los dos especialistas colaboran sin incorporar todas sus operaciones intermedias al contexto del agente principal.

La arquitectura permite separar la investigación, la implementación y la síntesis de resultados.

10. Protocolo de comunicación mediante máquinas de estados finitos

Cuando varios agentes intercambian mensajes simultáneamente, pueden aparecer condiciones de carrera y bloqueos.

Un agente podría enviar una segunda solicitud antes de recibir la respuesta de la primera.

También es posible que dos agentes queden esperando indefinidamente una respuesta del otro.

Para evitar estos problemas necesitamos establecer reglas sobre cuándo pueden comunicarse los participantes.

Una máquina de estados finitos, conocida como FSM por sus siglas en inglés, permite controlar esas transiciones.

Los cuatro estados del agente

En nuestra implementación, cada agente puede encontrarse en uno de estos estados:

IDLE: disponible, sin una operación activa.

REQUESTING: enviando una solicitud.

WAITING: esperando una respuesta.

RESPONDING: procesando una solicitud recibida.

La regla fundamental del protocolo es que un agente no puede iniciar una nueva solicitud mientras permanece en el estado WAITING.

Esta restricción evita determinados conflictos asociados a solicitudes simultáneas.

Implementación del protocolo

from enum import Enumclass AgentState(Enum):    IDLE = "idle"    REQUESTING = "requesting"    WAITING = "waiting"    RESPONDING = "responding"class ProtocolAgent:    def __init__(self, name: str, system: str):        self.name = name        self.system = system        self.state = AgentState.IDLE        self._inbox = (            MAILBOX_DIR / f"{name}_proto.jsonl"        )        self._lock = threading.Lock()    def send(        self,        to: "ProtocolAgent",        message: str    ):        with self._lock:            if self.state == AgentState.WAITING:                return (                    "Error: cannot send while "                    "waiting for a response"                )            self.state = AgentState.REQUESTING

Probar el protocolo

Solicitamos analizar la cobertura de pruebas de core.py y crear pruebas adicionales para las funciones que no alcancen un 80 % de cobertura de ramas.

El primer agente analiza la cobertura:

El resultado muestra una cobertura del 71 % e identifica tres funciones con ramas insuficientemente probadas.

El segundo agente recibe la tarea de escribir pruebas para esos casos:

Las transiciones de estado se registran durante todo el proceso.

El protocolo impide que un agente envíe una segunda solicitud mientras espera la respuesta de la primera.

La coordinación queda incorporada a la arquitectura y no depende únicamente de que el modelo recuerde cómo debe comunicarse.

11. Asignación autónoma de tareas

El protocolo FSM controla la comunicación, pero todavía requiere que un agente principal distribuya el trabajo.

En proyectos grandes, el coordinador también puede convertirse en un cuello de botella.

La asignación autónoma elimina la necesidad de que un agente central asigne cada tarea.

Cada integrante examina periódicamente el grafo compartido y, cuando encuentra una tarea disponible, intenta reservarla para ejecutarla.

La reserva debe ser atómica para impedir que dos agentes seleccionen el mismo trabajo simultáneamente.

Reclamar una tarea de manera segura

El principio consiste en proteger las operaciones de selección y modificación del estado mediante un bloqueo.

De esta forma, el primer agente que obtiene el bloqueo puede reservar la tarea.

Cuando el siguiente agente consulta el grafo, esa tarea ya no figura como pendiente.

Esto evita ejecuciones duplicadas y reduce la posibilidad de conflictos.

Ejecución mediante trabajadores autónomos

Cada trabajador mantiene un ciclo en el que consulta las tareas disponibles.

Cuando encuentra una, la reclama y ejecuta el trabajo utilizando su propio contexto y las herramientas registradas.

Después actualiza el resultado en el grafo compartido.

La asignación autónoma permite distribuir tareas sin que el agente principal tenga que intervenir constantemente.

También facilita aumentar o reducir el número de trabajadores disponibles según la carga del proyecto.

El principio sigue siendo el mismo: el modelo decide cómo resolver cada tarea, pero la infraestructura controla las reservas y los cambios de estado.

12. Aislamiento de tareas mediante Git Worktrees

La asignación autónoma permite que varios agentes trabajen simultáneamente, pero genera un problema adicional.

¿Qué ocurre cuando dos agentes necesitan modificar archivos del mismo repositorio?

Si todos comparten el mismo directorio de trabajo, sus operaciones pueden interferir entre sí.

Un agente podría modificar un archivo mientras otro está leyendo una versión anterior.

También podrían producirse sobrescrituras o cambios difíciles de integrar.

La solución consiste en utilizar Git Worktrees.

Un Worktree permite crear un directorio de trabajo independiente asociado al mismo repositorio Git.

Cada agente dispone de su propia rama y de sus propios archivos, evitando que las modificaciones paralelas afecten directamente a los demás trabajadores.

Crear un Worktree para una tarea

Podemos crear un Worktree de Git aislado para cada tarea mediante la siguiente función:

def create_worktree(task_id: str) -> tuple[str, str]:    """Create an isolated git worktree for a task."""    branch = f"task/{task_id}"    path = str(        Path(os.getcwd()).parent        / f".worktree-{task_id[:8]}"    )    if Path(path).exists():        shutil.rmtree(path, ignore_errors=True)        _git("worktree", "remove", "--force", path)    # Gestionar conflictos con ramas de ejecuciones anteriores    if _branch_exists(branch):        _git("branch", "-D", branch)    rc, _, err = _git(        "worktree", "add", "-b", branch, path    )    if rc != 0:        raise RuntimeError(            f"Failed to create worktree: {err}"        )    return path, branch

La función crea un directorio de trabajo independiente asociado a una rama específica de Git.

El nombre de la rama se construye a partir del identificador de la tarea. Si existe un directorio anterior, se intenta eliminar antes de crear uno nuevo.

También se comprueba si ya existe una rama con ese nombre para evitar conflictos con ejecuciones anteriores.

Ejecutar un agente dentro de un Worktree

def run_task_in_worktree(task: dict) -> str:    """Run an agent loop in an isolated worktree directory."""    task_id = task["id"]    wt_path, wt_branch = create_worktree(task_id)    system = (        f"You are a coding agent working in isolated directory: {wt_path}. "        f"Task: {task['description']}. "        "Your changes are on a separate git branch - you cannot affect other agents. "        "When done, summarise every file you changed."    )    messages = [        {            "role": "user",            "content": task["description"]        }    ]    try:        while True:            resp = client.messages.create(

La función ejecuta el ciclo completo de un agente dentro del entorno de trabajo aislado.

Cada agente tiene acceso a las mismas herramientas, pero trabaja en una rama independiente.

Cuando termina, la infraestructura elimina el Worktree y realiza las operaciones de limpieza correspondientes, incluso si la ejecución falla.

Prueba: dos agentes modificando el mismo archivo

Podemos probar este mecanismo creando dos tareas que necesiten modificar core.py y ejecutando dos agentes simultáneamente.

Cada agente creará su propio Worktree, realizará cambios en core.py de manera aislada y ejecutará sus pruebas.

Cuando ambos terminen, la infraestructura detectará que las dos ramas modificaron el mismo archivo y señalará la necesidad de una revisión humana antes de fusionarlas.

La instrucción utilizada es:

s12 >> Ejecuta dos tareas en paralelo:

Tarea A:
Añadir validación de entrada a todas las
funciones de herramientas en core.py.

Tarea B:
Añadir instrucciones de registro a todas las
funciones de herramientas en core.py.

Resultado de la ejecución

> Thinking...

Running 2 tasks in parallel worktrees...

[task-a] worktree created:
.worktree-a1b2c3d4
(branch: task/a1b2c3d4)

[task-b] worktree created:
.worktree-e5f6a7b8
(branch: task/e5f6a7b8)

[a1b2c3][read] agents/core.py
[e5f6a7][read] agents/core.py

[a1b2c3][write] agents/core.py
(added input validation to run_bash)

[e5f6a7][write] agents/core.py
(added logging to run_bash)

[a1b2c3][write] agents/core.py
(added input validation to run_read)

[e5f6a7][write] agents/core.py
(added logging to run_read)

[a1b2c3][bash]
python -m pytest tests/ -q
→ 44 passed

[e5f6a7][bash]
python -m pytest tests/ -q
→ 44 passed

[task-a] completed
[task-b] completed

Conflict detection:

Both branches modified: agents/core.py

Overlap detected on 1 file -
human review required before merging.

Resultados

Tarea A: se añadió validación de entrada a seis funciones de herramientas. Todas las pruebas se completaron correctamente.

Tarea B: se añadieron registros de actividad a seis funciones de herramientas. Todas las pruebas se completaron correctamente.

Se detectó un conflicto de integración: ambas tareas modificaron core.py en ramas diferentes.

Los cambios son compatibles, pero requieren una revisión manual antes de integrarse.

Las ramas task/a1b2c3d4 y task/e5f6a7b8 están listas para su revisión, según el resultado mostrado en el artículo.

¿Qué demuestra este ejemplo?

Ambos agentes leyeron y modificaron core.py simultáneamente.

Sin embargo, como trabajaban en Worktrees separados, las escrituras de uno no interfirieron con las del otro.

Cada agente ejecutó sus pruebas sobre su propia rama y obtuvo resultados correctos.

La detección de conflictos se ejecutó después de que ambas tareas terminaran y señaló correctamente que una persona debe revisar la integración antes de incorporar cualquiera de las ramas a main.

Este es el modelo de aislamiento más sólido del repositorio: ejecución paralela sin interferencias entre los archivos durante el desarrollo de las tareas.

Gracias por leer Código en Casa.
Si esto te a ayudado y te sumo algo Dale un 👏 , compártelo con tu red o dejame un comentario para saber tu opinión.