scripts.seed_component_dag

Reference › scripts › seed_component_dag

scripts/seed_component_dag.py seeds the orchestrata-ultimate component DAG into the operational knowledge graph; the DAG is the table COMPONENT_DAG in the module, one row per component with its import name, whether it is required and what it depends on (see the component DAG page). A human runs it after the table changes. It writes through ArchitectureWriter from orchestrata_ultimate.knowledge_graph.architecture_writer and resolves its target with orchestrata_ultimate.core.graph_factory; _load_dotenv first loads the RUL-019 environment contract through orchestrata_ultimate.core.env_bootstrap.

Options: --project (default orchestrata-ultimate) and --dry-run, which prints the resolved target URI and database, the project and every table row without connecting to the graph. A real run prints the target first, so the coordinator sees which graph it is about to write, then calls seed_component_dag, which is idempotent through MERGE, and closes the writer.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Constants

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/seed_component_dag.py

"""Seed the orchestrata-ultimate component DAG into the operational knowledge graph.

Thin CLI over ``ArchitectureWriter`` -- writes the ten-row component DAG defined
by ``plans/component-fixes-2026-10-19.md`` §4.2 (the ratified ontology table)
into the graph via ``upsert_component`` and ``link_depends_on`` ONLY.

This is operational seed tooling. Per the two-lane rule and AGENTS.md @traces
conventions, this module is DELIBERATELY UNTRACED: it implements no business
requirement from a root spec; it is infrastructure to populate the ontology.

Idempotent -- every write is a MERGE through ArchitectureWriter.
"""

from __future__ import annotations

import argparse
import sys
from pathlib import Path

# Ensure repo src is on path when run directly (mirrors sync_architecture_to_graph.py).
_repo_src = Path(__file__).resolve().parent.parent / "src"
if str(_repo_src) not in sys.path:
    sys.path.insert(0, str(_repo_src))

from orchestrata_ultimate.core.env_bootstrap import load_env_contract  # noqa: E402
from orchestrata_ultimate.core.graph_factory import (  # noqa: E402
    resolve_kg_database,
    resolve_uri,
)
from orchestrata_ultimate.knowledge_graph.architecture_writer import (  # noqa: E402
    ArchitectureWriter,
)

# ── Component DAG table (RUL-037: 10 rows from plans/component-fixes-2026-10-19.md §4.2) ──
# Columns: name, import_name, responsibility, depends_on (tuple), required (bool).
# Responsibility strings copied VERBATIM from the ruled table.
COMPONENT_DAG: tuple[dict[str, str | tuple[str, ...] | bool], ...] = (
    {
        "name": "contracts",
        "import_name": "orchestrata_contracts",
        "responsibility": "Protocols, DTOs, enums, errors. No behaviour, no I/O.",
        "depends_on": (),
        "required": True,
    },
    {
        "name": "governance",
        "import_name": "orchestrata_governance",
        "responsibility": "Prolog policy engine and the governance half of the factory.",
        "depends_on": ("contracts",),
        "required": True,
    },
    {
        "name": "knowledge_graph",
        "import_name": "orchestrata_kg",
        "responsibility": "Operational graph and semantic graph clients.",
        "depends_on": ("contracts",),
        "required": True,
    },
    {
        "name": "kernel",
        "import_name": "orchestrata_kernel",
        "responsibility": "Message path, CXP/cxpt, postman, pipeline, volk registry.",
        "depends_on": ("contracts", "governance", "knowledge_graph"),
        "required": True,
    },
    {
        "name": "session",
        "import_name": "orchestrata_session",
        "responsibility": "Session lifecycle, goals, handoff, context files.",
        "depends_on": ("contracts", "knowledge_graph", "governance"),
        "required": True,
    },
    {
        "name": "volks",
        "import_name": "orchestrata_volks",
        "responsibility": "Agents, LLM client, providers, drivers, harness, verification.",
        "depends_on": ("contracts", "kernel", "knowledge_graph", "governance"),
        "required": True,
    },
    {
        "name": "gateway",
        "import_name": "orchestrata_gateway",
        "responsibility": "WebSocket JSON-RPC front door, RPC table, MCP transport.",
        "depends_on": (
            "contracts",
            "kernel",
            "session",
            "knowledge_graph",
            "governance",
        ),
        "required": True,
    },
    {
        "name": "tools",
        "import_name": "orchestrata_tools",
        "responsibility": "Volk tools. Redis and embeddings through contracts protocols (ADR-0112), not kernel imports.",
        "depends_on": ("contracts", "knowledge_graph", "governance"),
        "required": True,
    },
    {
        "name": "integration",
        "import_name": "orchestrata",
        "responsibility": "Composition root. Not built. Boot is where ADR-0112's implementations are supplied.",
        "depends_on": (
            "kernel",
            "gateway",
            "volks",
            "session",
            "knowledge_graph",
            "governance",
            "tools",
        ),
        "required": False,
    },
    {
        "name": "plugin",
        "import_name": "",
        "responsibility": "Installs the composed release. Not a realize component.",
        "depends_on": ("gateway", "tools", "integration"),
        "required": False,
    },
)


def seed_component_dag(
    writer: ArchitectureWriter, project: str = "orchestrata-ultimate"
) -> dict[str, int]:
    """Write the component DAG into the graph.

    Phase 1: upsert_component for every row (boundary carries import_name).
    Phase 2: link_depends_on for every dependency (from_id, to_id are {project}:{name}).

    Args:
        writer: An ArchitectureWriter instance connected to the graph.
        project: The project slug (default: orchestrata-ultimate).

    Returns:
        A dict with keys "components" and "depends_on" giving the counts written.
    """
    # Phase 1: upsert all components
    component_ids: dict[str, str] = {}
    for row in COMPONENT_DAG:
        name = row["name"]
        import_name = row["import_name"]
        responsibility = row["responsibility"]
        required = row["required"]
        comp_id = writer.upsert_component(
            project,
            name,
            responsibility=responsibility,
            required=required,
            boundary=import_name,
        )
        component_ids[name] = comp_id

    # Phase 2: link dependencies (STRICTLY after phase 1)
    edge_count = 0
    for row in COMPONENT_DAG:
        from_name = row["name"]
        from_id = component_ids[from_name]
        for dep_name in row["depends_on"]:
            to_id = component_ids[dep_name]
            writer.link_depends_on(from_id, to_id)
            edge_count += 1

    return {
        "components": len(COMPONENT_DAG),
        "depends_on": edge_count,
    }


def _load_dotenv() -> None:
    """Load the RUL-019 two-phase environment contract."""
    load_env_contract()


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Seed the orchestrata-ultimate component DAG into the knowledge graph."
    )
    parser.add_argument(
        "--project",
        default="orchestrata-ultimate",
        help="Project slug (default: orchestrata-ultimate).",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Print the resolved target and table rows without connecting to the graph.",
    )
    return parser


def main(argv: list[str] | None = None) -> int:
    """Seed the component DAG into the graph. Idempotent via MERGE."""
    args = _build_parser().parse_args(argv)
    _load_dotenv()

    if args.dry_run:
        target_uri = resolve_uri()
        target_db = resolve_kg_database()
        print(f"Target URI: {target_uri}")
        print(f"Target database: {target_db}")
        print(f"Project: {args.project}")
        print("Component DAG rows:")
        for row in COMPONENT_DAG:
            deps = ", ".join(row["depends_on"]) if row["depends_on"] else "(none)"
            print(
                f"  - {row['name']}: {row['import_name'] or '(empty)'} | "
                f"required={row['required']} | depends_on={deps}"
            )
        return 0

    # Non-dry-run: print target FIRST (coordinator must see prod vs test)
    target_uri = resolve_uri()
    target_db = resolve_kg_database()
    print(f"Target URI: {target_uri}")
    print(f"Target database: {target_db}")

    writer = ArchitectureWriter()
    try:
        summary = seed_component_dag(writer, project=args.project)
        print(f"✅ Component DAG seeded to knowledge graph: {summary}")
    finally:
        writer.close()

    return 0


if __name__ == "__main__":
    raise SystemExit(main())