Directory structure:
└── Coven/
    ├── README.md
    ├── main.py
    ├── pyproject.toml
    ├── .env.example
    ├── .python-version
    ├── examples/
    │   └── research_report.py
    ├── Coven/
    │   ├── __init__.py
    │   ├── pipeline.py
    │   ├── compiler/
    │   │   ├── __init__.py
    │   │   ├── agent.py
    │   │   └── formatter.py
    │   ├── decomposer/
    │   │   ├── __init__.py
    │   │   ├── agent.py
    │   │   └── parser.py
    │   ├── graph_builder/
    │   │   ├── __init__.py
    │   │   ├── agent.py
    │   │   ├── parser.py
    │   │   └── validator.py
    │   ├── initiator/
    │   │   ├── __init__.py
    │   │   ├── agent_runner.py
    │   │   ├── artifact_store.py
    │   │   └── executor.py
    │   ├── mcp_builder/
    │   │   ├── __init__.py
    │   │   └── builder.py
    │   ├── models/
    │   │   ├── __init__.py
    │   │   ├── artifact.py
    │   │   ├── dag.py
    │   │   └── node.py
    │   ├── sorter/
    │   │   ├── __init__.py
    │   │   ├── topological.py
    │   │   └── validator.py
    │   └── synthesizer/
    │       ├── __init__.py
    │       ├── agent.py
    │       ├── injector.py
    │       └── parser.py
    ├── prompts/
    │   ├── compiler.txt
    │   ├── decomposer.txt
    │   ├── graph_builder.txt
    │   └── synthesizer.txt
    └── tests/
        ├── __init__.py
        ├── conftest.py
        ├── compiler/
        │   ├── __init__.py
        │   └── test_compiler.py
        ├── decomposer/
        │   ├── __init__.py
        │   └── test_decomposer.py
        ├── graph_builder/
        │   ├── __init__.py
        │   └── test_graph_builder.py
        ├── initiator/
        │   ├── __init__.py
        │   └── test_initiator.py
        ├── mcp_builder/
        │   ├── __init__.py
        │   └── test_mpc_builder.py
        ├── models/
        │   ├── __init__.py
        │   └── test_models.py
        ├── sorter/
        │   ├── __init__.py
        │   └── test_sorter.py
        └── synthesizer/
            ├── __init__.py
            └── test_synthesizer.py

================================================
FILE: README.md
================================================
[Empty file]


================================================
FILE: main.py
================================================
import asyncio
import logging
import sys

from coven import Coven

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)


async def main() -> None:
    if len(sys.argv) < 2:
        print("Usage: uv run main.py \"<your task here>\"")
        sys.exit(1)

    task = sys.argv[1]
    model = sys.argv[2] if len(sys.argv) > 2 else "gpt-4o"

    Coven = Coven(model=model)
    dag  = await Coven.run(task)
    print(Coven.to_text(dag))


if __name__ == "__main__":
    asyncio.run(main())


================================================
FILE: pyproject.toml
================================================
[project]
name = "Coven"
version = "0.1.0"
description = "A multi-agent DAG pipeline framework that decomposes complex tasks into coordinated agents with real tool execution via ToolStorePy"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "litellm>=1.40.0",
    "instructor>=1.3.0",
    "networkx>=3.3",
    "pydantic>=2.7.0",
    "httpx>=0.27.0",
    "python-dotenv>=1.0.0",
    "toolstorepy>=0.1.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0.0",
    "pytest-asyncio>=0.23.0",
    "pytest-mock>=3.14.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["Coven"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]


================================================
FILE: .env.example
================================================
# Copy to .env and fill in your keys
# Coven uses litellm â€” set whichever provider you use

# OpenAI
OPENAI_API_KEY=sk-...

# Anthropic
ANTHROPIC_API_KEY=sk-ant-...

# Azure OpenAI (optional)
# AZURE_API_KEY=...
# AZURE_API_BASE=...
# AZURE_API_VERSION=...



================================================
FILE: .python-version
================================================
3.12



================================================
FILE: examples/research_report.py
================================================
[Empty file]


================================================
FILE: Coven/__init__.py
================================================
from .pipeline import Coven
from .models import DAG, DAGStatus, Node, NodeType, NodeStatus, Artifact

__all__ = [
    "Coven",
    "DAG",
    "DAGStatus",
    "Node",
    "NodeType",
    "NodeStatus",
    "Artifact",
]


================================================
FILE: Coven/pipeline.py
================================================
from __future__ import annotations

import logging
import uuid
from pathlib import Path

from coven.models import DAG, DAGStatus
from coven.decomposer import DecomposerAgent, DecomposerParser
from coven.graph_builder import GraphBuilderAgent, GraphBuilderParser, GraphBuilderValidator
from coven.sorter import TopologicalSorter, SorterValidator
from coven.synthesizer import SynthesizerInjector
from coven.mcp_builder import MCPNodeBuilder
from coven.initiator import Executor
from coven.compiler import CompilerAgent, CompilerFormatter, CompilerResponse, OutputSection


logger = logging.getLogger(__name__)


class Coven:
    """
    Top-level orchestrator for the Coven pipeline.

    Runs all five stages in sequence:
        1. Decomposer     â€” breaks task into nodes + artifacts
        2. Graph Builder  â€” validates and formalizes DAG edges
        3. Sorter         â€” topological sort into execution levels
        4. Initiator      â€” level-wise async agent execution
                            (with per-node MCP server builds via ToolStorePy)
        5. Compiler       â€” assembles final output from all artifacts

    Usage:
        Coven = Coven(model="gpt-4o")
        dag  = await Coven.run("Produce a go-to-market strategy for a B2B SaaS product")
        print(Coven.to_text(dag))

    Tool usage:
        Nodes whose query_tool list is non-empty will have a ToolStorePy MCP
        server built for them automatically before execution. The decomposer
        LLM decides which tools each node needs â€” just describe the task and
        Coven handles the rest.
    """

    def __init__(
        self,
        model: str = "gpt-4o",
        workspace: str | Path = "coven_workspace",
        mcp_index: str | None = "core-tools",
        mcp_index_url: str | None = None,
        mcp_install_requirements: bool = False,
        mcp_verbose: bool = False,
    ):
        """
        Args:
            model: LiteLLM-compatible model string (e.g. "gpt-4o", "claude-sonnet-4-6").
            workspace: Root directory for all run artifacts and MCP workspaces.
            mcp_index: ToolStorePy built-in index name. Default: "core-tools".
            mcp_index_url: Direct URL to a custom ToolStorePy index. Overrides mcp_index.
            mcp_install_requirements: Install repo requirements in MCP venv.
            mcp_verbose: Enable verbose ToolStorePy logging.
        """
        self.model     = model
        self.workspace = Path(workspace)

        # â”€â”€ MCP builder â€” shared across all nodes in a run â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
        self._mcp_builder = MCPNodeBuilder(
            base_workspace=self.workspace,
            index=mcp_index if not mcp_index_url else None,
            index_url=mcp_index_url,
            install_requirements=mcp_install_requirements,
            verbose=mcp_verbose,
        )

        # â”€â”€ Pipeline stages â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
        self._decomposer        = DecomposerAgent(model=model)
        self._decomposer_parser = DecomposerParser()
        self._graph_builder     = GraphBuilderAgent(model=model)
        self._graph_parser      = GraphBuilderParser()
        self._graph_validator   = GraphBuilderValidator()
        self._sorter            = TopologicalSorter()
        self._sorter_validator  = SorterValidator()
        self._synth_injector    = SynthesizerInjector()
        self._executor          = Executor(model=model, mcp_builder=self._mcp_builder)
        self._compiler          = CompilerAgent(model=model)
        self._formatter         = CompilerFormatter()

    async def run(self, task: str) -> DAG:
        """
        Execute the full Coven pipeline for a given task.

        Args:
            task: The complex task to solve.

        Returns:
            Completed DAG with final_output populated.
        """
        dag_id = str(uuid.uuid4())[:8]
        logger.info(f"[{dag_id}] Starting Coven pipeline: {task[:80]}...")

        # â”€â”€ Stage 1: Decompose â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
        logger.info(f"[{dag_id}] Stage 1: Decomposing task...")
        decomp_response      = await self._decomposer.arun(task)
        nodes, artifacts     = self._decomposer_parser.parse(decomp_response)
        logger.info(f"[{dag_id}] â†’ {len(nodes)} nodes, {len(artifacts)} artifacts.")

        # â”€â”€ Stage 2: Build Graph â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
        logger.info(f"[{dag_id}] Stage 2: Building graph...")
        graph_response       = await self._graph_builder.arun(nodes, artifacts)
        nodes, artifacts, edges, synthesizer_targets = self._graph_parser.parse(graph_response)
        G                    = self._graph_validator.validate(nodes, artifacts, edges)
        logger.info(f"[{dag_id}] â†’ {len(edges)} edges, synthesizer targets: {synthesizer_targets}")

        # â”€â”€ Stage 2b: Inject Synthesizer Nodes â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
        if synthesizer_targets:
            logger.info(f"[{dag_id}] Injecting synthesizer nodes...")
            nodes, artifacts, G = self._synth_injector.inject(
                nodes, artifacts, synthesizer_targets, G
            )
            logger.info(f"[{dag_id}] â†’ {len(nodes)} total nodes after injection.")

        # â”€â”€ Stage 3: Topological Sort â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
        logger.info(f"[{dag_id}] Stage 3: Sorting DAG...")
        self._sorter_validator.validate(G, nodes, artifacts)
        levels = self._sorter.sort(G)
        logger.info(f"[{dag_id}]\n{self._sorter.describe(G, nodes)}")

        # â”€â”€ Assemble DAG â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
        dag = DAG(
            id=dag_id,
            task=task,
            nodes=nodes,
            artifacts=artifacts,
            levels=levels,
        )

        # â”€â”€ Stage 4: Execute â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
        logger.info(f"[{dag_id}] Stage 4: Executing {len(levels)} levels...")
        dag = await self._executor.execute(dag)

        if dag.status == DAGStatus.FAILED:
            logger.error(f"[{dag_id}] Pipeline execution failed.")
            return dag

        # â”€â”€ Stage 5: Compile â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
        logger.info(f"[{dag_id}] Stage 5: Compiling final output...")
        compiler_response = await self._compiler.arun(dag)
        dag               = self._formatter.format(compiler_response, dag)

        logger.info(f"[{dag_id}] Pipeline complete.")
        return dag

    def to_text(self, dag: DAG) -> str:
        """Render the final DAG output as plain text."""
        if not dag.final_output:
            return "Pipeline did not produce a final output."

        response = CompilerResponse(
            title=dag.final_output.get("title", ""),
            summary=dag.final_output.get("summary", ""),
            sections=[
                OutputSection(**s) for s in dag.final_output.get("sections", [])
            ],
            recommendations=dag.final_output.get("recommendations", []),
            metadata=dag.final_output.get("metadata", {}),
        )
        return self._formatter.to_text(response)


================================================
FILE: Coven/compiler/__init__.py
================================================
from .agent import CompilerAgent, CompilerResponse, OutputSection
from .formatter import CompilerFormatter

__all__ = [
    "CompilerAgent",
    "CompilerResponse",
    "OutputSection",
    "CompilerFormatter",
]


================================================
FILE: Coven/compiler/agent.py
================================================
from __future__ import annotations

import json
from pathlib import Path

import instructor
from litellm import completion
from pydantic import BaseModel

from coven.models import DAG, Artifact


# â”€â”€ Instructor client â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

_client = instructor.from_litellm(completion)

# â”€â”€ Prompt â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

_PROMPT_PATH = Path(__file__).parent.parent.parent / "prompts" / "compiler.txt"


def _load_prompt() -> str:
    return _PROMPT_PATH.read_text(encoding="utf-8")


# â”€â”€ Response schema â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class OutputSection(BaseModel):
    title: str
    content: str
    source_artifacts: list[str]


class CompilerResponse(BaseModel):
    title: str
    summary: str
    sections: list[OutputSection]
    recommendations: list[str]
    metadata: dict


# â”€â”€ Agent â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class CompilerAgent:
    """
    Stage 5 â€” Takes all completed artifacts from the executed DAG
    and compiles them into a single coherent final output.

    Responsibilities:
    - Synthesize all artifacts into a unified response
    - Organize content into logical sections
    - Surface synthesis decisions and metadata
    - Directly address the user's original task
    """

    def __init__(self, model: str = "gpt-4o"):
        self.model = model
        self.system_prompt = _load_prompt()

    def _build_user_message(self, dag: DAG) -> str:
        """
        Build compiler context from the completed DAG.

        Includes the original task and all produced artifact bodies.
        Filters out partial artifacts (synthesizer intermediates)
        since their content is already merged into final artifacts.
        """
        final_artifacts = {
            name: artifact
            for name, artifact in dag.artifacts.items()
            if "__partial__" not in name
        }

        payload = {
            "task": dag.task,
            "artifacts": [
                {
                    "name": artifact.name,
                    "description": artifact.description,
                    "body": artifact.body,
                }
                for artifact in final_artifacts.values()
            ],
            "total_agents": len(dag.nodes),
            "total_artifacts": len(final_artifacts),
        }

        return json.dumps(payload, indent=2)

    def run(self, dag: DAG) -> CompilerResponse:
        """
        Synchronous compilation call.

        Args:
            dag: Fully executed DAG with all artifact bodies populated.

        Returns:
            CompilerResponse â€” the final structured output.
        """
        response: CompilerResponse = _client.chat.completions.create(
            model=self.model,
            response_model=CompilerResponse,
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user",   "content": self._build_user_message(dag)},
            ],
        )
        return response

    async def arun(self, dag: DAG) -> CompilerResponse:
        """
        Async compilation call.

        Args:
            dag: Fully executed DAG with all artifact bodies populated.

        Returns:
            CompilerResponse â€” the final structured output.
        """
        response: CompilerResponse = await _client.chat.completions.acreate(
            model=self.model,
            response_model=CompilerResponse,
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user",   "content": self._build_user_message(dag)},
            ],
        )
        return response


================================================
FILE: Coven/compiler/formatter.py
================================================
from __future__ import annotations

from coven.models import DAG
from .agent import CompilerResponse


class CompilerFormatter:
    """
    Applies the CompilerResponse back onto the DAG as the final output.

    Converts the structured CompilerResponse into a clean dict
    stored in dag.final_output, ready for the caller to consume,
    serialize, or display.

    Also produces a plain text rendering for quick human reading.
    """

    def format(
        self,
        response: CompilerResponse,
        dag: DAG,
    ) -> DAG:
        """
        Write the compiled output into dag.final_output.

        Args:
            response: CompilerResponse from CompilerAgent.
            dag: The completed DAG.

        Returns:
            DAG with final_output populated.
        """
        final_output = {
            "title":           response.title,
            "summary":         response.summary,
            "sections":        [s.model_dump() for s in response.sections],
            "recommendations": response.recommendations,
            "metadata": {
                **response.metadata,
                "dag_id":            dag.id,
                "task":              dag.task,
                "total_agents":      len(dag.nodes),
                "artifacts_produced": len([
                    a for a in dag.artifacts
                    if "__partial__" not in a
                ]),
            },
        }

        return dag.model_copy(update={"final_output": final_output})

    def to_text(self, response: CompilerResponse) -> str:
        """
        Render the CompilerResponse as a plain text string.
        Useful for CLI output, logging, or simple display.

        Args:
            response: CompilerResponse from CompilerAgent.

        Returns:
            Formatted plain text string.
        """
        lines: list[str] = []

        lines.append("=" * 60)
        lines.append(response.title.upper())
        lines.append("=" * 60)
        lines.append("")
        lines.append(response.summary)
        lines.append("")

        for section in response.sections:
            lines.append(f"â”€â”€ {section.title} {'â”€' * (50 - len(section.title))}")
            lines.append(section.content)
            lines.append(
                f"  [sources: {', '.join(section.source_artifacts)}]"
            )
            lines.append("")

        if response.recommendations:
            lines.append("â”€â”€ Recommendations " + "â”€" * 41)
            for i, rec in enumerate(response.recommendations, 1):
                lines.append(f"  {i}. {rec}")
            lines.append("")

        meta = response.metadata
        lines.append("â”€â”€ Metadata " + "â”€" * 48)
        lines.append(f"  Agents: {meta.get('total_agents', 'â€”')}")
        lines.append(f"  Artifacts: {meta.get('artifacts_produced', 'â€”')}")

        if meta.get("synthesis_decisions"):
            lines.append("  Synthesis decisions:")
            for decision in meta["synthesis_decisions"]:
                lines.append(f"    â€¢ {decision}")

        lines.append("=" * 60)

        return "\n".join(lines)


================================================
FILE: Coven/decomposer/__init__.py
================================================
from .agent import DecomposerAgent, DecomposerResponse
from .parser import DecomposerParser

__all__ = [
    "DecomposerAgent",
    "DecomposerResponse",
    "DecomposerParser",
]


================================================
FILE: Coven/decomposer/agent.py
================================================
[Binary file]


================================================
FILE: Coven/decomposer/parser.py
================================================
from __future__ import annotations

from coven.models import Artifact, Node, NodeType, NodeStatus
from .agent import DecomposerResponse, DecomposedNode, DecomposedArtifact


class DecomposerParser:
    """
    Converts the raw DecomposerResponse from the LLM into
    validated Artifact and Node model instances.

    Keeps parsing logic separate from the LLM call so each
    can be tested and changed independently.
    """

    def parse_artifacts(
        self,
        raw_artifacts: list[DecomposedArtifact]
    ) -> dict[str, Artifact]:
        """
        Convert raw artifact dicts into Artifact models keyed by name.

        Args:
            raw_artifacts: List of DecomposedArtifact from LLM response.

        Returns:
            Dict of artifact name â†’ Artifact model.
        """
        artifacts: dict[str, Artifact] = {}

        for raw in raw_artifacts:
            artifact = Artifact(
                name=raw.name,
                description=raw.description,
                contributors=raw.contributors,
                users=raw.users,
                body=raw.body,
            )
            artifacts[artifact.name] = artifact

        return artifacts

    def parse_nodes(
        self,
        raw_nodes: list[DecomposedNode]
    ) -> dict[str, Node]:
        """
        Convert raw node dicts into Node models keyed by node ID.

        Args:
            raw_nodes: List of DecomposedNode from LLM response.

        Returns:
            Dict of node ID â†’ Node model.
        """
        nodes: dict[str, Node] = {}

        for raw in raw_nodes:
            node = Node(
                id=raw.id,
                name=raw.name,
                node_type=NodeType(raw.node_type),
                system_prompt=raw.system_prompt,
                query_tool=raw.query_tool,
                input_artifacts=raw.input_artifacts,
                output_artifacts=raw.output_artifacts,
                status=NodeStatus.PENDING,
            )
            nodes[node.id] = node

        return nodes

    def parse(self, response: DecomposerResponse) -> tuple[dict[str, Node], dict[str, Artifact]]:
        """
        Full parse of a DecomposerResponse into nodes and artifacts.

        Args:
            response: DecomposerResponse from DecomposerAgent.

        Returns:
            Tuple of (nodes dict, artifacts dict).
        """
        artifacts = self.parse_artifacts(response.artifacts)
        nodes     = self.parse_nodes(response.nodes)

        self._validate_references(nodes, artifacts)

        return nodes, artifacts

    def _validate_references(
        self,
        nodes: dict[str, Node],
        artifacts: dict[str, Artifact],
    ) -> None:
        """
        Ensure all artifact references in nodes point to real artifacts,
        and all contributor/user references in artifacts point to real nodes.

        Raises:
            ValueError: If any reference is broken.
        """
        artifact_names = set(artifacts.keys())
        node_ids       = set(nodes.keys())

        for node_id, node in nodes.items():
            for name in node.input_artifacts:
                if name not in artifact_names:
                    raise ValueError(
                        f"Node '{node_id}' references unknown input artifact '{name}'."
                    )
            for name in node.output_artifacts:
                if name not in artifact_names:
                    raise ValueError(
                        f"Node '{node_id}' references unknown output artifact '{name}'."
                    )

        for artifact_name, artifact in artifacts.items():
            for contributor in artifact.contributors:
                if contributor not in node_ids:
                    raise ValueError(
                        f"Artifact '{artifact_name}' contributor '{contributor}' "
                        f"does not match any node ID."
                    )
            for user in artifact.users:
                if user not in node_ids:
                    raise ValueError(
                        f"Artifact '{artifact_name}' user '{user}' "
                        f"does not match any node ID."
                    )


================================================
FILE: Coven/graph_builder/__init__.py
================================================
from .agent import GraphBuilderAgent, GraphBuilderResponse, Edge
from .parser import GraphBuilderParser
from .validator import GraphBuilderValidator, GraphValidationError

__all__ = [
    "GraphBuilderAgent",
    "GraphBuilderResponse",
    "Edge",
    "GraphBuilderParser",
    "GraphBuilderValidator",
    "GraphValidationError",
]


================================================
FILE: Coven/graph_builder/agent.py
================================================
[Binary file]


================================================
FILE: Coven/graph_builder/parser.py
================================================
from __future__ import annotations

from coven.models import Artifact, Node, NodeType, NodeStatus
from .agent import GraphBuilderResponse, Edge


class GraphBuilderParser:
    """
    Converts the GraphBuilderResponse into updated Node and Artifact
    model instances, incorporating any repairs the LLM made.

    Also extracts synthesizer injection hints from the issues list.
    """

    def parse(
        self,
        response: GraphBuilderResponse,
    ) -> tuple[dict[str, Node], dict[str, Artifact], list[Edge], list[str]]:
        """
        Full parse of GraphBuilderResponse.

        Args:
            response: GraphBuilderResponse from GraphBuilderAgent.

        Returns:
            Tuple of:
            - nodes dict (corrected)
            - artifacts dict (corrected)
            - edges list
            - synthesizer_targets: artifact names needing synthesizer injection
        """
        nodes     = self._parse_nodes(response.nodes)
        artifacts = self._parse_artifacts(response.artifacts)
        edges     = response.edges
        synthesizer_targets = self._extract_synthesizer_targets(response.issues)

        return nodes, artifacts, edges, synthesizer_targets

    def _parse_nodes(self, raw_nodes: list[dict]) -> dict[str, Node]:
        nodes: dict[str, Node] = {}
        for raw in raw_nodes:
            node = Node(
                id=raw["id"],
                name=raw["name"],
                node_type=NodeType(raw["node_type"]),
                system_prompt=raw["system_prompt"],
                query_tool=raw.get("query_tool", {}),
                input_artifacts=raw.get("input_artifacts", []),
                output_artifacts=raw.get("output_artifacts", []),
                status=NodeStatus.PENDING,
            )
            nodes[node.id] = node
        return nodes

    def _parse_artifacts(self, raw_artifacts: list[dict]) -> dict[str, Artifact]:
        artifacts: dict[str, Artifact] = {}
        for raw in raw_artifacts:
            artifact = Artifact(
                name=raw["name"],
                description=raw["description"],
                contributors=raw.get("contributors", []),
                users=raw.get("users", []),
                body=raw.get("body", {}),
            )
            artifacts[artifact.name] = artifact
        return artifacts

    def _extract_synthesizer_targets(self, issues: list[str]) -> list[str]:
        """
        Scan the issues list for SYNTHESIZER_NEEDED hints emitted by the LLM.

        Returns:
            List of artifact names that need a synthesizer node injected.
        """
        targets: list[str] = []
        for issue in issues:
            if issue.startswith("SYNTHESIZER_NEEDED:"):
                artifact_name = issue.split(":", 1)[1].strip()
                targets.append(artifact_name)
        return targets


================================================
FILE: Coven/graph_builder/validator.py
================================================
from __future__ import annotations

import networkx as nx

from coven.models import Artifact, Node
from .agent import Edge


class GraphValidationError(Exception):
    """Raised when the DAG fails structural validation."""
    pass


class GraphBuilderValidator:
    """
    Pure algorithmic validation of the DAG structure using networkx.

    Runs AFTER the LLM graph builder and parser have produced nodes,
    artifacts, and edges. This is the hard safety net â€” if the LLM
    missed a cycle or produced a disconnected graph, this catches it.

    Raises GraphValidationError on any structural problem.
    """

    def validate(
        self,
        nodes: dict[str, Node],
        artifacts: dict[str, Artifact],
        edges: list[Edge],
    ) -> nx.DiGraph:
        """
        Build and validate the networkx DiGraph.

        Args:
            nodes: Node dict from GraphBuilderParser.
            artifacts: Artifact dict from GraphBuilderParser.
            edges: Edge list from GraphBuilderParser.

        Returns:
            Valid nx.DiGraph ready for topological sort.

        Raises:
            GraphValidationError: On cycles, unknown node refs, or isolated nodes.
        """
        G = self._build_graph(nodes, edges)

        self._check_unknown_nodes(nodes, edges)
        self._check_cycles(G)
        self._check_isolated_nodes(G, nodes)

        return G

    def _build_graph(
        self,
        nodes: dict[str, Node],
        edges: list[Edge],
    ) -> nx.DiGraph:
        G = nx.DiGraph()
        G.add_nodes_from(nodes.keys())
        for edge in edges:
            G.add_edge(edge.from_node, edge.to_node, artifact=edge.artifact)
        return G

    def _check_unknown_nodes(
        self,
        nodes: dict[str, Node],
        edges: list[Edge],
    ) -> None:
        node_ids = set(nodes.keys())
        for edge in edges:
            if edge.from_node not in node_ids:
                raise GraphValidationError(
                    f"Edge references unknown from_node '{edge.from_node}'."
                )
            if edge.to_node not in node_ids:
                raise GraphValidationError(
                    f"Edge references unknown to_node '{edge.to_node}'."
                )

    def _check_cycles(self, G: nx.DiGraph) -> None:
        if not nx.is_directed_acyclic_graph(G):
            cycles = list(nx.simple_cycles(G))
            raise GraphValidationError(
                f"DAG contains cycles: {cycles}. "
                f"The graph builder failed to resolve all circular dependencies."
            )

    def _check_isolated_nodes(
        self,
        G: nx.DiGraph,
        nodes: dict[str, Node],
    ) -> None:
        """
        Warn on nodes with no edges at all.
        These are either entry points with no dependencies (acceptable)
        or genuinely disconnected nodes (problematic).

        We allow nodes with no incoming edges (DAG roots) but flag
        nodes with neither incoming nor outgoing edges.
        """
        isolated = list(nx.isolates(G))
        if isolated:
            raise GraphValidationError(
                f"The following nodes have no edges at all and are disconnected "
                f"from the DAG: {isolated}. Every node must produce or consume "
                f"at least one artifact."
            )


================================================
FILE: Coven/initiator/__init__.py
================================================
from .executor import Executor
from .agent_runner import AgentRunner
from .artifact_store import ArtifactStore

__all__ = [
    "Executor",
    "AgentRunner",
    "ArtifactStore",
]


================================================
FILE: Coven/initiator/agent_runner.py
================================================
from __future__ import annotations

import json
import logging
from pathlib import Path

import instructor
from litellm import completion
from pydantic import BaseModel

from coven.models import Node, NodeType, NodeStatus, Artifact
from coven.mcp_builder import MCPNodeBuilder
from .artifact_store import ArtifactStore
from coven.synthesizer import SynthesizerAgent, SynthesizerParser


logger = logging.getLogger(__name__)

# â”€â”€ Instructor client â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

_client = instructor.from_litellm(completion)


# â”€â”€ Domain agent response schema â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class DomainAgentResponse(BaseModel):
    """
    Structured output for a domain agent.
    Each output artifact gets its own body keyed by artifact name.
    """
    outputs: dict[str, dict]   # artifact_name â†’ artifact body


# â”€â”€ Agent Runner â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class AgentRunner:
    """
    Executes a single node in the DAG.

    Before executing any domain node with non-empty query_tool:
        1. Calls MCPNodeBuilder to build a dedicated MCP server via ToolStorePy
        2. Stores the MCP server path on the node
        3. Injects MCP server path into the agent's user message context

    Handles two node types at execution time:
        - DOMAIN: standard LLM call with input artifacts + optional MCP context
        - SYNTHESIZER: delegates to SynthesizerAgent with partial artifacts

    After execution:
        - Writes all output artifact bodies to the ArtifactStore
        - Updates the node status to COMPLETED or FAILED
    """

    def __init__(
        self,
        model: str = "gpt-4o",
        mcp_builder: MCPNodeBuilder | None = None,
    ):
        self.model        = model
        self._mcp_builder = mcp_builder
        self._synth_agent  = SynthesizerAgent(model=model)
        self._synth_parser = SynthesizerParser()

    async def run(
        self,
        node: Node,
        store: ArtifactStore,
        nodes: dict[str, Node],
    ) -> Node:
        """
        Execute a single node asynchronously.

        Args:
            node: The node to execute.
            store: Shared artifact store.
            nodes: Full node registry.

        Returns:
            Updated Node with status set to COMPLETED or FAILED.
        """
        try:
            # â”€â”€ Build MCP server if node has tool queries â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
            if node.query_tool and self._mcp_builder:
                mcp_path = self._mcp_builder.build_for_node(node)
                if mcp_path:
                    node = node.model_copy(
                        update={"mcp_server_path": str(mcp_path)}
                    )
                    logger.info(
                        f"Node '{node.id}' MCP server ready â†’ {mcp_path}"
                    )

            # â”€â”€ Execute node â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
            if node.node_type == NodeType.SYNTHESIZER:
                await self._run_synthesizer(node, store)
            else:
                await self._run_domain(node, store)

            return node.model_copy(update={"status": NodeStatus.COMPLETED})

        except Exception as e:
            logger.error(f"Node '{node.id}' failed: {e}")
            return node.model_copy(
                update={
                    "status": NodeStatus.FAILED,
                    "result": {"error": str(e)},
                }
            )

    async def _run_domain(self, node: Node, store: ArtifactStore) -> None:
        """
        Execute a domain agent node.

        Builds context from input artifacts and optional MCP server info,
        calls the LLM, and writes each output artifact body to the store.
        """
        input_artifacts = await store.get_many(node.input_artifacts)
        user_message    = self._build_domain_message(node, input_artifacts)

        response: DomainAgentResponse = await _client.chat.completions.acreate(
            model=self.model,
            response_model=DomainAgentResponse,
            messages=[
                {"role": "system", "content": node.system_prompt},
                {"role": "user",   "content": user_message},
            ],
        )

        for artifact_name, body in response.outputs.items():
            if artifact_name in node.output_artifacts:
                await store.put(artifact_name, body)

    async def _run_synthesizer(self, node: Node, store: ArtifactStore) -> None:
        """
        Execute a synthesizer node.

        Retrieves all partial artifacts, delegates to SynthesizerAgent,
        and writes the merged artifact body to the store.
        """
        partial_artifacts = await store.get_many(node.input_artifacts)
        target_artifact   = await store.get(node.output_artifacts[0])

        response = await self._synth_agent.arun(
            target_artifact=target_artifact,
            partial_artifacts=partial_artifacts,
            synth_node=node,
        )

        merged_artifact = self._synth_parser.parse(response, target_artifact)
        await store.put(target_artifact.name, merged_artifact.body)

    def _build_domain_message(
        self,
        node: Node,
        input_artifacts: list[Artifact],
    ) -> str:
        """
        Build the user message for a domain agent.

        Includes:
        - All input artifact bodies as context
        - Expected output artifact names
        - MCP server path and tool descriptions if available
        - Instructions for using MCP tools
        """
        payload: dict = {
            "input_artifacts": [
                {
                    "name":        artifact.name,
                    "description": artifact.description,
                    "body":        artifact.body,
                }
                for artifact in input_artifacts
            ],
            "output_artifacts": node.output_artifacts,
            "instructions": (
                "Using the input artifacts provided as context, "
                "produce the required output artifacts. "
                "Return each output artifact body keyed by its artifact name "
                "inside the 'outputs' field."
            ),
        }

        # â”€â”€ Inject MCP tool context if available â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
        if node.mcp_server_path and node.query_tool:
            payload["mcp_tools"] = {
                "server_path": node.mcp_server_path,
                "available_tools": [
                    tq.tool_description for tq in node.query_tool
                ],
                "instructions": (
                    "A ToolStorePy MCP server has been built for you and is available "
                    "at the path above. It contains real implementations of the tools "
                    "listed in 'available_tools'. Use these tools to complete your task "
                    "rather than relying on simulated outputs."
                ),
            }

        return json.dumps(payload, indent=2)


================================================
FILE: Coven/initiator/artifact_store.py
================================================
from __future__ import annotations

import asyncio
from typing import Any

from coven.models import Artifact


class ArtifactStore:
    """
    Shared in-memory store for all artifacts produced during DAG execution.

    Acts as the single source of truth for artifact state across all levels.
    Thread-safe via asyncio.Lock â€” multiple agents in the same level
    can safely write their outputs concurrently.

    Lifecycle:
    - Initialized with all artifact definitions at DAG start (bodies empty)
    - Agents write their output artifact bodies via put()
    - Downstream agents read their input artifacts via get()
    - At DAG completion the compiler reads all final artifacts
    """

    def __init__(self, artifacts: dict[str, Artifact]):
        self._store: dict[str, Artifact] = dict(artifacts)
        self._lock = asyncio.Lock()

    async def get(self, artifact_name: str) -> Artifact:
        """
        Retrieve an artifact by name.

        Args:
            artifact_name: The artifact's unique name.

        Returns:
            The Artifact model (may have empty body if not yet produced).

        Raises:
            KeyError: If artifact name is not registered in the store.
        """
        async with self._lock:
            if artifact_name not in self._store:
                raise KeyError(
                    f"Artifact '{artifact_name}' not found in store. "
                    f"Registered artifacts: {list(self._store.keys())}"
                )
            return self._store[artifact_name]

    async def put(self, artifact_name: str, body: dict[str, Any]) -> None:
        """
        Write a produced artifact body into the store.

        Args:
            artifact_name: The artifact's unique name.
            body: The produced artifact body from an agent.

        Raises:
            KeyError: If artifact name is not registered in the store.
        """
        async with self._lock:
            if artifact_name not in self._store:
                raise KeyError(
                    f"Cannot write artifact '{artifact_name}' â€” not registered in store."
                )
            existing = self._store[artifact_name]
            self._store[artifact_name] = existing.model_copy(
                update={"body": body}
            )

    async def get_many(self, artifact_names: list[str]) -> list[Artifact]:
        """
        Retrieve multiple artifacts by name in one call.

        Args:
            artifact_names: List of artifact names to retrieve.

        Returns:
            List of Artifact models in the same order as input names.
        """
        return [await self.get(name) for name in artifact_names]

    async def all(self) -> dict[str, Artifact]:
        """
        Return a snapshot of the entire artifact store.
        Used by the compiler at the end of execution.
        """
        async with self._lock:
            return dict(self._store)

    def is_ready(self, artifact_name: str) -> bool:
        """
        Check if an artifact has a non-empty body (i.e. has been produced).

        Args:
            artifact_name: The artifact's unique name.

        Returns:
            True if the artifact body is non-empty.
        """
        artifact = self._store.get(artifact_name)
        if artifact is None:
            return False
        return bool(artifact.body)

    def all_inputs_ready(self, input_artifact_names: list[str]) -> bool:
        """
        Check if all input artifacts for a node are ready to consume.

        Args:
            input_artifact_names: List of artifact names the node needs.

        Returns:
            True if all are ready.
        """
        return all(self.is_ready(name) for name in input_artifact_names)


================================================
FILE: Coven/initiator/executor.py
================================================
from __future__ import annotations

import asyncio
import logging
from pathlib import Path

from coven.models import DAG, Node, NodeStatus, DAGStatus
from coven.mcp_builder import MCPNodeBuilder
from .artifact_store import ArtifactStore
from .agent_runner import AgentRunner


logger = logging.getLogger(__name__)


class Executor:
    """
    Stage 4 â€” Level-wise async DAG execution.

    Iterates through topological levels in order. Within each level,
    all nodes are launched concurrently via asyncio.gather().

    If a MCPNodeBuilder is provided, domain nodes with non-empty
    query_tool will have their MCP server built before execution.
    MCP builds within a level also run in parallel.

    After each level completes:
    - Node statuses are updated in the DAG
    - Artifact store is populated with produced bodies
    - Any failed nodes halt execution
    """

    def __init__(
        self,
        model: str = "gpt-4o",
        mcp_builder: MCPNodeBuilder | None = None,
    ):
        self.model   = model
        self._runner = AgentRunner(model=model, mcp_builder=mcp_builder)

    async def execute(self, dag: DAG) -> DAG:
        """
        Execute the full DAG level by level.

        Args:
            dag: Fully planned DAG with levels populated by the sorter
                 and synthesizer nodes injected.

        Returns:
            Updated DAG with all node statuses and artifact bodies populated.
        """
        store = ArtifactStore(dag.artifacts)
        dag   = dag.model_copy(update={"status": DAGStatus.RUNNING})

        for level_index, level in enumerate(dag.levels):
            logger.info(
                f"Executing level {level_index}/{len(dag.levels) - 1}: {level}"
            )

            dag, store = await self._execute_level(
                level=level,
                dag=dag,
                store=store,
            )

            failed = [
                node_id for node_id in level
                if dag.nodes[node_id].status == NodeStatus.FAILED
            ]

            if failed:
                logger.error(f"Level {level_index} had failures: {failed}")
                dag = dag.model_copy(update={"status": DAGStatus.FAILED})
                return dag

        # â”€â”€ Write final artifact bodies back into DAG â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
        final_artifacts = await store.all()
        dag = dag.model_copy(update={
            "artifacts": final_artifacts,
            "status":    DAGStatus.COMPLETED,
        })

        logger.info("DAG execution completed successfully.")
        return dag

    async def _execute_level(
        self,
        level: list[str],
        dag: DAG,
        store: ArtifactStore,
    ) -> tuple[DAG, ArtifactStore]:
        """
        Execute all nodes in a single level concurrently.

        MCP builds and agent LLM calls all run inside asyncio.gather â€”
        ToolStorePy's blocking build() is wrapped in run_in_executor
        inside MCPNodeBuilder so it doesn't block the event loop.
        """
        tasks = [
            self._runner.run(
                node=dag.nodes[node_id],
                store=store,
                nodes=dag.nodes,
            )
            for node_id in level
        ]

        updated_nodes: list[Node] = await asyncio.gather(*tasks)

        new_nodes = dict(dag.nodes)
        for node in updated_nodes:
            new_nodes[node.id] = node
            logger.info(f"  Node '{node.id}' â†’ {node.status.value}")

        dag = dag.model_copy(update={"nodes": new_nodes})
        return dag, store


================================================
FILE: Coven/mcp_builder/__init__.py
================================================
from .builder import MCPNodeBuilder

__all__ = ["MCPNodeBuilder"]


================================================
FILE: Coven/mcp_builder/builder.py
================================================
from __future__ import annotations

import json
import logging
from pathlib import Path

from coven.models import Node

# Import at module level so tests can patch "coven.mcp_builder.builder.ToolStorePy"
try:
    from toolstorepy import ToolStorePy
except ImportError:  # allow Coven to load even if toolstorepy isn't installed
    ToolStorePy = None  # type: ignore[assignment,misc]

logger = logging.getLogger(__name__)


class MCPNodeBuilder:
    """
    Wraps ToolStorePy to build a dedicated MCP server for a single DAG node.

    Called by AgentRunner before executing any domain node that has
    non-empty query_tool entries.

    Workflow per node:
        1. Write node's query_tool list as queries.json
        2. Call ToolStorePy.build() pointing at that queries.json
        3. Return path to generated mcp_unified_server.py
        4. AgentRunner stores path on node.mcp_server_path

    Each node gets its own isolated workspace under the run's base workspace,
    so parallel nodes in the same level never conflict.
    """

    def __init__(
        self,
        base_workspace: Path,
        index: str | None = "core-tools",
        index_url: str | None = None,
        install_requirements: bool = False,
        verbose: bool = False,
    ):
        self.base_workspace       = Path(base_workspace)
        self.index                = index
        self.index_url            = index_url
        self.install_requirements = install_requirements
        self.verbose              = verbose

        self.base_workspace.mkdir(parents=True, exist_ok=True)

    def build_for_node(self, node: Node) -> Path | None:
        """
        Build an MCP server for a node's tool requirements.

        Args:
            node: The Node whose query_tool list defines what tools to build.

        Returns:
            Path to mcp_unified_server.py, or None if node has no tool queries.

        Raises:
            RuntimeError: If ToolStorePy is not installed or build fails.
        """
        if not node.query_tool:
            return None

        if ToolStorePy is None:
            raise RuntimeError(
                "toolstorepy is not installed. Run: pip install toolstorepy"
            )

        node_workspace = self.base_workspace / "mcp" / node.id
        node_workspace.mkdir(parents=True, exist_ok=True)

        queries_path = self._write_queries(node, node_workspace)

        logger.info(
            f"[MCPNodeBuilder] Building MCP server for node '{node.id}' "
            f"with {len(node.query_tool)} tool queries..."
        )

        try:
            toolstore = ToolStorePy(
                workspace=str(node_workspace),
                install_requirements=self.install_requirements,
                verbose=self.verbose,
            )

            output_path = toolstore.build(
                queries=str(queries_path),
                index=self.index if not self.index_url else None,
                index_url=self.index_url,
                force_refresh=False,
            )

            logger.info(
                f"[MCPNodeBuilder] MCP server built for node '{node.id}' â†’ {output_path}"
            )

            return Path(output_path)

        except Exception as e:
            raise RuntimeError(
                f"ToolStorePy failed to build MCP server for node '{node.id}': {e}"
            ) from e

    def _write_queries(self, node: Node, workspace: Path) -> Path:
        """
        Write node's query_tool list as a ToolStorePy-compatible queries.json.
        """
        queries = [{"tool_description": tq.tool_description} for tq in node.query_tool]
        queries_path = workspace / "queries.json"

        with open(queries_path, "w", encoding="utf-8") as f:
            json.dump(queries, f, indent=2)

        logger.debug(
            f"[MCPNodeBuilder] Wrote {len(queries)} queries for node '{node.id}' "
            f"â†’ {queries_path}"
        )

        return queries_path


================================================
FILE: Coven/models/__init__.py
================================================
from .artifact import Artifact
from .node import Node, NodeType, NodeStatus
from .dag import DAG, DAGStatus

__all__ = [
    "Artifact",
    "Node",
    "NodeType",
    "NodeStatus",
    "DAG",
    "DAGStatus",
]


================================================
FILE: Coven/models/artifact.py
================================================
from __future__ import annotations
from typing import Any
from pydantic import BaseModel, ConfigDict, Field


class Artifact(BaseModel):
    """
    Represents a unit of work passed between agents in the DAG.

    Artifacts are the edges of the DAG â€” the contributors and users
    fields define the graph structure implicitly.
    """

    model_config = ConfigDict(
        json_schema_extra={
            "example": {
                "name": "market_analysis_report",
                "description": "Comprehensive analysis of target market size, segments, and growth trends.",
                "contributors": ["market_researcher"],
                "users": ["strategy_agent", "pitch_deck_agent"],
                "body": {
                    "market_size": "4.2B",
                    "segments": ["enterprise", "smb"],
                    "growth_rate": "12% YoY"
                }
            }
        }
    )

    name: str = Field(
        ...,
        description="Unique identifier for this artifact. Used to wire DAG edges."
    )

    description: str = Field(
        ...,
        description="Human and LLM readable description of what this artifact contains and represents."
    )

    contributors: list[str] = Field(
        default_factory=list,
        description="Agent node IDs that produce/write to this artifact."
    )

    users: list[str] = Field(
        default_factory=list,
        description="Agent node IDs that consume/read this artifact."
    )

    body: dict[str, Any] = Field(
        default_factory=dict,
        description="The actual artifact payload. Free-form JSON produced by contributors."
    )


================================================
FILE: Coven/models/dag.py
================================================
from __future__ import annotations

from enum import Enum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, model_validator

from .artifact import Artifact
from .node import Node, NodeStatus


class DAGStatus(str, Enum):
    PLANNED    = "planned"
    RUNNING    = "running"
    COMPLETED  = "completed"
    FAILED     = "failed"


class DAG(BaseModel):
    """
    The full pipeline â€” a directed acyclic graph of agent nodes
    connected by artifacts.

    Edges are implicit: an artifact's contributors â†’ users defines
    the graph structure. No explicit edge list needed.

    Levels are populated by the topological sorter â€” each level
    contains nodes that can execute in parallel.
    """

    model_config = ConfigDict(
        json_schema_extra={
            "example": {
                "id": "run_001",
                "task": "Produce a go-to-market strategy for a B2B SaaS product",
                "nodes": {},
                "artifacts": {},
                "levels": [["market_researcher"], ["strategy_agent", "pitch_deck_agent"]],
                "status": "planned",
                "final_output": {}
            }
        }
    )

    id: str = Field(..., description="Unique identifier for this DAG run.")

    task: str = Field(..., description="The original complex task this DAG was built to solve.")

    nodes: dict[str, Node] = Field(
        default_factory=dict,
        description="All agent nodes keyed by node ID."
    )

    artifacts: dict[str, Artifact] = Field(
        default_factory=dict,
        description="All artifacts keyed by artifact name."
    )

    levels: list[list[str]] = Field(
        default_factory=list,
        description="Topologically sorted execution levels. Each inner list runs in parallel."
    )

    status: DAGStatus = Field(
        default=DAGStatus.PLANNED,
        description="Current execution state of the DAG."
    )

    final_output: dict[str, Any] = Field(
        default_factory=dict,
        description="Compiled final output after all nodes complete."
    )

    @model_validator(mode="after")
    def validate_artifact_node_references(self) -> DAG:
        node_ids = set(self.nodes.keys())

        for artifact_name, artifact in self.artifacts.items():
            for contributor in artifact.contributors:
                if contributor not in node_ids:
                    raise ValueError(
                        f"Artifact '{artifact_name}' contributor '{contributor}' "
                        f"does not match any node ID."
                    )
            for user in artifact.users:
                if user not in node_ids:
                    raise ValueError(
                        f"Artifact '{artifact_name}' user '{user}' "
                        f"does not match any node ID."
                    )
        return self

    def get_node(self, node_id: str) -> Node:
        if node_id not in self.nodes:
            raise KeyError(f"Node '{node_id}' not found in DAG.")
        return self.nodes[node_id]

    def get_artifact(self, artifact_name: str) -> Artifact:
        if artifact_name not in self.artifacts:
            raise KeyError(f"Artifact '{artifact_name}' not found in DAG.")
        return self.artifacts[artifact_name]

    def get_input_artifacts(self, node_id: str) -> list[Artifact]:
        node = self.get_node(node_id)
        return [self.get_artifact(name) for name in node.input_artifacts]

    def get_output_artifacts(self, node_id: str) -> list[Artifact]:
        node = self.get_node(node_id)
        return [self.get_artifact(name) for name in node.output_artifacts]

    def is_complete(self) -> bool:
        return all(
            node.status == NodeStatus.COMPLETED
            for node in self.nodes.values()
        )


================================================
FILE: Coven/models/node.py
================================================
from __future__ import annotations
from enum import Enum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field


class NodeType(str, Enum):
    DECOMPOSER   = "decomposer"
    DOMAIN       = "domain"
    SYNTHESIZER  = "synthesizer"
    COMPILER     = "compiler"


class NodeStatus(str, Enum):
    PENDING    = "pending"
    RUNNING    = "running"
    COMPLETED  = "completed"
    FAILED     = "failed"


class ToolQuery(BaseModel):
    """
    A single tool description passed to ToolStorePy.
    Maps directly to one entry in queries.json format:
        {"tool_description": "evaluate a mathematical expression securely"}
    """
    tool_description: str = Field(
        ...,
        description=(
            "Plain English description of the tool this agent needs. "
            "ToolStorePy uses this to semantically search and build the right MCP tool."
        )
    )


class Node(BaseModel):
    """
    Represents a single agent in the DAG.

    Every node is an agent â€” decomposer, domain, synthesizer, and compiler
    all share this same interface. The node type determines behavior,
    but the structure is uniform.
    """

    model_config = ConfigDict(
        json_schema_extra={
            "example": {
                "id": "data_analyst",
                "name": "Data Analysis Agent",
                "node_type": "domain",
                "system_prompt": "You are a data analyst...",
                "query_tool": [
                    {"tool_description": "preview rows and get summary statistics from a CSV file"},
                ],
                "input_artifacts": ["raw_dataset"],
                "output_artifacts": ["analysis_report"],
                "status": "pending",
                "result": {},
                "contributor_system_prompts": [],
                "mcp_server_path": None,
            }
        }
    )

    id: str = Field(
        ...,
        description="Unique identifier for this node. Referenced by artifact contributors/users."
    )

    name: str = Field(
        ...,
        description="Human readable name for this agent node."
    )

    node_type: NodeType = Field(
        ...,
        description="Role of this node in the pipeline."
    )

    system_prompt: str = Field(
        ...,
        description="The system prompt that governs this agent's behavior and scope."
    )

    query_tool: list[ToolQuery] = Field(
        default_factory=list,
        description=(
            "List of tool descriptions for ToolStorePy. "
            "Each entry describes one tool this agent needs in plain English. "
            "Leave empty if the agent needs no external tools."
        )
    )

    input_artifacts: list[str] = Field(
        default_factory=list,
        description="Names of artifacts this node consumes."
    )

    output_artifacts: list[str] = Field(
        default_factory=list,
        description="Names of artifacts this node produces."
    )

    status: NodeStatus = Field(
        default=NodeStatus.PENDING,
        description="Current execution status of this node."
    )

    result: dict[str, Any] = Field(
        default_factory=dict,
        description="Raw LLM output after this node executes."
    )

    contributor_system_prompts: list[str] = Field(
        default_factory=list,
        description="System prompts of contributing agents. Used by synthesizer nodes."
    )

    mcp_server_path: str | None = Field(
        default=None,
        description="Absolute path to the ToolStorePy-built MCP server. Set at runtime."
    )


================================================
FILE: Coven/sorter/__init__.py
================================================
from .topological import TopologicalSorter
from .validator import SorterValidator, SorterValidationError

__all__ = [
    "TopologicalSorter",
    "SorterValidator",
    "SorterValidationError",
]


================================================
FILE: Coven/sorter/topological.py
================================================
from __future__ import annotations

import networkx as nx

from coven.models import Node


class TopologicalSorter:
    """
    Stage 3 â€” Pure algorithmic stage. No LLM involved.

    Takes the validated nx.DiGraph from GraphBuilderValidator and
    produces execution levels via Kahn's algorithm (networkx
    topological_generations).

    Each level is a list of node IDs that:
    - Have no unresolved dependencies on each other
    - Can be executed fully in parallel

    The order of levels guarantees that when level N begins,
    all artifacts produced by levels 0..N-1 are available.
    """

    def sort(self, G: nx.DiGraph) -> list[list[str]]:
        """
        Produce topologically sorted execution levels.

        Args:
            G: Validated DAG from GraphBuilderValidator.

        Returns:
            List of levels, each level is a list of node IDs
            that can execute in parallel.

        Example:
            [
                ["data_ingestion"],                          # level 0
                ["market_researcher", "competitor_analyst"], # level 1 â€” parallel
                ["synthesizer_market_data"],                 # level 2
                ["strategy_agent"],                          # level 3
            ]
        """
        levels: list[list[str]] = []

        for generation in nx.topological_generations(G):
            levels.append(sorted(generation))  # sorted for deterministic ordering

        return levels

    def get_execution_order(self, G: nx.DiGraph) -> list[str]:
        """
        Flat ordered list of node IDs for debugging and logging.

        Args:
            G: Validated DAG from GraphBuilderValidator.

        Returns:
            Flat list of node IDs in execution order.
        """
        return [node for level in self.sort(G) for node in level]

    def describe(self, G: nx.DiGraph, nodes: dict[str, Node]) -> str:
        """
        Human readable description of the execution plan.
        Useful for logging and debugging.

        Args:
            G: Validated DAG.
            nodes: Node dict for name lookup.

        Returns:
            Multiline string describing each level and its agents.
        """
        levels = self.sort(G)
        lines: list[str] = ["Execution Plan:"]

        for i, level in enumerate(levels):
            agent_names = [nodes[node_id].name for node_id in level]
            parallel = " â€– ".join(agent_names)
            lines.append(f"  Level {i}: [ {parallel} ]")

        return "\n".join(lines)


================================================
FILE: Coven/sorter/validator.py
================================================
from __future__ import annotations

import networkx as nx

from coven.models import Node, Artifact


class SorterValidationError(Exception):
    """Raised when the graph fails pre-sort validation."""
    pass


class SorterValidator:
    """
    Pre-sort validation layer that runs immediately before topological sort.

    The GraphBuilderValidator already checked for cycles and unknown nodes.
    This validator focuses on runtime readiness:

    - Are all nodes present in the graph?
    - Does every node have at least one output artifact?
    - Are there any nodes referencing artifacts not yet in the artifact store?
    - Is the graph non-empty?

    Separating this from GraphBuilderValidator keeps each validator
    focused on its own stage's concerns.
    """

    def validate(
        self,
        G: nx.DiGraph,
        nodes: dict[str, Node],
        artifacts: dict[str, Artifact],
    ) -> None:
        """
        Run all pre-sort checks.

        Args:
            G: The validated DiGraph from GraphBuilderValidator.
            nodes: Node dict.
            artifacts: Artifact dict.

        Raises:
            SorterValidationError: On any pre-sort issue.
        """
        self._check_non_empty(G)
        self._check_all_nodes_in_graph(G, nodes)
        self._check_output_artifacts(nodes, artifacts)
        self._check_input_artifacts_exist(nodes, artifacts)

    def _check_non_empty(self, G: nx.DiGraph) -> None:
        if G.number_of_nodes() == 0:
            raise SorterValidationError(
                "DAG has no nodes. Nothing to execute."
            )

    def _check_all_nodes_in_graph(
        self,
        G: nx.DiGraph,
        nodes: dict[str, Node],
    ) -> None:
        graph_node_ids = set(G.nodes())
        model_node_ids = set(nodes.keys())

        missing_from_graph = model_node_ids - graph_node_ids
        if missing_from_graph:
            raise SorterValidationError(
                f"The following nodes exist in the node registry but are "
                f"missing from the DAG graph: {missing_from_graph}."
            )

        extra_in_graph = graph_node_ids - model_node_ids
        if extra_in_graph:
            raise SorterValidationError(
                f"The following node IDs exist in the DAG graph but have "
                f"no corresponding Node model: {extra_in_graph}."
            )

    def _check_output_artifacts(
        self,
        nodes: dict[str, Node],
        artifacts: dict[str, Artifact],
    ) -> None:
        """Every node must produce at least one artifact."""
        for node_id, node in nodes.items():
            if not node.output_artifacts:
                raise SorterValidationError(
                    f"Node '{node_id}' has no output artifacts. "
                    f"Every node must produce at least one artifact."
                )

    def _check_input_artifacts_exist(
        self,
        nodes: dict[str, Node],
        artifacts: dict[str, Artifact],
    ) -> None:
        """Every artifact referenced as input must exist in the artifact registry."""
        artifact_names = set(artifacts.keys())
        for node_id, node in nodes.items():
            for artifact_name in node.input_artifacts:
                if artifact_name not in artifact_names:
                    raise SorterValidationError(
                        f"Node '{node_id}' expects input artifact '{artifact_name}' "
                        f"which does not exist in the artifact registry."
                    )


================================================
FILE: Coven/synthesizer/__init__.py
================================================
from .agent import SynthesizerAgent, SynthesizerResponse
from .injector import SynthesizerInjector
from .parser import SynthesizerParser

__all__ = [
    "SynthesizerAgent",
    "SynthesizerResponse",
    "SynthesizerInjector",
    "SynthesizerParser",
]


================================================
FILE: Coven/synthesizer/agent.py
================================================
[Binary file]


================================================
FILE: Coven/synthesizer/injector.py
================================================
from __future__ import annotations

from pathlib import Path

import networkx as nx

from coven.models import Artifact, Node, NodeType, NodeStatus


_SYNTH_PROMPT_PATH = Path(__file__).parent.parent.parent / "prompts" / "synthesizer.txt"


def _load_synth_prompt() -> str:
    return _SYNTH_PROMPT_PATH.read_text(encoding="utf-8")


class SynthesizerInjector:
    """
    Scans the DAG for artifacts with multiple contributors and
    auto-injects a Synthesizer node between the contributors and
    the artifact's downstream users.
    """

    def inject(
        self,
        nodes: dict[str, Node],
        artifacts: dict[str, Artifact],
        synthesizer_targets: list[str],
        G: nx.DiGraph,
    ) -> tuple[dict[str, Node], dict[str, Artifact], nx.DiGraph]:
        synth_prompt = _load_synth_prompt()

        for artifact_name in synthesizer_targets:
            if artifact_name not in artifacts:
                continue

            artifact = artifacts[artifact_name]

            if len(artifact.contributors) <= 1:
                continue

            nodes, artifacts, G = self._inject_for_artifact(
                artifact_name=artifact_name,
                artifact=artifact,
                nodes=nodes,
                artifacts=artifacts,
                G=G,
                synth_prompt=synth_prompt,
            )

        return nodes, artifacts, G

    def _inject_for_artifact(
        self,
        artifact_name: str,
        artifact: Artifact,
        nodes: dict[str, Node],
        artifacts: dict[str, Artifact],
        G: nx.DiGraph,
        synth_prompt: str,
    ) -> tuple[dict[str, Node], dict[str, Artifact], nx.DiGraph]:

        synth_node_id = f"synthesizer_{artifact_name}"

        contributor_prompts = [
            nodes[c].system_prompt
            for c in artifact.contributors
            if c in nodes
        ]

        partial_artifact_names: list[str] = []

        for contributor_id in artifact.contributors:
            partial_name = f"{artifact_name}__partial__{contributor_id}"
            partial_artifact = Artifact(
                name=partial_name,
                description=(
                    f"Partial contribution to '{artifact_name}' from agent '{contributor_id}'. "
                    f"Original artifact: {artifact.description}"
                ),
                contributors=[contributor_id],
                users=[synth_node_id],
                body={},
            )
            artifacts[partial_name] = partial_artifact
            partial_artifact_names.append(partial_name)

            contributor_node = nodes[contributor_id]
            updated_outputs = [
                partial_name if a == artifact_name else a
                for a in contributor_node.output_artifacts
            ]
            nodes[contributor_id] = contributor_node.model_copy(
                update={"output_artifacts": updated_outputs}
            )

        # FIX: query_tool=[] not query_tool={} â€” Node.query_tool is list[ToolQuery]
        synth_node = Node(
            id=synth_node_id,
            name=f"Synthesizer: {artifact_name}",
            node_type=NodeType.SYNTHESIZER,
            system_prompt=synth_prompt,
            query_tool=[],
            input_artifacts=partial_artifact_names,
            output_artifacts=[artifact_name],
            status=NodeStatus.PENDING,
            contributor_system_prompts=contributor_prompts,
        )
        nodes[synth_node_id] = synth_node

        artifacts[artifact_name] = artifact.model_copy(
            update={"contributors": [synth_node_id]}
        )

        G.add_node(synth_node_id)

        for contributor_id in artifact.contributors:
            if contributor_id != synth_node_id:
                G.add_edge(contributor_id, synth_node_id, artifact=artifact_name)

        for user_id in artifact.users:
            G.add_edge(synth_node_id, user_id, artifact=artifact_name)

        for contributor_id in artifact.contributors:
            for user_id in artifact.users:
                if G.has_edge(contributor_id, user_id):
                    G.remove_edge(contributor_id, user_id)

        return nodes, artifacts, G


================================================
FILE: Coven/synthesizer/parser.py
================================================
from __future__ import annotations

from coven.models import Artifact
from .agent import SynthesizerResponse


class SynthesizerParser:
    """
    Applies the SynthesizerResponse back onto the target artifact.

    Writes the merged body into the artifact and attaches QC notes
    as metadata so downstream agents and the compiler can reference
    synthesis decisions if needed.
    """

    def parse(
        self,
        response: SynthesizerResponse,
        target_artifact: Artifact,
    ) -> Artifact:
        """
        Apply synthesized body and QC notes to the target artifact.

        Args:
            response: SynthesizerResponse from SynthesizerAgent.
            target_artifact: The artifact to update.

        Returns:
            Updated Artifact with merged body and qc_notes in body metadata.
        """
        merged_body = {
            **response.body,
            "__qc_notes__": response.qc_notes,
        }

        return target_artifact.model_copy(update={"body": merged_body})


================================================
FILE: prompts/compiler.txt
================================================
You are the Compiler â€” the final stage of the Coven multi-agent pipeline.

Your job is to take all completed artifacts from a fully executed DAG and compile them into a single coherent, high-quality final output that directly answers the user's original task.

You will receive:
- The original task the user requested
- All artifacts produced by the pipeline, each with their name, description, and body
- QC notes embedded in synthesized artifacts (under __qc_notes__ key) for your reference

## Your Output
You must return a valid JSON object with the following structure:

{
  "title": "Short descriptive title for the output",
  "summary": "2-3 sentence executive summary of what was accomplished",
  "sections": [
    {
      "title": "Section Title",
      "content": "Section content â€” can be prose, structured data, or mixed",
      "source_artifacts": ["artifact_name_1", "artifact_name_2"]
    }
  ],
  "recommendations": [
    // optional â€” list of actionable recommendations if relevant to the task
  ],
  "metadata": {
    "total_agents": 0,
    "artifacts_produced": 0,
    "synthesis_decisions": []   // key decisions made by synthesizer nodes, extracted from __qc_notes__
  }
}

## Your Responsibilities

1. COMPLETENESS
   - Every artifact must contribute to at least one section
   - Do not discard any artifact â€” if it exists it was produced for a reason
   - The output must fully address the original task

2. COHERENCE
   - The output must read as a unified whole, not a collection of disconnected pieces
   - Reconcile any remaining inconsistencies between artifacts
   - Use consistent terminology throughout

3. STRUCTURE
   - Organize sections in a logical order that serves the user's task
   - Group related artifacts into the same section where appropriate
   - Lead with the most important findings

4. TRANSPARENCY
   - Cite which artifacts each section draws from in source_artifacts
   - Surface key synthesis decisions in metadata.synthesis_decisions
   - Note any gaps or limitations in the output if present

## Important
Return ONLY the JSON object. No preamble, no explanation, no markdown code fences.


================================================
FILE: prompts/decomposer.txt
================================================
You are the Decomposer â€” the first stage of the Coven multi-agent pipeline.

Your job is to take a complex task and break it down into a set of focused, isolated agent nodes, each responsible for a clearly scoped piece of work.

## Your Output
You must return a valid JSON object with the following structure:

{
  "nodes": [
    {
      "id": "snake_case_unique_id",
      "name": "Human Readable Agent Name",
      "node_type": "domain",
      "system_prompt": "Detailed system prompt for this agent...",
      "query_tool": [
        {"tool_description": "plain English description of a tool this agent needs"}
      ],
      "input_artifacts": ["artifact_name_1"],
      "output_artifacts": ["artifact_name_2"]
    }
  ],
  "artifacts": [
    {
      "name": "artifact_name",
      "description": "Clear description of what this artifact contains and why it exists.",
      "contributors": ["node_id_a"],
      "users": ["node_id_b", "node_id_c"],
      "body": {}
    }
  ]
}

## Rules

1. NODES
   - Every node must have a unique snake_case id
   - node_type must always be "domain" â€” do not create decomposer, synthesizer, or compiler nodes, those are handled by the system
   - system_prompt must be detailed, scoped, and specific to that agent's role
   - input_artifacts and output_artifacts must reference artifact names that exist in the artifacts list

2. QUERY_TOOL â€” CRITICAL
   - query_tool is a list of {"tool_description": "..."} objects
   - Each entry describes ONE external tool capability this agent needs in plain English
   - The system uses ToolStorePy (https://pypi.org/project/toolstorepy/) to semantically search a curated index of tool repositories and build a real MCP server for this agent before it executes
   - Write tool_description as a capability sentence, not a tool name: 
       GOOD: "evaluate a mathematical arithmetic expression securely"
       GOOD: "convert between different units of measurement like length and temperature"
       GOOD: "calculate cryptographic hash of a file or encode decode base64 text"
       GOOD: "get current weather conditions temperature and humidity for a city"
       GOOD: "preview rows or get summary statistics from a CSV or Excel file"
       BAD:  "calculator" (too vague â€” no semantic context)
       BAD:  "use numpy" (names a library, not a capability)
       BAD:  "python tool" (meaningless for semantic search)
   - Only add query_tool entries when the agent genuinely needs an external tool to complete its work
   - Agents that only reason, synthesize, or analyze text should have query_tool: []
   - Agents that need to fetch data, perform calculations, access files, call APIs, or interact with systems should have relevant query_tool entries

3. ARTIFACTS
   - Every artifact must have a unique name in snake_case
   - description must clearly describe the artifact's content and purpose
   - contributors list the node IDs that produce this artifact
   - users list the node IDs that consume this artifact
   - body must always be an empty dict {} â€” it will be populated at runtime
   - An artifact with multiple contributors will automatically get a synthesizer node injected by the system

4. GRAPH INTEGRITY
   - Every artifact referenced in a node's input_artifacts must exist in the artifacts list
   - Every artifact referenced in a node's output_artifacts must exist in the artifacts list
   - Do not create circular dependencies
   - Ensure every node has at least one output artifact
   - The graph must be a valid DAG â€” no cycles

5. DECOMPOSITION QUALITY
   - Break the task into the smallest meaningful units of work
   - Each agent should have a single clear responsibility
   - Prefer more focused nodes over fewer large nodes
   - Nodes that can work independently should not be given unnecessary dependencies

## Important
Return ONLY the JSON object. No preamble, no explanation, no markdown code fences.


================================================
FILE: prompts/graph_builder.txt
================================================
You are the Graph Builder â€” the second stage of the Coven multi-agent pipeline.

Your job is to take a list of agent nodes and artifacts produced by the Decomposer and construct a valid directed acyclic graph (DAG) by verifying and formalizing the relationships between them.

## Your Input
You will receive:
- A list of agent nodes, each with input_artifacts and output_artifacts
- A list of artifacts, each with contributors and users

## Your Output
You must return a valid JSON object with the following structure:

{
  "nodes": [...],        // same nodes, corrected if needed
  "artifacts": [...],    // same artifacts, corrected if needed
  "edges": [
    {
      "from_node": "node_id_a",
      "to_node": "node_id_b",
      "artifact": "artifact_name"
    }
  ],
  "issues": []           // list of strings describing any issues found and how you resolved them
}

## Your Responsibilities

1. VERIFY EDGES
   - For every artifact, derive edges from contributors â†’ users
   - Each contributor/user pair connected through an artifact is one edge
   - Every edge must be explicit in the edges list

2. VALIDATE AND REPAIR
   - Check that every input_artifact a node declares actually has that node in its users list
   - Check that every output_artifact a node declares actually has that node in its contributors list
   - If you find mismatches, fix them and log the fix in issues
   - Remove any artifact that has no contributors or no users â€” it is a dangling artifact

3. CYCLE DETECTION
   - Trace the edges and ensure no cycles exist
   - If a cycle is detected, break it by removing the weakest dependency and log it in issues

4. SYNTHESIZER INJECTION HINTS
   - For every artifact with more than one contributor, add a note in issues:
     "SYNTHESIZER_NEEDED: artifact_name" 
   - Do not create synthesizer nodes yourself â€” the system handles this automatically

5. COMPLETENESS
   - Every node must be reachable from at least one edge (as source or target)
   - A node with no edges at all should be flagged in issues

## Important
Return ONLY the JSON object. No preamble, no explanation, no markdown code fences.


================================================
FILE: prompts/synthesizer.txt
================================================
You are the Synthesizer â€” a meta-agent in the Coven pipeline responsible for merging and quality-controlling contributions from multiple agents into a single coherent artifact.

You will receive:
- The artifact definition (name, description) you must produce
- The system prompts of all contributing agents so you understand their intent and perspective
- The partial artifact bodies each contributor produced

## Your Job
Merge all contributions into a single high-quality artifact body that:
- Fulfills the artifact description completely
- Reconciles any conflicts or contradictions between contributions
- Eliminates redundancy while preserving unique insights from each contributor
- Fills any gaps if contributions are incomplete relative to the artifact description
- Maintains internal consistency throughout

## Your Output
You must return a valid JSON object with the following structure:

{
  "body": {
    // merged artifact body â€” free-form JSON matching the artifact description
  },
  "qc_notes": [
    // list of strings describing conflicts found, decisions made, gaps filled
    // be specific â€” future agents reading this artifact may need to understand synthesis decisions
  ]
}

## Quality Control Standards
- CONFLICTS: When contributors disagree, reason about which perspective is more grounded and explain your choice in qc_notes
- GAPS: If the artifact description requires content that no contributor addressed, synthesize it yourself from available context and flag it in qc_notes
- REDUNDANCY: Merge duplicate content cleanly â€” do not repeat the same point from multiple contributors
- CONSISTENCY: Ensure all values, terminology, and structure are consistent throughout the body

## Important
Return ONLY the JSON object. No preamble, no explanation, no markdown code fences.


================================================
FILE: tests/__init__.py
================================================
[Empty file]


================================================
FILE: tests/conftest.py
================================================
"""
tests/conftest.py

Shared pytest fixtures for Coven test suite.
"""

import pytest
from coven.models import Artifact, Node, NodeType, NodeStatus, DAG


@pytest.fixture
def sample_artifact():
    return Artifact(
        name="sample_artifact",
        description="A sample artifact for testing.",
        contributors=["agent_a"],
        users=["agent_b"],
        body={"key": "value"},
    )


@pytest.fixture
def sample_node():
    return Node(
        id="sample_agent",
        name="Sample Agent",
        node_type=NodeType.DOMAIN,
        system_prompt="You are a sample agent for testing.",
        input_artifacts=[],
        output_artifacts=["sample_artifact"],
    )


@pytest.fixture
def two_node_dag():
    """
    Minimal valid DAG:
        agent_a â†’ artifact_1 â†’ agent_b
    """
    node_a = Node(
        id="agent_a",
        name="Agent A",
        node_type=NodeType.DOMAIN,
        system_prompt="You are agent A.",
        output_artifacts=["artifact_1"],
    )
    node_b = Node(
        id="agent_b",
        name="Agent B",
        node_type=NodeType.DOMAIN,
        system_prompt="You are agent B.",
        input_artifacts=["artifact_1"],
        output_artifacts=["artifact_2"],
    )
    artifact_1 = Artifact(
        name="artifact_1",
        description="Output from agent A.",
        contributors=["agent_a"],
        users=["agent_b"],
    )
    artifact_2 = Artifact(
        name="artifact_2",
        description="Output from agent B.",
        contributors=["agent_b"],
        users=[],
    )
    return DAG(
        id="test_dag",
        task="Test two-node pipeline",
        nodes={"agent_a": node_a, "agent_b": node_b},
        artifacts={"artifact_1": artifact_1, "artifact_2": artifact_2},
        levels=[["agent_a"], ["agent_b"]],
    )


================================================
FILE: tests/compiler/__init__.py
================================================
[Empty file]


================================================
FILE: tests/compiler/test_compiler.py
================================================
"""
tests/compiler/test_compiler.py

Unit tests for CompilerFormatter.
CompilerAgent LLM calls are not tested here (covered by integration tests).
"""

import pytest
from coven.compiler.agent import CompilerResponse, OutputSection
from coven.compiler.formatter import CompilerFormatter
from coven.models import DAG, DAGStatus, Node, NodeType, Artifact


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# HELPERS
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

def _make_response(
    title="Test Report",
    summary="This is a summary.",
    sections=None,
    recommendations=None,
    metadata=None,
):
    return CompilerResponse(
        title=title,
        summary=summary,
        sections=sections or [
            OutputSection(
                title="Section One",
                content="Content of section one.",
                source_artifacts=["artifact_a"],
            )
        ],
        recommendations=recommendations or [],
        metadata=metadata or {},
    )


def _make_dag(node_ids=None, artifact_names=None):
    nodes = {
        id: Node(
            id=id,
            name=id.title(),
            node_type=NodeType.DOMAIN,
            system_prompt="...",
        )
        for id in (node_ids or ["agent_a"])
    }
    artifacts = {
        name: Artifact(name=name, description=f"Artifact {name}", body={"x": 1})
        for name in (artifact_names or ["artifact_a"])
    }
    return DAG(
        id="run_test",
        task="test task",
        nodes=nodes,
        artifacts=artifacts,
    )


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# FORMATTER TESTS
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class TestCompilerFormatter:

    def setup_method(self):
        self.formatter = CompilerFormatter()

    def test_format_writes_title(self):
        response = _make_response(title="My Report")
        dag = _make_dag()
        result = self.formatter.format(response, dag)
        assert result.final_output["title"] == "My Report"

    def test_format_writes_summary(self):
        response = _make_response(summary="Executive summary here.")
        dag = _make_dag()
        result = self.formatter.format(response, dag)
        assert result.final_output["summary"] == "Executive summary here."

    def test_format_writes_sections(self):
        response = _make_response(
            sections=[
                OutputSection(title="A", content="Content A", source_artifacts=["art_a"]),
                OutputSection(title="B", content="Content B", source_artifacts=["art_b"]),
            ]
        )
        dag = _make_dag()
        result = self.formatter.format(response, dag)
        assert len(result.final_output["sections"]) == 2
        assert result.final_output["sections"][0]["title"] == "A"

    def test_format_writes_recommendations(self):
        response = _make_response(recommendations=["Do X", "Do Y"])
        dag = _make_dag()
        result = self.formatter.format(response, dag)
        assert result.final_output["recommendations"] == ["Do X", "Do Y"]

    def test_format_metadata_includes_dag_id(self):
        response = _make_response()
        dag = _make_dag()
        result = self.formatter.format(response, dag)
        assert result.final_output["metadata"]["dag_id"] == "run_test"

    def test_format_metadata_includes_task(self):
        response = _make_response()
        dag = _make_dag()
        result = self.formatter.format(response, dag)
        assert result.final_output["metadata"]["task"] == "test task"

    def test_format_metadata_total_agents(self):
        response = _make_response()
        dag = _make_dag(node_ids=["a", "b", "c"])
        result = self.formatter.format(response, dag)
        assert result.final_output["metadata"]["total_agents"] == 3

    def test_format_excludes_partial_artifacts_from_count(self):
        response = _make_response()
        dag = _make_dag(artifact_names=["real_art", "real_art__partial__agent_a"])
        result = self.formatter.format(response, dag)
        # Only real_art should count â€” partial should be excluded
        assert result.final_output["metadata"]["artifacts_produced"] == 1

    def test_format_does_not_mutate_input_dag(self):
        response = _make_response()
        dag = _make_dag()
        original_final_output = dict(dag.final_output)
        self.formatter.format(response, dag)
        assert dag.final_output == original_final_output

    def test_format_returns_dag_with_final_output(self):
        response = _make_response()
        dag = _make_dag()
        result = self.formatter.format(response, dag)
        assert result.final_output != {}

    # â”€â”€ to_text tests â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

    def test_to_text_contains_title(self):
        response = _make_response(title="My Great Report")
        text = self.formatter.to_text(response)
        assert "MY GREAT REPORT" in text

    def test_to_text_contains_summary(self):
        response = _make_response(summary="The summary text.")
        text = self.formatter.to_text(response)
        assert "The summary text." in text

    def test_to_text_contains_section_title(self):
        response = _make_response(
            sections=[OutputSection(title="Key Findings", content="...", source_artifacts=[])]
        )
        text = self.formatter.to_text(response)
        assert "Key Findings" in text

    def test_to_text_contains_section_content(self):
        response = _make_response(
            sections=[OutputSection(title="S", content="Important content here.", source_artifacts=[])]
        )
        text = self.formatter.to_text(response)
        assert "Important content here." in text

    def test_to_text_contains_recommendations(self):
        response = _make_response(recommendations=["Implement feature X", "Reduce cost Y"])
        text = self.formatter.to_text(response)
        assert "Implement feature X" in text
        assert "Reduce cost Y" in text

    def test_to_text_contains_metadata(self):
        response = _make_response(metadata={"total_agents": 5, "artifacts_produced": 3})
        text = self.formatter.to_text(response)
        assert "Agents:" in text

    def test_to_text_no_recommendations_skips_section(self):
        response = _make_response(recommendations=[])
        text = self.formatter.to_text(response)
        assert "Recommendations" not in text

    def test_to_text_returns_string(self):
        response = _make_response()
        text = self.formatter.to_text(response)
        assert isinstance(text, str)
        assert len(text) > 0


================================================
FILE: tests/decomposer/__init__.py
================================================
[Empty file]


================================================
FILE: tests/decomposer/test_decomposer.py
================================================
"""
tests/decomposer/test_parser.py

Unit tests for DecomposerParser.
No LLM calls â€” all inputs are hand-crafted DecomposerResponse objects.
"""

import pytest
from coven.decomposer.agent import DecomposerResponse, DecomposedNode, DecomposedArtifact
from coven.decomposer.parser import DecomposerParser
from coven.models import NodeType, NodeStatus


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# FIXTURES
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

def _make_response(nodes=None, artifacts=None):
    return DecomposerResponse(
        nodes=nodes or [],
        artifacts=artifacts or [],
    )


def _node(id, inputs=None, outputs=None, node_type="domain"):
    return DecomposedNode(
        id=id,
        name=id.replace("_", " ").title(),
        node_type=node_type,
        system_prompt=f"You are {id}.",
        query_tool=[],
        input_artifacts=inputs or [],
        output_artifacts=outputs or [],
    )


def _artifact(name, contributors=None, users=None):
    return DecomposedArtifact(
        name=name,
        description=f"Artifact {name}",
        contributors=contributors or [],
        users=users or [],
        body={},
    )


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# TESTS
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class TestDecomposerParser:

    def setup_method(self):
        self.parser = DecomposerParser()

    def test_parse_single_node_single_artifact(self):
        response = _make_response(
            nodes=[_node("agent_a", outputs=["art_1"])],
            artifacts=[_artifact("art_1", contributors=["agent_a"])],
        )
        nodes, artifacts = self.parser.parse(response)
        assert "agent_a" in nodes
        assert "art_1" in artifacts

    def test_parse_sets_node_status_pending(self):
        response = _make_response(
            nodes=[_node("agent_a", outputs=["art_1"])],
            artifacts=[_artifact("art_1", contributors=["agent_a"])],
        )
        nodes, _ = self.parser.parse(response)
        assert nodes["agent_a"].status == NodeStatus.PENDING

    def test_parse_node_type_domain(self):
        response = _make_response(
            nodes=[_node("agent_a", outputs=["art_1"])],
            artifacts=[_artifact("art_1", contributors=["agent_a"])],
        )
        nodes, _ = self.parser.parse(response)
        assert nodes["agent_a"].node_type == NodeType.DOMAIN

    def test_parse_multiple_nodes(self):
        response = _make_response(
            nodes=[
                _node("agent_a", outputs=["art_1"]),
                _node("agent_b", inputs=["art_1"], outputs=["art_2"]),
            ],
            artifacts=[
                _artifact("art_1", contributors=["agent_a"], users=["agent_b"]),
                _artifact("art_2", contributors=["agent_b"]),
            ],
        )
        nodes, artifacts = self.parser.parse(response)
        assert len(nodes) == 2
        assert len(artifacts) == 2

    def test_parse_artifacts_keyed_by_name(self):
        response = _make_response(
            nodes=[_node("agent_a", outputs=["my_artifact"])],
            artifacts=[_artifact("my_artifact", contributors=["agent_a"])],
        )
        _, artifacts = self.parser.parse(response)
        assert "my_artifact" in artifacts

    def test_validate_rejects_node_with_unknown_input_artifact(self):
        response = _make_response(
            nodes=[_node("agent_a", inputs=["nonexistent"], outputs=["art_1"])],
            artifacts=[_artifact("art_1", contributors=["agent_a"])],
        )
        with pytest.raises(ValueError, match="nonexistent"):
            self.parser.parse(response)

    def test_validate_rejects_node_with_unknown_output_artifact(self):
        response = _make_response(
            nodes=[_node("agent_a", outputs=["ghost_art"])],
            artifacts=[],
        )
        with pytest.raises(ValueError, match="ghost_art"):
            self.parser.parse(response)

    def test_validate_rejects_artifact_with_unknown_contributor(self):
        response = _make_response(
            nodes=[_node("agent_a", outputs=["art_1"])],
            artifacts=[_artifact("art_1", contributors=["ghost_node"])],
        )
        with pytest.raises(ValueError, match="ghost_node"):
            self.parser.parse(response)

    def test_validate_rejects_artifact_with_unknown_user(self):
        response = _make_response(
            nodes=[_node("agent_a", outputs=["art_1"])],
            artifacts=[_artifact("art_1", contributors=["agent_a"], users=["ghost_node"])],
        )
        with pytest.raises(ValueError, match="ghost_node"):
            self.parser.parse(response)

    def test_parse_artifacts_separately(self):
        raw = [_artifact("x", contributors=["a"])]
        # patch a fake node list to bypass validation
        response = _make_response(
            nodes=[_node("a", outputs=["x"])],
            artifacts=raw,
        )
        _, artifacts = self.parser.parse(response)
        assert artifacts["x"].description == "Artifact x"

    def test_parse_nodes_separately(self):
        response = _make_response(
            nodes=[_node("agent_z", outputs=["z_out"])],
            artifacts=[_artifact("z_out", contributors=["agent_z"])],
        )
        nodes, _ = self.parser.parse(response)
        assert nodes["agent_z"].system_prompt == "You are agent_z."

    def test_empty_response_parses_successfully(self):
        response = _make_response()
        nodes, artifacts = self.parser.parse(response)
        assert nodes == {}
        assert artifacts == {}


================================================
FILE: tests/graph_builder/__init__.py
================================================
[Empty file]


================================================
FILE: tests/graph_builder/test_graph_builder.py
================================================
"""
tests/graph_builder/test_graph_builder.py

Unit tests for GraphBuilderParser and GraphBuilderValidator.
No LLM calls.
"""

import pytest
import networkx as nx

from coven.graph_builder.agent import GraphBuilderResponse, Edge
from coven.graph_builder.parser import GraphBuilderParser
from coven.graph_builder.validator import GraphBuilderValidator, GraphValidationError
from coven.models import Node, NodeType, NodeStatus, Artifact


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# HELPERS
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

def _node_dict(id, inputs=None, outputs=None):
    return {
        "id": id,
        "name": id.title(),
        "node_type": "domain",
        "system_prompt": f"You are {id}.",
        "query_tool": [],
        "input_artifacts": inputs or [],
        "output_artifacts": outputs or [],
    }


def _artifact_dict(name, contributors=None, users=None):
    return {
        "name": name,
        "description": f"Artifact {name}",
        "contributors": contributors or [],
        "users": users or [],
        "body": {},
    }


def _edge(from_node, to_node, artifact):
    return Edge(from_node=from_node, to_node=to_node, artifact=artifact)


def _response(nodes, artifacts, edges, issues=None):
    return GraphBuilderResponse(
        nodes=nodes,
        artifacts=artifacts,
        edges=edges,
        issues=issues or [],
    )


def _make_node(id, inputs=None, outputs=None):
    return Node(
        id=id,
        name=id.title(),
        node_type=NodeType.DOMAIN,
        system_prompt=f"You are {id}.",
        input_artifacts=inputs or [],
        output_artifacts=outputs or [],
    )


def _make_artifact(name, contributors=None, users=None):
    return Artifact(
        name=name,
        description=f"Artifact {name}",
        contributors=contributors or [],
        users=users or [],
    )


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# PARSER TESTS
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class TestGraphBuilderParser:

    def setup_method(self):
        self.parser = GraphBuilderParser()

    def test_parse_simple_graph(self):
        response = _response(
            nodes=[_node_dict("agent_a", outputs=["art_1"]),
                   _node_dict("agent_b", inputs=["art_1"])],
            artifacts=[_artifact_dict("art_1", contributors=["agent_a"], users=["agent_b"])],
            edges=[_edge("agent_a", "agent_b", "art_1")],
        )
        nodes, artifacts, edges, synth_targets = self.parser.parse(response)
        assert "agent_a" in nodes
        assert "agent_b" in nodes
        assert "art_1" in artifacts
        assert len(edges) == 1

    def test_parse_returns_node_models(self):
        response = _response(
            nodes=[_node_dict("agent_a", outputs=["art_1"])],
            artifacts=[_artifact_dict("art_1", contributors=["agent_a"])],
            edges=[],
        )
        nodes, _, _, _ = self.parser.parse(response)
        assert isinstance(nodes["agent_a"], Node)
        assert nodes["agent_a"].status == NodeStatus.PENDING

    def test_parse_returns_artifact_models(self):
        response = _response(
            nodes=[_node_dict("agent_a", outputs=["art_1"])],
            artifacts=[_artifact_dict("art_1", contributors=["agent_a"])],
            edges=[],
        )
        _, artifacts, _, _ = self.parser.parse(response)
        assert isinstance(artifacts["art_1"], Artifact)

    def test_extracts_synthesizer_targets_from_issues(self):
        response = _response(
            nodes=[_node_dict("agent_a", outputs=["art_1"])],
            artifacts=[_artifact_dict("art_1", contributors=["agent_a"])],
            edges=[],
            issues=["SYNTHESIZER_NEEDED: art_1", "Some other note"],
        )
        _, _, _, synth_targets = self.parser.parse(response)
        assert "art_1" in synth_targets
        assert len(synth_targets) == 1

    def test_ignores_non_synthesizer_issues(self):
        response = _response(
            nodes=[_node_dict("agent_a", outputs=["art_1"])],
            artifacts=[_artifact_dict("art_1", contributors=["agent_a"])],
            edges=[],
            issues=["Fixed dangling artifact", "Removed cycle"],
        )
        _, _, _, synth_targets = self.parser.parse(response)
        assert synth_targets == []

    def test_multiple_synthesizer_targets(self):
        response = _response(
            nodes=[],
            artifacts=[],
            edges=[],
            issues=[
                "SYNTHESIZER_NEEDED: artifact_x",
                "SYNTHESIZER_NEEDED: artifact_y",
            ],
        )
        _, _, _, synth_targets = self.parser.parse(response)
        assert "artifact_x" in synth_targets
        assert "artifact_y" in synth_targets

    def test_edges_preserved(self):
        edges = [
            _edge("a", "b", "art_1"),
            _edge("b", "c", "art_2"),
        ]
        response = _response(
            nodes=[_node_dict("a"), _node_dict("b"), _node_dict("c")],
            artifacts=[
                _artifact_dict("art_1", contributors=["a"], users=["b"]),
                _artifact_dict("art_2", contributors=["b"], users=["c"]),
            ],
            edges=edges,
        )
        _, _, parsed_edges, _ = self.parser.parse(response)
        assert len(parsed_edges) == 2


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# VALIDATOR TESTS
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class TestGraphBuilderValidator:

    def setup_method(self):
        self.validator = GraphBuilderValidator()

    def _simple_setup(self):
        nodes = {
            "agent_a": _make_node("agent_a", outputs=["art_1"]),
            "agent_b": _make_node("agent_b", inputs=["art_1"]),
        }
        artifacts = {
            "art_1": _make_artifact("art_1", contributors=["agent_a"], users=["agent_b"]),
        }
        edges = [_edge("agent_a", "agent_b", "art_1")]
        return nodes, artifacts, edges

    def test_valid_dag_returns_digraph(self):
        nodes, artifacts, edges = self._simple_setup()
        G = self.validator.validate(nodes, artifacts, edges)
        assert isinstance(G, nx.DiGraph)

    def test_valid_dag_has_correct_nodes(self):
        nodes, artifacts, edges = self._simple_setup()
        G = self.validator.validate(nodes, artifacts, edges)
        assert "agent_a" in G.nodes
        assert "agent_b" in G.nodes

    def test_valid_dag_has_correct_edges(self):
        nodes, artifacts, edges = self._simple_setup()
        G = self.validator.validate(nodes, artifacts, edges)
        assert G.has_edge("agent_a", "agent_b")

    def test_unknown_from_node_raises(self):
        nodes = {"agent_a": _make_node("agent_a")}
        artifacts = {}
        edges = [_edge("ghost_node", "agent_a", "art_1")]
        with pytest.raises(GraphValidationError, match="ghost_node"):
            self.validator.validate(nodes, artifacts, edges)

    def test_unknown_to_node_raises(self):
        nodes = {"agent_a": _make_node("agent_a")}
        artifacts = {}
        edges = [_edge("agent_a", "ghost_node", "art_1")]
        with pytest.raises(GraphValidationError, match="ghost_node"):
            self.validator.validate(nodes, artifacts, edges)

    def test_cycle_raises(self):
        nodes = {
            "agent_a": _make_node("agent_a"),
            "agent_b": _make_node("agent_b"),
        }
        artifacts = {}
        edges = [
            _edge("agent_a", "agent_b", "art_1"),
            _edge("agent_b", "agent_a", "art_2"),  # cycle
        ]
        with pytest.raises(GraphValidationError, match="cycle"):
            self.validator.validate(nodes, artifacts, edges)

    def test_isolated_node_raises(self):
        nodes = {
            "agent_a": _make_node("agent_a"),
            "agent_b": _make_node("agent_b"),  # isolated
        }
        artifacts = {}
        edges = []  # no edges â€” agent_b is isolated
        with pytest.raises(GraphValidationError, match="disconnected"):
            self.validator.validate(nodes, artifacts, edges)

    def test_linear_chain_is_valid(self):
        nodes = {
            "a": _make_node("a"),
            "b": _make_node("b"),
            "c": _make_node("c"),
        }
        artifacts = {}
        edges = [_edge("a", "b", "art_1"), _edge("b", "c", "art_2")]
        G = self.validator.validate(nodes, artifacts, edges)
        assert nx.is_directed_acyclic_graph(G)

    def test_parallel_nodes_valid(self):
        nodes = {
            "root":  _make_node("root"),
            "left":  _make_node("left"),
            "right": _make_node("right"),
            "sink":  _make_node("sink"),
        }
        artifacts = {}
        edges = [
            _edge("root",  "left",  "art_1"),
            _edge("root",  "right", "art_2"),
            _edge("left",  "sink",  "art_3"),
            _edge("right", "sink",  "art_4"),
        ]
        G = self.validator.validate(nodes, artifacts, edges)
        assert nx.is_directed_acyclic_graph(G)


================================================
FILE: tests/initiator/__init__.py
================================================
[Empty file]


================================================
FILE: tests/initiator/test_initiator.py
================================================
"""
tests/initiator/test_initiator.py

Unit tests for ArtifactStore and Executor.
AgentRunner LLM calls are mocked.
"""

import asyncio
import pytest
from unittest.mock import AsyncMock, MagicMock, patch

from coven.initiator.artifact_store import ArtifactStore
from coven.initiator.executor import Executor
from coven.models import (
    Artifact, Node, NodeType, NodeStatus,
    DAG, DAGStatus,
)


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# HELPERS
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

def _make_artifact(name, body=None):
    return Artifact(
        name=name,
        description=f"Artifact {name}",
        contributors=[],
        users=[],
        body=body or {},
    )


def _make_node(id, inputs=None, outputs=None, node_type=NodeType.DOMAIN):
    return Node(
        id=id,
        name=id.title(),
        node_type=node_type,
        system_prompt=f"You are {id}.",
        input_artifacts=inputs or [],
        output_artifacts=outputs or [],
    )


def _make_dag(nodes=None, artifacts=None, levels=None):
    return DAG(
        id="test_run",
        task="test task",
        nodes=nodes or {},
        artifacts=artifacts or {},
        levels=levels or [],
    )


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# ARTIFACT STORE TESTS
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class TestArtifactStore:

    def _store_with(self, *artifact_names):
        artifacts = {n: _make_artifact(n) for n in artifact_names}
        return ArtifactStore(artifacts)

    @pytest.mark.asyncio
    async def test_get_existing_artifact(self):
        store = self._store_with("art_1")
        art = await store.get("art_1")
        assert art.name == "art_1"

    @pytest.mark.asyncio
    async def test_get_missing_artifact_raises(self):
        store = self._store_with("art_1")
        with pytest.raises(KeyError, match="art_2"):
            await store.get("art_2")

    @pytest.mark.asyncio
    async def test_put_updates_body(self):
        store = self._store_with("art_1")
        await store.put("art_1", {"result": 42})
        art = await store.get("art_1")
        assert art.body == {"result": 42}

    @pytest.mark.asyncio
    async def test_put_unknown_artifact_raises(self):
        store = self._store_with("art_1")
        with pytest.raises(KeyError, match="ghost_art"):
            await store.put("ghost_art", {"x": 1})

    @pytest.mark.asyncio
    async def test_get_many_returns_ordered(self):
        store = self._store_with("art_1", "art_2", "art_3")
        await store.put("art_1", {"v": 1})
        await store.put("art_2", {"v": 2})
        await store.put("art_3", {"v": 3})
        result = await store.get_many(["art_3", "art_1"])
        assert result[0].name == "art_3"
        assert result[1].name == "art_1"

    @pytest.mark.asyncio
    async def test_all_returns_snapshot(self):
        store = self._store_with("art_1", "art_2")
        snapshot = await store.all()
        assert "art_1" in snapshot
        assert "art_2" in snapshot

    def test_is_ready_false_when_empty_body(self):
        store = self._store_with("art_1")
        assert not store.is_ready("art_1")

    @pytest.mark.asyncio
    async def test_is_ready_true_after_put(self):
        store = self._store_with("art_1")
        await store.put("art_1", {"data": "filled"})
        assert store.is_ready("art_1")

    def test_is_ready_false_for_unknown(self):
        store = self._store_with("art_1")
        assert not store.is_ready("nonexistent")

    def test_all_inputs_ready_true(self):
        artifacts = {
            "a": _make_artifact("a", body={"x": 1}),
            "b": _make_artifact("b", body={"y": 2}),
        }
        store = ArtifactStore(artifacts)
        assert store.all_inputs_ready(["a", "b"])

    def test_all_inputs_ready_false_when_one_empty(self):
        artifacts = {
            "a": _make_artifact("a", body={"x": 1}),
            "b": _make_artifact("b"),  # empty body
        }
        store = ArtifactStore(artifacts)
        assert not store.all_inputs_ready(["a", "b"])

    @pytest.mark.asyncio
    async def test_concurrent_puts_are_safe(self):
        store = self._store_with("art_1", "art_2", "art_3")

        async def write(name, val):
            await store.put(name, {"v": val})

        await asyncio.gather(
            write("art_1", 1),
            write("art_2", 2),
            write("art_3", 3),
        )
        assert (await store.get("art_1")).body == {"v": 1}
        assert (await store.get("art_2")).body == {"v": 2}
        assert (await store.get("art_3")).body == {"v": 3}


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# EXECUTOR TESTS
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class TestExecutor:

    def _make_completed_node(self, id, outputs=None):
        """Returns a node that the mock runner will report as COMPLETED."""
        return _make_node(id, outputs=outputs or [f"{id}_out"])

    def _make_dag_with_levels(self, levels):
        """Build a minimal DAG with the given levels."""
        all_node_ids = [n for level in levels for n in level]
        nodes = {id: self._make_completed_node(id) for id in all_node_ids}
        artifacts = {
            f"{id}_out": _make_artifact(f"{id}_out")
            for id in all_node_ids
        }
        return _make_dag(nodes=nodes, artifacts=artifacts, levels=levels)

    @pytest.mark.asyncio
    async def test_executor_completes_single_level(self):
        dag = self._make_dag_with_levels([["agent_a"]])

        async def mock_run(node, store, nodes):
            return node.model_copy(update={"status": NodeStatus.COMPLETED})

        executor = Executor(model="gpt-4o")
        with patch.object(executor._runner, "run", side_effect=mock_run):
            result = await executor.execute(dag)

        assert result.status == DAGStatus.COMPLETED
        assert result.nodes["agent_a"].status == NodeStatus.COMPLETED

    @pytest.mark.asyncio
    async def test_executor_completes_multiple_levels(self):
        dag = self._make_dag_with_levels([["agent_a"], ["agent_b"]])

        async def mock_run(node, store, nodes):
            return node.model_copy(update={"status": NodeStatus.COMPLETED})

        executor = Executor(model="gpt-4o")
        with patch.object(executor._runner, "run", side_effect=mock_run):
            result = await executor.execute(dag)

        assert result.status == DAGStatus.COMPLETED

    @pytest.mark.asyncio
    async def test_executor_fails_on_failed_node(self):
        dag = self._make_dag_with_levels([["agent_a"]])

        async def mock_run(node, store, nodes):
            return node.model_copy(update={"status": NodeStatus.FAILED})

        executor = Executor(model="gpt-4o")
        with patch.object(executor._runner, "run", side_effect=mock_run):
            result = await executor.execute(dag)

        assert result.status == DAGStatus.FAILED

    @pytest.mark.asyncio
    async def test_executor_halts_after_failed_level(self):
        """If level 0 fails, level 1 should never execute."""
        dag = self._make_dag_with_levels([["agent_a"], ["agent_b"]])

        call_order = []

        async def mock_run(node, store, nodes):
            call_order.append(node.id)
            if node.id == "agent_a":
                return node.model_copy(update={"status": NodeStatus.FAILED})
            return node.model_copy(update={"status": NodeStatus.COMPLETED})

        executor = Executor(model="gpt-4o")
        with patch.object(executor._runner, "run", side_effect=mock_run):
            result = await executor.execute(dag)

        assert result.status == DAGStatus.FAILED
        assert "agent_b" not in call_order

    @pytest.mark.asyncio
    async def test_executor_parallel_level_runs_concurrently(self):
        """Nodes in the same level should all be invoked."""
        dag = self._make_dag_with_levels([["agent_a", "agent_b", "agent_c"]])
        called = []

        async def mock_run(node, store, nodes):
            called.append(node.id)
            return node.model_copy(update={"status": NodeStatus.COMPLETED})

        executor = Executor(model="gpt-4o")
        with patch.object(executor._runner, "run", side_effect=mock_run):
            result = await executor.execute(dag)

        assert sorted(called) == ["agent_a", "agent_b", "agent_c"]
        assert result.status == DAGStatus.COMPLETED

    @pytest.mark.asyncio
    async def test_executor_writes_artifacts_to_dag(self):
        dag = self._make_dag_with_levels([["agent_a"]])

        async def mock_run(node, store, nodes):
            await store.put("agent_a_out", {"computed": True})
            return node.model_copy(update={"status": NodeStatus.COMPLETED})

        executor = Executor(model="gpt-4o")
        with patch.object(executor._runner, "run", side_effect=mock_run):
            result = await executor.execute(dag)

        assert result.artifacts["agent_a_out"].body == {"computed": True}

    @pytest.mark.asyncio
    async def test_executor_sets_running_status(self):
        dag = self._make_dag_with_levels([["agent_a"]])

        statuses_seen = []

        async def mock_run(node, store, nodes):
            statuses_seen.append(dag.status)
            return node.model_copy(update={"status": NodeStatus.COMPLETED})

        executor = Executor(model="gpt-4o")
        with patch.object(executor._runner, "run", side_effect=mock_run):
            result = await executor.execute(dag)

        # DAG should have been set to RUNNING before any node ran
        assert DAGStatus.RUNNING in statuses_seen or result.status == DAGStatus.COMPLETED


================================================
FILE: tests/mcp_builder/__init__.py
================================================
[Empty file]


================================================
FILE: tests/mcp_builder/test_mpc_builder.py
================================================
"""
tests/mcp_builder/test_mcp_builder.py

Unit tests for MCPNodeBuilder.
ToolStorePy.build() is mocked â€” no network calls or real builds.
"""

import json
import pytest
from pathlib import Path
from unittest.mock import MagicMock, patch

from coven.mcp_builder.builder import MCPNodeBuilder
from coven.models import Node, NodeType, NodeStatus
from coven.models.node import ToolQuery


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# HELPERS
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

def _make_node(id, query_tool=None):
    return Node(
        id=id,
        name=id.title(),
        node_type=NodeType.DOMAIN,
        system_prompt=f"You are {id}.",
        query_tool=query_tool or [],
        output_artifacts=[f"{id}_out"],
    )


def _make_builder(tmp_path):
    return MCPNodeBuilder(
        base_workspace=tmp_path,
        index="core-tools",
        index_url=None,
        install_requirements=False,
        verbose=False,
    )


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# TESTS
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class TestMCPNodeBuilder:

    def test_build_for_node_returns_none_when_no_query_tool(self, tmp_path):
        builder = _make_builder(tmp_path)
        node = _make_node("agent_a", query_tool=[])
        result = builder.build_for_node(node)
        assert result is None

    def test_build_for_node_returns_path_when_toolstorepy_succeeds(self, tmp_path):
        builder = _make_builder(tmp_path)
        node = _make_node("agent_a", query_tool=[
            ToolQuery(tool_description="evaluate a mathematical expression"),
        ])

        mock_output = tmp_path / "mcp" / "agent_a" / "mcp_unified_server.py"
        mock_output.parent.mkdir(parents=True, exist_ok=True)
        mock_output.touch()

        mock_toolstore = MagicMock()
        mock_toolstore.build.return_value = str(mock_output)

        with patch("coven.mcp_builder.builder.ToolStorePy", return_value=mock_toolstore):
            result = builder.build_for_node(node)

        assert result == mock_output

    def test_build_for_node_raises_on_toolstorepy_failure(self, tmp_path):
        builder = _make_builder(tmp_path)
        node = _make_node("agent_a", query_tool=[
            ToolQuery(tool_description="some tool"),
        ])

        mock_toolstore = MagicMock()
        mock_toolstore.build.side_effect = RuntimeError("Build failed")

        with patch("coven.mcp_builder.builder.ToolStorePy", return_value=mock_toolstore):
            with pytest.raises(RuntimeError, match="ToolStorePy failed"):
                builder.build_for_node(node)

    def test_write_queries_creates_file(self, tmp_path):
        builder = _make_builder(tmp_path)
        node = _make_node("agent_a", query_tool=[
            ToolQuery(tool_description="evaluate a mathematical expression"),
            ToolQuery(tool_description="convert units of measurement"),
        ])
        workspace = tmp_path / "test_workspace"
        workspace.mkdir()

        queries_path = builder._write_queries(node, workspace)

        assert queries_path.exists()
        assert queries_path.name == "queries.json"

    def test_write_queries_correct_format(self, tmp_path):
        builder = _make_builder(tmp_path)
        node = _make_node("agent_a", query_tool=[
            ToolQuery(tool_description="evaluate a mathematical expression"),
            ToolQuery(tool_description="convert units of measurement"),
        ])
        workspace = tmp_path / "ws"
        workspace.mkdir()

        queries_path = builder._write_queries(node, workspace)

        with open(queries_path) as f:
            data = json.load(f)

        assert len(data) == 2
        assert data[0] == {"tool_description": "evaluate a mathematical expression"}
        assert data[1] == {"tool_description": "convert units of measurement"}

    def test_each_node_gets_isolated_workspace(self, tmp_path):
        builder = _make_builder(tmp_path)

        node_a = _make_node("agent_a", query_tool=[ToolQuery(tool_description="tool a")])
        node_b = _make_node("agent_b", query_tool=[ToolQuery(tool_description="tool b")])

        mock_toolstore = MagicMock()

        def fake_build(queries, index, index_url, force_refresh):
            # Return a path based on the workspace being built
            ws = Path(queries).parent
            out = ws / "mcp_unified_server.py"
            out.touch()
            return str(out)

        mock_toolstore.build.side_effect = fake_build

        with patch("coven.mcp_builder.builder.ToolStorePy", return_value=mock_toolstore):
            path_a = builder.build_for_node(node_a)
            path_b = builder.build_for_node(node_b)

        assert path_a != path_b
        assert "agent_a" in str(path_a)
        assert "agent_b" in str(path_b)

    def test_toolstorepy_called_with_correct_index(self, tmp_path):
        builder = MCPNodeBuilder(
            base_workspace=tmp_path,
            index="core-tools",
            index_url=None,
        )
        node = _make_node("agent_a", query_tool=[ToolQuery(tool_description="some tool")])

        mock_toolstore = MagicMock()
        mock_toolstore.build.return_value = str(tmp_path / "mcp_unified_server.py")

        with patch("coven.mcp_builder.builder.ToolStorePy", return_value=mock_toolstore) as MockTS:
            builder.build_for_node(node)

        call_kwargs = mock_toolstore.build.call_args
        assert call_kwargs.kwargs.get("index") == "core-tools" or \
               (call_kwargs.args and "core-tools" in str(call_kwargs))

    def test_toolstorepy_called_with_index_url_when_provided(self, tmp_path):
        builder = MCPNodeBuilder(
            base_workspace=tmp_path,
            index=None,
            index_url="https://example.com/my-index.zip",
        )
        node = _make_node("agent_a", query_tool=[ToolQuery(tool_description="some tool")])

        mock_toolstore = MagicMock()
        mock_toolstore.build.return_value = str(tmp_path / "mcp_unified_server.py")

        with patch("coven.mcp_builder.builder.ToolStorePy", return_value=mock_toolstore):
            builder.build_for_node(node)

        call_kwargs = mock_toolstore.build.call_args
        assert call_kwargs.kwargs.get("index_url") == "https://example.com/my-index.zip" or \
               "https://example.com/my-index.zip" in str(call_kwargs)


================================================
FILE: tests/models/__init__.py
================================================
[Empty file]


================================================
FILE: tests/models/test_models.py
================================================
"""
tests/models/test_models.py

Unit tests for Coven/models/ â€” Artifact, Node, DAG, DAGStatus, NodeType, NodeStatus.
No LLM calls. Pure Pydantic validation.
"""

import pytest
from pydantic import ValidationError

from coven.models import Artifact, Node, NodeType, NodeStatus, DAG, DAGStatus
from coven.models.node import ToolQuery


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# ARTIFACT
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class TestArtifact:

    def test_basic_creation(self):
        a = Artifact(
            name="market_report",
            description="Market analysis output",
            contributors=["agent_a"],
            users=["agent_b"],
        )
        assert a.name == "market_report"
        assert a.body == {}

    def test_defaults(self):
        a = Artifact(name="x", description="y")
        assert a.contributors == []
        assert a.users == []
        assert a.body == {}

    def test_body_is_free_form(self):
        a = Artifact(
            name="x",
            description="y",
            body={"key": "value", "nested": {"a": 1}},
        )
        assert a.body["nested"]["a"] == 1

    def test_missing_name_raises(self):
        with pytest.raises(ValidationError):
            Artifact(description="no name")

    def test_missing_description_raises(self):
        with pytest.raises(ValidationError):
            Artifact(name="no_desc")

    def test_multiple_contributors_and_users(self):
        a = Artifact(
            name="shared",
            description="shared artifact",
            contributors=["a", "b", "c"],
            users=["d", "e"],
        )
        assert len(a.contributors) == 3
        assert len(a.users) == 2


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# NODE
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class TestNode:

    def _make_node(self, **kwargs):
        defaults = dict(
            id="test_node",
            name="Test Node",
            node_type=NodeType.DOMAIN,
            system_prompt="You are a test agent.",
        )
        defaults.update(kwargs)
        return Node(**defaults)

    def test_basic_creation(self):
        n = self._make_node()
        assert n.id == "test_node"
        assert n.status == NodeStatus.PENDING
        assert n.query_tool == []
        assert n.mcp_server_path is None

    def test_node_type_enum(self):
        for nt in NodeType:
            n = self._make_node(node_type=nt)
            assert n.node_type == nt

    def test_status_enum(self):
        for ns in NodeStatus:
            n = self._make_node(status=ns)
            assert n.status == ns

    def test_query_tool_typed(self):
        n = self._make_node(
            query_tool=[
                {"tool_description": "evaluate a mathematical expression"},
                {"tool_description": "convert units of measurement"},
            ]
        )
        assert len(n.query_tool) == 2
        assert isinstance(n.query_tool[0], ToolQuery)
        assert n.query_tool[0].tool_description == "evaluate a mathematical expression"

    def test_query_tool_empty_by_default(self):
        n = self._make_node()
        assert n.query_tool == []

    def test_mcp_server_path_can_be_set(self):
        n = self._make_node(mcp_server_path="/some/path/mcp_server.py")
        assert n.mcp_server_path == "/some/path/mcp_server.py"

    def test_model_copy_immutability(self):
        n = self._make_node()
        n2 = n.model_copy(update={"status": NodeStatus.COMPLETED})
        assert n.status == NodeStatus.PENDING
        assert n2.status == NodeStatus.COMPLETED

    def test_missing_required_fields_raises(self):
        with pytest.raises(ValidationError):
            Node(name="x", node_type=NodeType.DOMAIN, system_prompt="y")  # missing id

    def test_contributor_system_prompts(self):
        n = self._make_node(
            node_type=NodeType.SYNTHESIZER,
            contributor_system_prompts=["prompt_a", "prompt_b"],
        )
        assert len(n.contributor_system_prompts) == 2


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# DAG
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class TestDAG:

    def _make_node(self, node_id: str, inputs=None, outputs=None):
        return Node(
            id=node_id,
            name=node_id.replace("_", " ").title(),
            node_type=NodeType.DOMAIN,
            system_prompt=f"You are {node_id}.",
            input_artifacts=inputs or [],
            output_artifacts=outputs or [],
        )

    def _make_artifact(self, name, contributors=None, users=None):
        return Artifact(
            name=name,
            description=f"Artifact {name}",
            contributors=contributors or [],
            users=users or [],
        )

    def test_empty_dag(self):
        dag = DAG(id="run_001", task="test task")
        assert dag.nodes == {}
        assert dag.artifacts == {}
        assert dag.levels == []
        assert dag.status == DAGStatus.PLANNED

    def test_dag_with_valid_nodes_and_artifacts(self):
        node_a = self._make_node("agent_a", outputs=["artifact_1"])
        node_b = self._make_node("agent_b", inputs=["artifact_1"])
        art    = self._make_artifact("artifact_1", contributors=["agent_a"], users=["agent_b"])

        dag = DAG(
            id="run_001",
            task="test",
            nodes={"agent_a": node_a, "agent_b": node_b},
            artifacts={"artifact_1": art},
        )
        assert len(dag.nodes) == 2
        assert len(dag.artifacts) == 1

    def test_dag_validator_rejects_unknown_contributor(self):
        art = self._make_artifact("art_1", contributors=["ghost_node"], users=[])
        with pytest.raises(ValueError, match="ghost_node"):
            DAG(id="run_001", task="t", artifacts={"art_1": art})

    def test_dag_validator_rejects_unknown_user(self):
        node_a = self._make_node("agent_a")
        art = self._make_artifact("art_1", contributors=["agent_a"], users=["ghost_node"])
        with pytest.raises(ValueError, match="ghost_node"):
            DAG(
                id="run_001",
                task="t",
                nodes={"agent_a": node_a},
                artifacts={"art_1": art},
            )

    def test_get_node(self):
        node_a = self._make_node("agent_a")
        dag = DAG(id="r", task="t", nodes={"agent_a": node_a})
        assert dag.get_node("agent_a") is node_a

    def test_get_node_missing_raises(self):
        dag = DAG(id="r", task="t")
        with pytest.raises(KeyError):
            dag.get_node("does_not_exist")

    def test_get_artifact(self):
        node_a = self._make_node("agent_a")
        art = self._make_artifact("art_1", contributors=["agent_a"])
        dag = DAG(
            id="r", task="t",
            nodes={"agent_a": node_a},
            artifacts={"art_1": art},
        )
        assert dag.get_artifact("art_1") is art

    def test_is_complete_false_when_pending(self):
        node_a = self._make_node("agent_a")
        dag = DAG(id="r", task="t", nodes={"agent_a": node_a})
        assert not dag.is_complete()

    def test_is_complete_true_when_all_completed(self):
        node_a = self._make_node("agent_a")
        node_a = node_a.model_copy(update={"status": NodeStatus.COMPLETED})
        dag = DAG(id="r", task="t", nodes={"agent_a": node_a})
        assert dag.is_complete()

    def test_dag_status_transitions(self):
        dag = DAG(id="r", task="t")
        assert dag.status == DAGStatus.PLANNED
        dag2 = dag.model_copy(update={"status": DAGStatus.RUNNING})
        assert dag2.status == DAGStatus.RUNNING
        dag3 = dag2.model_copy(update={"status": DAGStatus.COMPLETED})
        assert dag3.status == DAGStatus.COMPLETED


================================================
FILE: tests/sorter/__init__.py
================================================
[Empty file]


================================================
FILE: tests/sorter/test_sorter.py
================================================
Error reading file with 'cp1252': 'charmap' codec can't decode byte 0x90 in position 1599: character maps to <undefined>


================================================
FILE: tests/synthesizer/__init__.py
================================================
[Empty file]


================================================
FILE: tests/synthesizer/test_synthesizer.py
================================================
"""
tests/synthesizer/test_synthesizer.py

Unit tests for SynthesizerInjector and SynthesizerParser.
No LLM calls.
"""

import pytest
import networkx as nx

from coven.synthesizer.injector import SynthesizerInjector
from coven.synthesizer.parser import SynthesizerParser
from coven.synthesizer.agent import SynthesizerResponse
from coven.models import Node, NodeType, NodeStatus, Artifact


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# HELPERS
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

def _make_node(id, outputs=None, inputs=None):
    return Node(
        id=id,
        name=id.title(),
        node_type=NodeType.DOMAIN,
        system_prompt=f"System prompt for {id}.",
        input_artifacts=inputs or [],
        output_artifacts=outputs or [],
    )


def _make_artifact(name, contributors=None, users=None):
    return Artifact(
        name=name,
        description=f"Artifact {name}",
        contributors=contributors or [],
        users=users or [],
    )


def _build_graph(nodes, edges):
    G = nx.DiGraph()
    G.add_nodes_from(nodes)
    G.add_edges_from(edges)
    return G


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# SYNTHESIZER INJECTOR TESTS
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class TestSynthesizerInjector:

    def setup_method(self):
        self.injector = SynthesizerInjector()

    def _make_two_contributor_setup(self):
        nodes = {
            "agent_a": _make_node("agent_a", outputs=["shared_art"]),
            "agent_b": _make_node("agent_b", outputs=["shared_art"]),
            "agent_c": _make_node("agent_c", inputs=["shared_art"]),
        }
        artifacts = {
            "shared_art": _make_artifact(
                "shared_art",
                contributors=["agent_a", "agent_b"],
                users=["agent_c"],
            )
        }
        G = _build_graph(
            nodes=list(nodes.keys()),
            edges=[("agent_a", "agent_c"), ("agent_b", "agent_c")],
        )
        return nodes, artifacts, G

    def test_inject_creates_synthesizer_node(self):
        nodes, artifacts, G = self._make_two_contributor_setup()
        nodes, artifacts, G = self.injector.inject(
            nodes, artifacts, ["shared_art"], G
        )
        assert "synthesizer_shared_art" in nodes

    def test_synthesizer_node_type(self):
        nodes, artifacts, G = self._make_two_contributor_setup()
        nodes, artifacts, G = self.injector.inject(
            nodes, artifacts, ["shared_art"], G
        )
        synth = nodes["synthesizer_shared_art"]
        assert synth.node_type == NodeType.SYNTHESIZER

    def test_synthesizer_has_contributor_system_prompts(self):
        nodes, artifacts, G = self._make_two_contributor_setup()
        nodes, artifacts, G = self.injector.inject(
            nodes, artifacts, ["shared_art"], G
        )
        synth = nodes["synthesizer_shared_art"]
        assert len(synth.contributor_system_prompts) == 2
        assert "System prompt for agent_a." in synth.contributor_system_prompts
        assert "System prompt for agent_b." in synth.contributor_system_prompts

    def test_partial_artifacts_created(self):
        nodes, artifacts, G = self._make_two_contributor_setup()
        nodes, artifacts, G = self.injector.inject(
            nodes, artifacts, ["shared_art"], G
        )
        assert "shared_art__partial__agent_a" in artifacts
        assert "shared_art__partial__agent_b" in artifacts

    def test_partial_artifacts_have_correct_contributor(self):
        nodes, artifacts, G = self._make_two_contributor_setup()
        nodes, artifacts, G = self.injector.inject(
            nodes, artifacts, ["shared_art"], G
        )
        partial_a = artifacts["shared_art__partial__agent_a"]
        assert partial_a.contributors == ["agent_a"]
        assert partial_a.users == ["synthesizer_shared_art"]

    def test_original_artifact_contributor_updated_to_synthesizer(self):
        nodes, artifacts, G = self._make_two_contributor_setup()
        nodes, artifacts, G = self.injector.inject(
            nodes, artifacts, ["shared_art"], G
        )
        assert artifacts["shared_art"].contributors == ["synthesizer_shared_art"]

    def test_contributor_outputs_updated_to_partial(self):
        nodes, artifacts, G = self._make_two_contributor_setup()
        nodes, artifacts, G = self.injector.inject(
            nodes, artifacts, ["shared_art"], G
        )
        # agent_a's output should now be the partial, not the original
        assert "shared_art__partial__agent_a" in nodes["agent_a"].output_artifacts
        assert "shared_art" not in nodes["agent_a"].output_artifacts

    def test_synthesizer_in_graph(self):
        nodes, artifacts, G = self._make_two_contributor_setup()
        nodes, artifacts, G = self.injector.inject(
            nodes, artifacts, ["shared_art"], G
        )
        assert "synthesizer_shared_art" in G.nodes

    def test_synthesizer_input_output_artifacts(self):
        nodes, artifacts, G = self._make_two_contributor_setup()
        nodes, artifacts, G = self.injector.inject(
            nodes, artifacts, ["shared_art"], G
        )
        synth = nodes["synthesizer_shared_art"]
        assert "shared_art__partial__agent_a" in synth.input_artifacts
        assert "shared_art__partial__agent_b" in synth.input_artifacts
        assert synth.output_artifacts == ["shared_art"]

    def test_skip_single_contributor_artifact(self):
        nodes = {
            "agent_a": _make_node("agent_a", outputs=["art_1"]),
            "agent_b": _make_node("agent_b", inputs=["art_1"]),
        }
        artifacts = {
            "art_1": _make_artifact("art_1", contributors=["agent_a"], users=["agent_b"]),
        }
        G = _build_graph(["agent_a", "agent_b"], [("agent_a", "agent_b")])
        original_node_count = len(nodes)

        nodes, artifacts, G = self.injector.inject(nodes, artifacts, ["art_1"], G)

        # No synthesizer should be injected â€” only one contributor
        assert len(nodes) == original_node_count

    def test_skip_unknown_artifact_name(self):
        nodes = {"agent_a": _make_node("agent_a", outputs=["art_1"])}
        artifacts = {"art_1": _make_artifact("art_1", contributors=["agent_a"])}
        G = _build_graph(["agent_a"], [])
        original_node_count = len(nodes)

        nodes, artifacts, G = self.injector.inject(
            nodes, artifacts, ["nonexistent_artifact"], G
        )
        assert len(nodes) == original_node_count


# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€
# SYNTHESIZER PARSER TESTS
# â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€â”€

class TestSynthesizerParser:

    def setup_method(self):
        self.parser = SynthesizerParser()

    def _make_target_artifact(self):
        return Artifact(
            name="merged_artifact",
            description="Merged output from multiple contributors",
            contributors=["synthesizer_merged_artifact"],
            users=["compiler"],
        )

    def test_parse_applies_body(self):
        response = SynthesizerResponse(
            body={"key": "value", "score": 42},
            qc_notes=[],
        )
        target = self._make_target_artifact()
        result = self.parser.parse(response, target)
        assert result.body["key"] == "value"
        assert result.body["score"] == 42

    def test_parse_injects_qc_notes(self):
        response = SynthesizerResponse(
            body={"key": "value"},
            qc_notes=["Resolved conflict between A and B", "Filled gap in section 2"],
        )
        target = self._make_target_artifact()
        result = self.parser.parse(response, target)
        assert "__qc_notes__" in result.body
        assert len(result.body["__qc_notes__"]) == 2

    def test_parse_preserves_artifact_metadata(self):
        response = SynthesizerResponse(body={"x": 1}, qc_notes=[])
        target = self._make_target_artifact()
        result = self.parser.parse(response, target)
        assert result.name == "merged_artifact"
        assert result.description == "Merged output from multiple contributors"

    def test_parse_empty_body(self):
        response = SynthesizerResponse(body={}, qc_notes=["Note 1"])
        target = self._make_target_artifact()
        result = self.parser.parse(response, target)
        assert result.body["__qc_notes__"] == ["Note 1"]

    def test_parse_does_not_mutate_original(self):
        response = SynthesizerResponse(body={"x": 1}, qc_notes=[])
        target = self._make_target_artifact()
        original_body = dict(target.body)
        self.parser.parse(response, target)
        assert target.body == original_body  # original unchanged

