Guía de Migración para Cortex — CrewMaster Legacy a v2.0.0¤

Audiencia: Coding agent del equipo Cortex Versión destino: CrewMaster v2.0.0 Ruptura: Limpia — sin adaptadores, sin backward compatibility

Tabla de Contenidos¤

Bundle → Operation
ReviewStep → Operation
AgentWithBundle → AgentConfig + Operation
Team graph → ExecutionPlan
Crew/CrewRouter → HTTP de aplicación
ContextTransformer → ContextProjector
PromptEngine aplicación → PromptEngine librería
Skill → Capability / Tool
RunnableConfig → ExecutionConfig

1. Bundle → Operation¤

Resumen¤

Bundle agrupaba Specialty instances bajo un mismo output schema y manejaba dependencias vía BundleDependency. En v2.0.0, el concepto se descompone en Operation con sub_operations para partes atómicas y consumes basado en tipos para dependencias.

Antes (legacy)¤

from scouters_crew.agent_with_bundle import (
    Bundle, BundleGroup, BundleDependency, SpecialtyWithJobs
)
from scouters_crew.router.prompt_engine import PromptComposition

# Definición de bundles con dependencias explícitas por nombre
bundles_to_create = [
    Bundle[CorporateColors].from_output_model(
        name="corporate_colors",
        output=CorporateColors,
        prompt_system=PromptComposition(blocks=[
            "blocks://agent/task.j2",
            "blocks://agent/colors/methodology.j2",
        ]),
        prompt_user=PromptComposition(blocks=["blocks://user.j2"]),
        specialty_implemented=SpecialtyWithJobs,
        deps=[],
    ),
    Bundle[Typography].from_output_model(
        name="typography",
        output=Typography,
        prompt_system=PromptComposition(blocks=[
            "blocks://agent/task.j2",
            "blocks://agent/typography/methodology.j2",
        ]),
        prompt_user=PromptComposition(blocks=["blocks://user.j2"]),
        specialty_implemented=SpecialtyWithJobs,
        deps=[
            BundleDependency(
                name="corporate_colors",
                context_model=CorporateColorsContext,
                composer=CorporateColorsComposer.compose,
            )
        ],
    ),
    Bundle[VisualIdentity].from_output_model(
        name="visual_identity",
        output=VisualIdentity,
        prompt_system=PromptComposition(blocks=[
            "blocks://agent/task.j2",
            "blocks://agent/visual_identity/methodology.j2",
        ]),
        prompt_user=PromptComposition(blocks=["blocks://user.j2"]),
        specialty_implemented=SpecialtyWithJobs,
        deps=[
            BundleDependency(name="corporate_colors", ...),
            BundleDependency(name="typography", ...),
        ],
    ),
]

bundle_group = BundleGroup(bundles=bundles_to_create)

El Bundle.from_output_model() creaba automáticamente un Specialty por cada campo del modelo de salida. La ejecución recorría los specialties secuencialmente, compartiendo estado vía WorkInProgress. Las dependencias entre bundles se declaraban por nombre vía BundleDependency con funciones composer manuales.

Después (v2.0.0)¤

from pydantic import BaseModel
from crewmaster.operations.operation import Operation
from crewmaster.agents.agent import AgentConfig

# Cada Bundle se convierte en una Operation hoja.
# Las dependencias entre bundles se convierten en consumes basado en tipos.
# Los specialties (antes creados por from_output_model) se convierten en
# sub_operations o se colapsan en un solo Operation si el modelo es simple.

corporate_colors_op = Operation(
    name="corporate_colors",
    produces=CorporateColors,
    kind="artifact",
    agent=AgentConfig(
        name="paula",
        identity_blocks=["blocks://agent/identity.j2"],
        default_context=[BrandContextProjector],
    ),
    task_blocks=[
        "blocks://agent/task.j2",
        "blocks://agent/colors/methodology.j2",
    ],
    consumes=[BrandContextProjector],
)

typography_op = Operation(
    name="typography",
    produces=Typography,
    kind="artifact",
    agent=AgentConfig(
        name="paula",
        identity_blocks=["blocks://agent/identity.j2"],
        default_context=[BrandContextProjector],
    ),
    task_blocks=[
        "blocks://agent/task.j2",
        "blocks://agent/typography/methodology.j2",
    ],
    # consumes basado en TIPOS, no en nombres de string
    consumes=[BrandContextProjector, CorporateColors],
)

visual_identity_op = Operation(
    name="visual_identity",
    produces=VisualIdentity,
    kind="artifact",
    agent=AgentConfig(
        name="paula",
        identity_blocks=["blocks://agent/identity.j2"],
        default_context=[BrandContextProjector],
    ),
    task_blocks=[
        "blocks://agent/task.j2",
        "blocks://agent/visual_identity/methodology.j2",
    ],
    consumes=[BrandContextProjector, CorporateColors, Typography],
)

# La operación raíz compone las hojas
brand_generation = Operation(
    name="brand_generation",
    produces=BrandResult,
    kind="artifact",
    sub_operations=[
        corporate_colors_op,
        typography_op,
        visual_identity_op,
    ],
    # No tiene agent propio — delega a sub_operations
)

Notas sobre cambios de comportamiento¤

Legacy	v2.0.0
`Bundle.from_output_model()` crea specialties por cada campo	Los campos del modelo se manejan vía `output_schema` en `RuntimeRequest`; el LLM recibe el schema como structured output
`BundleDependency(name=..., composer=...)` vincula por string	`consumes=[Type]` vincula por tipo Python — el resolver matchea `produces` de otro nodo
`WorkInProgress` compartido entre specialties	Los artefactos se pasan vía el namespace `deps` del ExecutionPlan
`BundleGroup` orquesta bundles	`Operation.sub_operations` forma el árbol; `resolve_plan()` construye el DAG
`Bundle.specialties` son unidades de ejecución	Cada campo de un modelo de salida se maneja como structured output del LLM, no como nodo separado

Pasos de verificación¤

Identificar cada Bundle en el código legacy.
Convertir cada Bundle a una Operation hoja con produces=BundleOutput.
Reemplazar BundleDependency por consumes=[Type] usando el tipo de salida del bundle upstream.
Agrupar las operaciones relacionadas bajo una Operation raíz con sub_operations.
Ejecutar resolve_plan(root_operation, context_store, block_store, default_runtime) y verificar que no arroja PlanBuildError.
Verificar que el orden topológico del plan refleja las dependencias esperadas.

2. ReviewStep → Operation¤

Resumen¤

ReviewStep declaraba pasos de revisión en PeriodicAgent con un output model, un prompt, y validación de reglas de negocio. En v2.0.0, cada ReviewStep es un Operation(kind="cognition") independiente.

Antes (legacy)¤

from scouters_crew.review_steps import ReviewStep, BusinessRuleError
from scouters_crew.router.prompt_engine import PromptComposition

class ReviewPendingDirectives(ReviewStep[PendingDirectivesOutput]):
    name: str = "review_pending_directives"
    prompt: PromptComposition = PromptComposition(blocks=[
        "blocks://periodic/review_pending_directives/system.j2",
        "blocks://periodic/shared/context_rules.j2",
    ])
    output_model: type[PendingDirectivesOutput] = PendingDirectivesOutput
    mode: Literal["free_form", "structured_only"] = "free_form"
    terminal_skill: SkillStructuredResponse = submit_directives_skill

    async def validate_business_rules(
        self, domain_data: dict, biz_deps: dict, memory: BrandMemory
    ) -> list[BusinessRuleError]:
        errors = []
        valid_ids = biz_deps.get("valid_ids", {}).get("initiative", set())
        for directive in domain_data.get("directives", []):
            if directive.get("initiative_id") not in valid_ids:
                errors.append(BusinessRuleError(
                    field="initiative_id",
                    message=f"Invalid initiative ID: {directive.get('initiative_id')}"
                ))
        return errors

class SynthesizeLearning(ReviewStep[LearningOutput]):
    name: str = "synthesize_learning"
    prompt: PromptComposition = PromptComposition(blocks=[
        "blocks://periodic/synthesize_learning/system.j2",
    ])
    output_model: type[LearningOutput] = LearningOutput
    mode: Literal["free_form", "structured_only"] = "structured_only"
    terminal_skill: SkillStructuredResponse = submit_learning_skill

Cada ReviewStep era un paso dentro de un agente periódico. Las dependencias entre pasos eran implícitas — el agente iteraba sobre una lista de steps. La validación de reglas de negocio era un método de instancia.

Después (v2.0.0)¤

from pydantic import BaseModel
from crewmaster.operations.operation import Operation
from crewmaster.agents.agent import AgentConfig

class PendingDirectivesOutput(BaseModel):
    directives: list[DirectiveItem]

class LearningOutput(BaseModel):
    insights: list[InsightItem]

reviewer_agent = AgentConfig(
    name="periodic_reviewer",
    identity_blocks=["blocks://periodic/reviewer_identity.j2"],
    default_context=[BrandMemoryProjector],
)

review_directives_op = Operation(
    name="review_pending_directives",
    produces=PendingDirectivesOutput,
    kind="cognition",  # review es cognición, no artifacto
    agent=reviewer_agent,
    task_blocks=[
        "blocks://periodic/review_pending_directives/system.j2",
        "blocks://periodic/shared/context_rules.j2",
    ],
    consumes=[BrandMemoryProjector],
)

synthesize_learning_op = Operation(
    name="synthesize_learning",
    produces=LearningOutput,
    kind="cognition",
    agent=reviewer_agent,
    task_blocks=[
        "blocks://periodic/synthesize_learning/system.j2",
    ],
    consumes=[BrandMemoryProjector],
)

# Las reglas de negocio se mueven a un post-procesador externo
# o se implementan como validación en el output_schema de RuntimeRequest.
# El ExecutionPlan maneja el orden topológico si hay dependencias.

Notas sobre cambios de comportamiento¤

Legacy	v2.0.0
`ReviewStep` es una clase con herencia	`Operation` es una instancia de datos
`validate_business_rules()` como método	La validación de negocio se externaliza: puede ser un post-procesador que la aplicación ejecuta después de `crewmaster.execute()`, o un validador en el output_schema
`mode` (free_form vs structured_only)	`output_schema` del `RuntimeRequest` define si la salida es estructurada
`terminal_skill`	Los tools se registran en `ToolRegistry` y se asignan por scope en `AgentConfig.default_tool_scope`
Pasos ejecutados secuencialmente por un agente	Cada paso es un `Operation` independiente; si hay dependencias entre pasos, se declaran vía `consumes`

Pasos de verificación¤

Identificar cada subclase de ReviewStep en el código legacy.
Crear una Operation con kind="cognition" para cada step.
Extraer la lógica de validación de reglas de negocio a funciones standalone o validadores de Pydantic.
Si los steps tienen dependencias entre sí, declararlas vía consumes=[Type].
Construir un ExecutionPlan con todos los steps y verificar que el orden topológico es correcto.
Ejecutar cada step con crewmaster.execute(op, ...) y verificar que la salida cumple las mismas reglas de negocio.

3. AgentWithBundle → AgentConfig + Operation¤

Resumen¤

AgentWithBundle fusionaba identidad del agente (prompt, name, context_id) con declaraciones de bundles. En v2.0.0, identidad y operaciones se separan: AgentConfig define quién es el agente, Operation define qué hace.

Antes (legacy)¤

from scouters_crew.agent_with_bundle import AgentWithBundle, BundleGroup, ExecutionContext
from scouters_crew.router.prompt_engine import PromptComposition, BlockRegistry

class Paula(AgentWithBundle[PaulaContext]):
    name: str = "paula"
    root_path: Path = Path(__file__).parent
    prompt: PromptComposition = PromptComposition(blocks=[
        "blocks://agent/identity/paula.j2",
        "blocks://agent/shared/tool_use.j2",
    ])
    bundles: BundleGroup = bundle_group  # definido en otro archivo
    context_id: str = "paula_context"
    context_model: type[PaulaContext] = PaulaContext
    default_bundle: str = "corporate_colors"
    block_registry: BlockRegistry = paula_block_registry

    async def ainvoke(
        self,
        input: CollaboratorInputFresh,
        config: RunnableConfig | None = None,
    ) -> CollaboratorOutputResponseStructured:
        # Delega al default_bundle o default_specialty
        ...

El agente era un Runnable de LangChain que contenía bundles, prompts, contexto y lógica de ejecución. La identidad (prompt) y las operaciones (bundles) eran inseparables.

Después (v2.0.0)¤

from crewmaster.agents.agent import AgentConfig
from crewmaster.operations.operation import Operation
from crewmaster.agents.context.projector import ContextProjector

# ── Identidad del agente ──────────────────────────────────────────
# AgentConfig es una instancia de datos, no una clase con herencia

class PaulaContext(BaseModel):
    brand_guidelines: str
    target_audience: str

class PaulaContextProjector:
    """Proyecta PaulaContext → dict para templates Jinja2."""

    @classmethod
    def from_domain(cls, data: PaulaContext) -> "PaulaContextProjector":
        return cls()

    def to_context(self) -> dict[str, Any]:
        # Los datos reales vienen del ContextStore
        return {}

paula = AgentConfig(
    name="paula",
    identity_blocks=[
        "blocks://agent/identity/paula.j2",
        "blocks://agent/shared/tool_use.j2",
    ],
    default_context=[PaulaContextProjector],
    default_tool_scope="brand_design",
)

# ── Operaciones ────────────────────────────────────────────────────
# Cada bundle legacy es ahora una Operation independiente

generate_colors = Operation(
    name="generate_corporate_colors",
    produces=CorporateColors,
    kind="artifact",
    agent=paula,
    task_blocks=[
        "blocks://agent/task.j2",
        "blocks://agent/colors/methodology.j2",
    ],
    consumes=[PaulaContextProjector],
)

generate_typography = Operation(
    name="generate_typography",
    produces=Typography,
    kind="artifact",
    agent=paula,
    task_blocks=[
        "blocks://agent/task.j2",
        "blocks://agent/typography/methodology.j2",
    ],
    consumes=[PaulaContextProjector, CorporateColors],
)

Notas sobre cambios de comportamiento¤

Legacy	v2.0.0
`AgentWithBundle` hereda de `BaseModel` y `Runnable`	`AgentConfig` es un modelo de datos puro, no ejecutable
`ainvoke()` en el agente	`crewmaster.execute(operation, ...)` ejecuta operations, no agentes
`context_id` + `context_model` para inyectar contexto vía `RunnableConfig`	`ContextStore` + `ContextProjector` para proyección declarativa
`default_bundle` / `default_specialty`	Cada `Operation` referencia explícitamente a su `AgentConfig`
`block_registry` a nivel de agente	`BlockStore` compartido a nivel de aplicación
`root_path` para resolver archivos	`LocalDiskStore(base_dir=...)` configurado una vez
`_build_dict_for_prompt()` con namespaces `ctx`/`exc`/`wip`/`deps`	`ContextStore` + `ExecutionPlan` ensamblan `ctx`/`deps`/`cfg` automáticamente

Pasos de verificación¤

Extraer todos los campos de identidad del AgentWithBundle (name, prompt, context_id, context_model) a un AgentConfig.
Convertir cada Bundle del BundleGroup a una Operation con agent=agent_config.
Reemplazar context_model por un ContextProjector registrado en ContextStore.
Eliminar root_path, engine, block_registry — se configuran a nivel de aplicación.
Verificar que crewmaster.execute(operation, context_store, default_runtime, block_store) produce resultados equivalentes a agent.ainvoke().

4. Team graph → ExecutionPlan¤

Resumen¤

En legacy, los equipos se definían como DAGs de LangGraph con funciones nodo imperativas. En v2.0.0, el flujo de trabajo se describe declarativamente como árbol de Operation con sub_operations y consumes; resolve_plan() infiere el orden topológico.

Antes (legacy)¤

from langgraph.graph import StateGraph, END
from typing import TypedDict

class TeamState(TypedDict):
    research: ResearchResult | None
    analysis: AnalysisResult | None
    report: ReportResult | None

def research_node(state: TeamState, config: RunnableConfig) -> dict:
    result = run_research_agent(state, config)
    return {"research": result}

def analysis_node(state: TeamState, config: RunnableConfig) -> dict:
    # Lee state["research"] como dependencia implícita
    result = run_analysis_agent(state["research"], config)
    return {"analysis": result}

def report_node(state: TeamState, config: RunnableConfig) -> dict:
    result = run_report_agent(state["analysis"], config)
    return {"report": result}

graph = StateGraph(TeamState)
graph.add_node("research", research_node)
graph.add_node("analysis", analysis_node)
graph.add_node("report", report_node)
graph.add_edge("research", "analysis")
graph.add_edge("analysis", "report")
graph.set_entry_point("research")
graph.set_finish_point("report")
compiled = graph.compile()

# Ejecución
result = await compiled.ainvoke({"research": None, ...}, config)

Las dependencias entre nodos eran implícitas — cada función nodo leía del estado compartido. El orden se definía manualmente con add_edge. Agregar un nuevo nodo requería modificar el estado, las aristas y las funciones.

Después (v2.0.0)¤

from crewmaster.operations.operation import Operation
from crewmaster.agents.agent import AgentConfig

researcher = AgentConfig(
    name="researcher",
    identity_blocks=["blocks://agents/researcher.j2"],
)

analyst = AgentConfig(
    name="analyst",
    identity_blocks=["blocks://agents/analyst.j2"],
)

writer = AgentConfig(
    name="writer",
    identity_blocks=["blocks://agents/writer.j2"],
)

research_op = Operation(
    name="research",
    produces=ResearchResult,
    kind="artifact",
    agent=researcher,
    task_blocks=["blocks://tasks/research.j2"],
    # No consume nada — es el nodo raíz
)

analysis_op = Operation(
    name="analysis",
    produces=AnalysisResult,
    kind="cognition",
    agent=analyst,
    task_blocks=["blocks://tasks/analysis.j2"],
    consumes=[ResearchResult],  # Depende del tipo, no del nombre
)

report_op = Operation(
    name="report",
    produces=ReportResult,
    kind="artifact",
    agent=writer,
    task_blocks=["blocks://tasks/report.j2"],
    consumes=[AnalysisResult],
)

# La operación raíz compone el DAG
pipeline = Operation(
    name="research_pipeline",
    produces=ReportResult,
    kind="artifact",
    sub_operations=[research_op, analysis_op, report_op],
    # Sin agent — el plan delega a los sub_operations
)

# ── Construcción y ejecución ───────────────────────────────────────
from crewmaster.operations.plans import resolve_plan

plan = resolve_plan(
    operation=pipeline,
    context_store=context_store,
    block_store=block_store,
    default_runtime=pydantic_ai_driver,
)

# El plan infiere automáticamente: research → analysis → report
# basado en consumes=[ResearchResult] y consumes=[AnalysisResult]

result = await crewmaster.execute(
    operation=pipeline,
    context_store=context_store,
    default_runtime=pydantic_ai_driver,
    block_store=block_store,
)

Notas sobre cambios de comportamiento¤

Legacy	v2.0.0
DAG definido con `add_node` + `add_edge`	Árbol definido con `Operation.sub_operations` + `consumes`
Orden explícito vía aristas	Orden inferido topológicamente de `consumes` → `produces`
Estado compartido `TypedDict` mutable	Artefactos inmutables pasados vía `deps`; cada nodo solo ve sus upstreams
Funciones nodo imperativas	Operations declarativas — la lógica está en el runtime driver
`checkpointer` para persistencia de estado	Fuera del alcance de v2.0.0 — la aplicación maneja persistencia
Agregar nodo = modificar estado + aristas + funciones	Agregar nodo = agregar `Operation` con `consumes` del tipo correcto

Pasos de verificación¤

Identificar el StateGraph y sus nodos en el código legacy.
Por cada nodo, crear una Operation con produces igual al tipo de salida del nodo.
Las dependencias entre nodos se infieren: si el nodo B necesita el output del nodo A, B.consumes debe incluir A.produces.
Construir una Operation raíz con todos los nodos como sub_operations.
Ejecutar resolve_plan() y verificar que el orden topológico coincide con el DAG manual.
Verificar que falla en tiempo de build si una dependencia no se puede resolver.

5. Crew/CrewRouter → HTTP de aplicación¤

Resumen¤

CrewBase y CrewRouterBase acoplaban CrewMaster a LangChain Runnable y FastAPI. En v2.0.0, CrewMaster es una librería pura — el HTTP es responsabilidad de la aplicación.

Antes (legacy)¤

from crewmaster.features import CrewBase, TeamBase
from crewmaster.features.http_driver import CrewRouterBase, CrewSettings

# ── Capa de dominio ────────────────────────────────────────────────
class Crew(CrewBase):
    team: TeamBase = Team()

# ── Capa HTTP ───────────────────────────────────────────────────────
class CrewRouter(CrewRouterBase):
    path: str = "crew_events"
    runnable: Runnable = Crew()  # CrewBase es Runnable
    settings: CrewSettings = CrewSettings(
        llm_api_key_open_ai=os.environ["OPENAI_API_KEY"],
        llm_model_open_ai="gpt-4o",
    )
    dependencies_factory: Callable = custom_deps_factory

# FastAPI router se obtiene con:
app.include_router(CrewRouter().fastapi_router)

CrewBase era un Runnable de LangChain que envolvía un StateGraph con nodos team y cleaner. CrewRouterBase exponía un endpoint SSE vía FastAPI con autenticación, checkpointer, y transformación de input.

Después (v2.0.0)¤

# ── CrewMaster es una librería pura ────────────────────────────────
from crewmaster import execute, execute_stream
from crewmaster.operations.operation import Operation
from crewmaster.agents.agent import AgentConfig
from crewmaster.agents.context.store import ContextStore
from crewmaster.agents.prompts.block import LocalDiskStore
from crewmaster.execution.runtime import RuntimeDriver

# La aplicación define SU PROPIO endpoint HTTP
from fastapi import FastAPI, APIRouter
from sse_starlette import EventSourceResponse

app = FastAPI()

@app.post("/crew_events")
async def stream_events(request: CrewRequest):
    context_store = ContextStore()
    context_store.register(BrandContext, request.brand_context)

    block_store = LocalDiskStore(base_dir=Path("prompts"))

    operation = build_operation_from_request(request)

    async def event_generator():
        async for chunk in execute_stream(
            operation=operation,
            context_store=context_store,
            default_runtime=pydantic_ai_driver,
            block_store=block_store,
        ):
            yield {
                "event": chunk.kind,
                "data": chunk.model_dump_json(),
            }

    return EventSourceResponse(event_generator())

Notas sobre cambios de comportamiento¤

Legacy	v2.0.0
`CrewBase` extiende `Runnable`	`crewmaster.execute()` / `execute_stream()` son funciones puras
`CrewRouterBase` acoplado a FastAPI	La aplicación define sus propios endpoints HTTP
`CrewSettings` con API keys de OpenAI	La aplicación maneja configuración de proveedores LLM
`checkpointer` de LangGraph para estado	Fuera del alcance de v2.0.0 — la aplicación maneja persistencia
`llm` como dependencia del router	`RuntimeDriver` inyectado en `execute()` o en `AgentConfig`
`AuthStrategyInterface` integrado	La aplicación implementa autenticación en su capa HTTP
`HttpInput` → `CrewInput` transformación	La aplicación transforma sus propios inputs a `Operation`

Pasos de verificación¤

Extraer la lógica de negocio del CrewBase (el team y sus agentes) a Operation + AgentConfig.
Eliminar CrewBase, CrewRouterBase, CrewSettings.
Crear endpoints FastAPI (o cualquier framework HTTP) en la aplicación.
En cada endpoint, construir ContextStore, BlockStore, y llamar a crewmaster.execute() o execute_stream().
Verificar que el endpoint responde con el mismo comportamiento SSE que antes.
Verificar que la autenticación funciona correctamente (implementada por la aplicación, no por CrewMaster).

6. ContextTransformer → ContextProjector¤

Resumen¤

ContextTransformer era una clase abstracta con from_domain() y to_context(). En v2.0.0, se convierte en el protocolo ContextProjector, con soporte nativo en ContextStore y ExecutionPlan.

Antes (legacy)¤

from abc import ABC, abstractmethod
from pydantic import BaseModel

class ContextTransformer(BaseModel, ABC, Generic[InputType]):
    @classmethod
    @abstractmethod
    def from_domain(cls, data: InputType) -> "ContextTransformer[InputType]":
        raise NotImplementedError

    @abstractmethod
    def to_context(self) -> dict[str, Any]:
        raise NotImplementedError

# Implementación concreta
class BrandContextTransformer(ContextTransformer[BrandContext]):
    brand_name: str
    industry: str
    target_audience: str
    archetype: str

    @classmethod
    def from_domain(cls, data: BrandContext) -> "BrandContextTransformer":
        return cls(
            brand_name=data.name,
            industry=data.industry,
            target_audience=data.target_audience,
            archetype=data.archetype,
        )

    def to_context(self) -> dict[str, Any]:
        return {
            "brand_name": self.brand_name,
            "industry": self.industry,
            "target_audience": self.target_audience,
            "archetype": self.archetype,
        }

El transformer se instanciaba manualmente y se pasaba al contexto del agente vía RunnableConfig.

Después (v2.0.0)¤

from crewmaster.agents.context.projector import ContextProjector
from crewmaster.agents.context.store import ContextStore

# ── El proyector implementa el protocolo ───────────────────────────
# Nota: ContextProjector es un Protocol, no una clase base.
# La herencia de BaseModel es opcional si no necesitas validación.

class BrandContextProjector:
    """Proyecta BrandContext a variables de template."""

    @classmethod
    def from_domain(cls, data: BrandContext) -> "BrandContextProjector":
        # Almacena los datos internamente para to_context()
        projector = cls.__new__(cls)
        projector._brand_name = data.name
        projector._industry = data.industry
        projector._target_audience = data.target_audience
        projector._archetype = data.archetype
        return projector

    def to_context(self) -> dict[str, Any]:
        return {
            "brand_name": self._brand_name,
            "industry": self._industry,
            "target_audience": self._target_audience,
            "archetype": self._archetype,
        }

# ── Registro y uso ─────────────────────────────────────────────────
store = ContextStore()
store.register(BrandContext, brand_context_instance)

# Resolución automática vía el ExecutionPlan
operation = Operation(
    name="generate_brand",
    produces=BrandResult,
    kind="artifact",
    agent=paula,
    task_blocks=["blocks://tasks/brand.j2"],
    consumes=[BrandContextProjector],  # El plan resuelve esto automáticamente
)

Notas sobre cambios de comportamiento¤

Legacy	v2.0.0
`ContextTransformer` es ABC con herencia de `BaseModel`	`ContextProjector` es un `Protocol` — duck typing, no herencia
Instanciación manual + inyección vía `RunnableConfig`	`ContextStore.register()` + `Operation.consumes` — resolución automática
`from_domain` devuelve `Self`	`from_domain` devuelve `ContextProjector[TDomain]`
`to_json()` helper incluido	No incluido — el PromptEngine maneja la serialización
Sin inferencia de tipos	`ContextStore._infer_domain_type()` infiere `TDomain` de los type hints

Pasos de verificación¤

Identificar todas las subclases de ContextTransformer.
Convertir cada una a una clase que implemente el protocolo ContextProjector (sin herencia).
Eliminar la herencia de BaseModel — usar atributos simples o BaseModel solo si se necesita validación.
Cambiar from_domain(cls, data) → from_domain(cls, data) (la firma es idéntica).
Cambiar to_context(self) → to_context(self) (la firma es idéntica).
Registrar las instancias de dominio en ContextStore en lugar de inyectarlas vía RunnableConfig.
Verificar que context_store.resolve(BrandContextProjector) devuelve el mismo diccionario que transformer.to_context().

7. PromptEngine aplicación → PromptEngine librería¤

Resumen¤

En legacy, el PromptEngine era código de aplicación (scouters-crew) que componía PromptComposition con BlockRegistry. En v2.0.0, PromptEngine es parte de CrewMaster con BlockStore inyectable.

Antes (legacy)¤

from scouters_crew.router.prompt_engine import (
    PromptEngineLocal, PromptComposition, BlockRegistry
)

# Block registry manual
block_registry = BlockRegistry(blocks={
    "blocks://agent/task.j2": PromptBlock(id="...", content="{{ ctx.brand_name }}"),
    "blocks://agent/identity.j2": PromptBlock(id="...", content="You are {{ ctx.role }}"),
})

engine = PromptEngineLocal(
    root_path=Path("prompts"),
    enable_async=False,
    block_registry=block_registry,
)

prompt = PromptComposition(blocks=[
    "blocks://agent/identity.j2",
    "blocks://agent/task.j2",
    "blocks://context_rules",
])

rendered = engine.render(
    prompt=prompt,
    data={"ctx": {"brand_name": "Acme", "role": "designer"}},
    escape_braces=True,
)

El PromptEngine de la aplicación resolvía bloques, renderizaba Jinja2 y componía prompts. Los bloques se pasaban como PromptComposition con strings de ID.

Después (v2.0.0)¤

from crewmaster.agents.prompts.block import LocalDiskStore
from crewmaster.agents.prompts.engine import PromptEngine

# ── BlockStore resuelve archivos .j2 del disco ─────────────────────
# Cada archivo .j2 tiene frontmatter YAML con requires/provides/kind
#
# Archivo: prompts/agent/identity.j2
# ---
# requires:
#   - ctx.role
# provides: agent_identity
# kind: identity
# ---
# You are an expert {{ ctx.role }}. Your task is to assist with brand design.
#
# Archivo: prompts/agent/task.j2
# ---
# requires:
#   - ctx.brand_name
# provides: task_description
# kind: task
# ---
# Create a visual identity for {{ ctx.brand_name }}.

block_store = LocalDiskStore(base_dir="prompts")
engine = PromptEngine(block_store=block_store)

# Composición y renderizado en un solo paso
prompt = engine.compose(
    block_uris=[
        "blocks://agent/identity.j2",
        "blocks://agent/task.j2",
    ],
    variables={
        "ctx": {
            "brand_name": "Acme",
            "role": "designer",
        },
        "deps": {},
        "cfg": {},
    },
)

# Validación pre-vuelo: ¿todos los requires están satisfechos?
engine.validate_blocks(
    block_uris=[
        "blocks://agent/identity.j2",
        "blocks://agent/task.j2",
    ],
    variables={
        "ctx": {"brand_name": "Acme", "role": "designer"},
        "deps": {},
        "cfg": {},
    },
)
# Si ctx.brand_name falta, lanza ValueError ANTES de cualquier llamada al LLM

Notas sobre cambios de comportamiento¤

Legacy	v2.0.0
`PromptEngineLocal` con `root_path` + `block_registry`	`LocalDiskStore` con `base_dir` — same same
`PromptComposition(blocks=[...])` como objeto	`engine.compose(block_uris=[...])` como método
`engine.render(prompt, data, escape_braces=True)`	`engine.compose(block_uris, variables)` — renderiza y compone
Sin validación de frontmatter	Frontmatter YAML con `requires`, `provides`, `kind` validados
Sin validación de dependencias	`engine.validate_blocks()` falla en tiempo de construcción si `requires` no se satisfacen
`BlockRegistry` manual con objetos `PromptBlock`	`LocalDiskStore` lee `.j2` del disco; frontmatter se parsea automáticamente
Templates sin validación de namespaces	Namespaces fijos: `ctx.`, `deps.`, `cfg.*`

Pasos de verificación¤

Migrar los templates .j2 a la nueva estructura de directorios (si cambia).

Agregar frontmatter YAML a cada archivo .j2:

---
requires:
  - ctx.brand_name
provides: task_description
kind: task
---

Reemplazar PromptEngineLocal por PromptEngine(block_store=LocalDiskStore(...)).
Reemplazar PromptComposition + engine.render() por engine.compose().
Agregar engine.validate_blocks() en tiempo de construcción.
Verificar que un requires insatisfecho lanza ValueError antes de cualquier llamada al LLM.
Verificar que ctx.*, deps.*, cfg.* son los únicos namespaces permitidos.

8. Skill → Capability / Tool¤

Resumen¤

Skill y sus variantes (SkillStructuredResponse, SkillComputation, SkillContribute) se renombran a ToolSchema / Capability en crewmaster.tools. El cambio es mayoritariamente de nombre — la semántica de herramientas del agente se preserva.

Antes (legacy)¤

from crewmaster.features.skill import (
    Skill,
    SkillBase,
    SkillStructuredResponse,
    SkillComputation,
    SkillContribute,
)
from crewmaster.features.skill.types import SkillType

# Creación de skills
search_skill = SkillBase(
    name="web_search",
    description="Search the web for information",
    skill_type=SkillType.COMPUTATION,
)

structured_output_skill = SkillStructuredResponse(
    name="submit_design",
    description="Submit the final design proposal",
    output_schema=DesignProposal,
)

# Registro manual por agente
agent_skills = [search_skill, structured_output_skill]

Después (v2.0.0)¤

from crewmaster.tools.models import ToolSchema, Capability
from crewmaster.tools.registry import ToolRegistry

# ── ToolSchema: la unidad básica ───────────────────────────────────
search_tool = ToolSchema(
    name="web_search",
    description="Search the web for information",
    input_schema={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "The search query"}
        },
        "required": ["query"],
    },
)

# ── Capability: ToolSchema + metadata para request_capability ───────
design_capability = Capability(
    name="submit_design",
    description="Submit the final design proposal for brand identity generation",
    tool_schema=ToolSchema(
        name="submit_design",
        description="Submit the final design proposal",
        input_schema=DesignProposal.model_json_schema(),
    ),
)

# ── Registro centralizado ───────────────────────────────────────────
registry = ToolRegistry()
registry.register("brand_design", search_tool)
registry.register("brand_design", design_capability)

# Recuperación por scope
tools = registry.retrieve("brand_design")  # → [search_tool, design_capability.tool_schema]

# Recuperación semántica (request_capability)
caps = registry.retrieve_by_description("submit design proposal")  # → [design_capability]

Mapeo de tipos¤

Legacy	v2.0.0
`SkillBase`	`ToolSchema`
`SkillStructuredResponse`	`Capability(tool_schema=ToolSchema(...))`
`SkillComputation`	`ToolSchema` (la semántica está en cómo el runtime invoca la tool)
`SkillContribute`	`ToolSchema`
`SkillType` enum	Eliminado — no necesario en v2.0.0
`Skill.skill_type`	Eliminado
`Skill.output_schema`	`ToolSchema.input_schema` (schema de entrada de la tool)
Registro manual por agente	`ToolRegistry` centralizado con scopes

Pasos de verificación¤

Identificar todas las instancias de Skill, SkillBase, SkillStructuredResponse, etc.
Convertir SkillBase → ToolSchema con input_schema como JSON Schema dict.
Convertir SkillStructuredResponse → Capability con tool_schema que incluya el output_schema.
Eliminar referencias a SkillType y skill_type.
Registrar todas las tools en un ToolRegistry con scopes adecuados.
Asignar scopes a los AgentConfig vía default_tool_scope.
Verificar que registry.retrieve(scope) devuelve las tools esperadas.

9. RunnableConfig → ExecutionConfig¤

Resumen¤

RunnableConfig de LangChain se usaba para pasar configuración y dependencias a los runnables. En v2.0.0, la configuración es nativa de CrewMaster: ContextStore para dependencias de dominio y runtime_hints en RuntimeRequest para configuración del LLM.

Antes (legacy)¤

from langchain_core.runnables import RunnableConfig
from langchain_core.runnables.utils import ConfigurableFieldSpec

# ── Declaración de dependencias ────────────────────────────────────
class Paula(AgentWithBundle[PaulaContext]):
    config_specs = [
        ConfigurableFieldSpec(
            id="paula_context",
            name="Contexto para Paula",
            description="Contexto requerido para el agente Paula",
            annotation=PaulaContext,
            default=...,
        ),
    ]

# ── Construcción del config ────────────────────────────────────────
config = RunnableConfig(
    configurable={
        "paula_context": PaulaContext(
            brand_name="Acme",
            industry="Technology",
        ),
        "biz_deps": {
            "valid_ids": {"initiative": {"INIT_001", "INIT_002"}},
        },
        "llm": chat_openai_model,
        "checkpointer": memory_checkpointer,
        "thread_id": "thread_123",
    },
    tags=["crew:paula"],
    metadata={"run_name": "brand_generation"},
)

# ── Inyección en el agente ─────────────────────────────────────────
result = await paula.ainvoke(input, config)

# Dentro del agente, el contexto se extraía con:
context = config["configurable"]["paula_context"]

El RunnableConfig era un diccionario genérico con fields declarativos. Las dependencias de negocio (biz_deps), el modelo LLM y el checkpointer se mezclaban en el mismo objeto.

Después (v2.0.0)¤

from crewmaster.agents.context.store import ContextStore
from crewmaster.agents.context.projector import ContextProjector
from crewmaster.execution.runtime import RuntimeRequest

# ── Contexto de dominio vía ContextStore ────────────────────────────
store = ContextStore()
store.register(PaulaContext, PaulaContext(
    brand_name="Acme",
    industry="Technology",
))

# ── Dependencias de negocio (biz_deps) vía ContextStore ─────────────
store.register(BizDeps, BizDeps(
    valid_ids={"initiative": {"INIT_001", "INIT_002"}},
))

# ── Configuración del LLM vía runtime_hints ─────────────────────────
# Se pasa al RuntimeRequest, no a un config global
request = RuntimeRequest(
    instructions=rendered_prompt,
    tools=tools,
    context={"ctx": projected_context, "deps": {}, "cfg": {}},
    output_schema=DesignOutput,
    runtime_hints={
        "model": "gpt-4o",
        "temperature": 0.7,
        "max_tokens": 4000,
    },
)

# ── Ejecución ───────────────────────────────────────────────────────
result = await crewmaster.execute(
    operation=operation,
    context_store=store,
    default_runtime=pydantic_ai_driver,
    block_store=block_store,
)

Notas sobre cambios de comportamiento¤

Legacy	v2.0.0
`RunnableConfig(configurable={...})`	`ContextStore` + `RuntimeRequest.runtime_hints`
`ConfigurableFieldSpec` para declarar dependencias	`Operation.consumes` declara qué contextos necesita
`biz_deps` en `configurable`	Registrar `BizDeps` en `ContextStore`; crear un `BizDepsProjector` si se necesita en templates
`llm` en config	El `RuntimeDriver` se inyecta en `execute()` o en `AgentConfig.runtime`
`checkpointer` en config	Fuera del alcance de v2.0.0
`thread_id` en config	Fuera del alcance de v2.0.0
`tags` y `metadata`	Fuera del alcance de v2.0.0
`_ensure_context()` extraía del config	`ContextStore.resolve()` + `ExecutionPlan` manejan la resolución

Pasos de verificación¤

Identificar todos los ConfigurableFieldSpec en el código legacy.
Por cada field spec, crear un ContextProjector y registrar la instancia en ContextStore.
Eliminar RunnableConfig y ConfigurableFieldSpec.
Mover configuración del LLM (model, temperature) a runtime_hints en RuntimeRequest.
Mover biz_deps a ContextStore con su propio tipo de dominio.
Verificar que todas las dependencias que antes llegaban vía config["configurable"] ahora se resuelven vía ContextStore.

Resumen de cambios de imports¤

# ── ANTES (legacy) ─────────────────────────────────────────────────
from crewmaster.features import AgentBase, CrewBase, TeamBase
from crewmaster.features.skill import Skill, SkillBase, SkillStructuredResponse
from crewmaster.features.crew import CrewBase, CrewInput, CrewOutput
from crewmaster.features.http_driver import CrewRouterBase, CrewSettings
from crewmaster.features.agent import AgentBase, AgentAbstract
from scouters_crew.agent_with_bundle import AgentWithBundle, Bundle, BundleGroup, BundleDependency
from scouters_crew.router.prompt_engine import PromptEngineLocal, PromptComposition, BlockRegistry
from scouters_crew.review_steps import ReviewStep, BusinessRuleError
from langchain_core.runnables import RunnableConfig
from langchain_core.runnables.utils import ConfigurableFieldSpec
from langgraph.graph import StateGraph

# ── DESPUÉS (v2.0.0) ────────────────────────────────────────────────
from crewmaster import execute, execute_stream
from crewmaster.operations.operation import Operation
from crewmaster.operations.plans import resolve_plan, ExecutionPlan
from crewmaster.agents.agent import AgentConfig
from crewmaster.agents.context.projector import ContextProjector
from crewmaster.agents.context.store import ContextStore
from crewmaster.agents.prompts.block import BlockStore, LocalDiskStore, PromptBlock
from crewmaster.agents.prompts.engine import PromptEngine
from crewmaster.tools.models import ToolSchema, Capability
from crewmaster.tools.registry import ToolRegistry
from crewmaster.execution.runtime import RuntimeDriver, RuntimeRequest, RuntimeResponse
from crewmaster.collaboration.protocol import (
    CollaborationProtocol, GrillMeProtocol, GrillMeConfig,
    DebateProtocol, BlackboardProtocol, SupervisorProtocol, ConsensusProtocol,
)
from crewmaster.conversation.channel import channel_dispatch, ChannelAgentMessage, Silent
from crewmaster.conversation.dialogue import dialogue, AgentOutputClarification

Verificación final de migración¤

Para confirmar que la migración de un módulo de Cortex fue exitosa:

Compilación: poetry run python -c "from myapp.migrated_module import *" no arroja errores.
Sin imports legacy: grep -r "crewmaster.features" myapp/ no devuelve resultados (excepto crewmaster.features.performance que se preserva).
Sin imports LangChain: grep -r "langchain_core.runnables.RunnableConfig" myapp/ no devuelve resultados.
Plan válido: resolve_plan(operation, context_store, block_store, driver) no arroja PlanBuildError.
Ejecución: await crewmaster.execute(operation, ...) produce un resultado tipado correcto.
Validación temprana: Un requires insatisfecho en un bloque .j2 lanza ValueError en validate_blocks(), no en tiempo de ejecución.
Streaming: execute_stream(operation, ...) emite chunks en orden: text_delta → tool_call → tool_result → node_complete → final.