scripts.context_pack

Reference › scripts › context_pack

A thin command-line interface to the SPEC-106 context pack (requirements FR-002, FR-003, FR-004 and FR-005), which selects the code and knowledge relevant to a request and packs it for an agent. The selection and packing logic is in the knowledge package; this script only parses arguments and connects to the graph through graph_factory and core.graph. benchmark_context_pack times the same pipeline.

Invocation: main calls run with the arguments built in _build_parser. _driver_from_env creates the graph driver from the environment, and a missing setting raises ConfigurationError, which main prints as an error with exit 1. Otherwise the exit code is that of run.

At a glance

Class diagram

Dependency diagram

API

Classes

ConfigurationError

Bases: RuntimeError · Line 61

Raised when required environment configuration is missing.

Functions

Source

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

#!/usr/bin/env python3
"""Thin CLI for the SPEC-106 context pack (FR-002/003/004/005).

Emits a deterministic context pack for an agent brief — replacing
hand-written context-safety excerpts (REQ-106-041):

    python scripts/context_pack.py <task-type> <target> --budget N [--output FILE]

``<task-type>`` is one of build/refactor/debug/document/test/review.
``<target>`` is free text (a module/class/function name, or a short
description) fed to ``analyze_request`` for entity linking.

Connection is read from ``KG_ROUTING_URI``/``KG_URI`` at call time — no
hardcoded URI/port default. Both unset is a
configuration error (raised, non-zero exit); a Neo4j that is configured but
unreachable degrades to an explicit empty pack with a ``reason`` in its
frontmatter, exit 0 (REQ-106-043) — never a silent partial.

Examples
--------
    KG_ROUTING_URI=bolt://10.53.2.10:7687 python scripts/context_pack.py build "PolicyEngine" --budget 2000
    python scripts/context_pack.py refactor "src/foo/bar.py" --budget 4000 --output pack.md
"""

from __future__ import annotations

import argparse
import os
import sys
from collections.abc import Mapping
from pathlib import Path
from typing import cast

# Ensure repo src is on path when run directly (mirrors scripts/kg_schema.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 neo4j import Driver  # noqa: E402
from neo4j.exceptions import GqlError  # noqa: E402

from orchestrata_ultimate.core.graph import BoltSyncRepository  # noqa: E402
from orchestrata_ultimate.core.graph_factory import create_graph  # noqa: E402
from orchestrata_ultimate.knowledge_graph.context_injector import TaskType  # noqa: E402
from orchestrata_ultimate.knowledge_graph.context_pack import (  # noqa: E402
    RequestFrame,
    analyze_request,
    empty_pack,
    pack,
    render,
    select,
)

# This CLI always presents itself as the "agent" requester — it serves
# swarm-task briefs (SPEC-106 FR-005), not interactive human/volk sessions.
# No --requester flag: the MVP scope is exactly {task-type, target, budget,
# output}, nothing more.
_CLI_REQUESTER = "agent"


class ConfigurationError(RuntimeError):
    """Raised when required environment configuration is missing."""


def _driver_from_env(env: Mapping[str, str] | None = None) -> Driver:
    """Build a graph driver from env, read AT CALL TIME — no hardcoded default.

    ``env`` defaults to ``os.environ`` but may be an explicit mapping so tests
    and callers can supply configuration without mutating the process
    environment.

    ``KG_ROUTING_URI``/``KG_URI`` unset is a configuration error, not "KG
    unavailable" — it
    means this invocation was never told where to connect at all, so it
    raises rather than falling back to any default host/port.
    """
    source = env if env is not None else os.environ
    uri = source.get("KG_ROUTING_URI") or source.get("KG_URI")
    if not uri:
        raise ConfigurationError(
            "Neither KG_ROUTING_URI nor KG_URI is set — scripts/context_pack.py "
            "requires an explicit graph connection (no hardcoded default). Set "
            "KG_ROUTING_URI (or KG_URI) before running; the endpoint negotiates "
            "no auth, so no credentials are needed."
        )
    user = ""
    password = ""
    return cast(
        Driver, create_graph(uri=uri, auth=(user, password), connection_timeout=5)
    )


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="context_pack",
        description="Emit a deterministic SPEC-106 context pack for an agent brief.",
    )
    parser.add_argument(
        "task_type",
        choices=[t.value for t in TaskType],
        help="Task type — drives the selection template.",
    )
    parser.add_argument(
        "target",
        help="Free-text target/request (e.g. a module, class, or description).",
    )
    parser.add_argument(
        "--budget", type=int, required=True, help="Token budget for the pack."
    )
    parser.add_argument(
        "--output", default=None, help="Write the pack to this file instead of stdout."
    )
    return parser


def run(argv: list[str] | None = None, env: Mapping[str, str] | None = None) -> int:
    parser = _build_parser()
    args = parser.parse_args(argv)
    task_type = TaskType(args.task_type)

    driver = _driver_from_env(env)
    try:
        driver.verify_connectivity()
        repo = BoltSyncRepository(driver=driver)
        frame = analyze_request(
            args.target,
            repo=repo,
            task_type=task_type,
            budget=args.budget,
            requester=_CLI_REQUESTER,
        )
        subgraph = select(frame, repo)
        context_pack = pack(subgraph, frame)
    except (GqlError, OSError) as exc:
        # KG unavailable (unreachable/refused/timed out) — explicit empty
        # pack with a reason, never a silent partial (REQ-106-043).
        fallback_frame = RequestFrame(
            task_type=task_type,
            entities=(),
            budget=args.budget,
            requester=_CLI_REQUESTER,
        )
        context_pack = empty_pack(fallback_frame, reason=f"graph unavailable: {exc}")
    finally:
        driver.close()

    body = render(context_pack)
    if args.output:
        Path(args.output).write_text(body, encoding="utf-8")
    else:
        print(body, end="")
    return 0


def main(argv: list[str] | None = None, env: Mapping[str, str] | None = None) -> None:
    try:
        sys.exit(run(argv, env=env))
    except ConfigurationError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()