scripts.gen_level0

Reference › scripts › gen_level0

scripts/gen_level0.py generates docs/LEVEL0.md, the always-in-context Level-0 foundation block that is imported into every session. A human runs it after the underlying facts change, and its --check mode is the drift gate form. It imports only dump_text_atomic from orchestrata_ultimate.core.atomic_json.

render assembles the block from the environment and the code. _parse_dotenv reads KEY=VALUE lines from a dotenv file without shell expansion, _env merges that with the process environment, _profile_from_env picks the active profile (default test), _host reduces a Bolt URI to host and port, _tool_names extracts dispatch tool names from the agents tool-registry dispatch table, and _volks lists the volks. Without arguments main writes the file atomically, because a session may be reading it at the moment it regenerates. With --check it writes nothing and exits 1 with a "LEVEL0 drift" message if the file is missing or stale, 0 if current.

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/gen_level0.py

#!/usr/bin/env python3
"""Generate ``docs/LEVEL0.md`` — the always-in-context Level-0 foundation block.

Level 0 is the small, always-loaded map that stops agents from re-deriving
fundamentals (are the graphs docker? where is the operational KG? which
Memgraph has the tools?). It is **generated** from authoritative sources so
it never goes stale:

  - infra endpoints  <- ``.env`` + ``env/<ORCHESTRATA_ENV_PROFILE>.env``
                        (default profile: ``test``)
  - tool inventory   <- ``agents/tool_registry.py`` dispatch table
  - volk roster      <- ``agents/volks/*/``

Every source above is REPO state. Session-scoped state is deliberately absent:
this file is COMMITTED and drift-gated, so anything that changes without a
commit turns the gate red for a reason that has nothing to do with the code.
The SPEC-023 session goal used to be rendered here from ``context.json`` and
did exactly that, twice — once mid-suite (the session CLI wrote ``context.json``
while the suite was running) and again on 2026-08-14 when ``/goal set`` left
the committed file stale and failed the ship. The recorded fix the first time
was an instruction to "hold session logging during suite runs"; an instruction
is not a mechanism, and it did not hold. The goal now reaches context the way
session state should — live, at session start, via ``project_context.sh``.

Design: ``plans/context-tiers-design.md``. Loaded via ``@docs/LEVEL0.md`` in CLAUDE.md.

Usage::

    python scripts/gen_level0.py            # regenerate docs/LEVEL0.md
    python scripts/gen_level0.py --check    # non-zero exit if the committed file is stale

Secrets are never embedded — passwords live in ``.env`` and are referenced, not copied.
Deterministic and offline: no network, no heavy imports, no DB connection.
"""

from __future__ import annotations

import os
import re
import sys
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
ENV = REPO / ".env"
ENV_DIR = REPO / "env"
TOOL_DISPATCH = REPO / "src" / "orchestrata_ultimate" / "agents" / "tool_registry.py"
VOLKS_DIR = REPO / "src" / "orchestrata_ultimate" / "agents" / "volks"
OUT = REPO / "docs" / "LEVEL0.md"

# Default profile for the generated doc. The test fleet is the authoritative
# suite target, so Level 0 describes that profile unless overridden.
DEFAULT_PROFILE = "test"

# Known profile subnets (for rendering fleet identity when an endpoint is
# missing or abbreviated). These are documented in INFRASTRUCTURE.md.
_PROFILE_SUBNETS: dict[str, str] = {
    "dev": "10.53.3.0/24",
    "test": "10.53.2.0/24",
    "prod": "10.53.1.0/24",
}


def _parse_dotenv(path: Path) -> dict[str, str]:
    """Parse KEY=VALUE lines from a dotenv file (no shell expansion)."""
    out: dict[str, str] = {}
    if not path.is_file():
        return out
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, val = line.partition("=")
        out[key.strip()] = val.strip()
    return out


def _env(profile: str | None = None) -> dict[str, str]:
    """Load secrets from ``.env`` then overlay the selected profile.

    With no profile, returns the legacy ``.env``-only view (used by tests and
    callers that have not migrated to profile selection). ``render()`` always
    passes the active profile.
    """
    out = _parse_dotenv(ENV)
    if profile is not None:
        out.update(_parse_dotenv(ENV_DIR / f"{profile}.env"))
    return out


def _host(uri: str) -> str:
    """bolt://10.53.2.10:7687 -> 10.53.2.10:7687 ; tolerate junk."""
    return re.sub(r"^\w+://", "", uri).strip()


def _profile_from_env() -> str:
    """Active profile from the environment, defaulting to ``test``."""
    return os.environ.get("ORCHESTRATA_ENV_PROFILE", DEFAULT_PROFILE)


def _tool_names() -> list[str]:
    """Extract dispatch tool names from the agents tool-registry dispatch table."""
    if not TOOL_DISPATCH.is_file():
        return []
    names = re.findall(
        r'tool_name == "([a-z_]+)"', TOOL_DISPATCH.read_text(encoding="utf-8")
    )
    seen: list[str] = []
    for n in names:
        if n not in seen:
            seen.append(n)
    return seen


def _volks() -> list[str]:
    if not VOLKS_DIR.is_dir():
        return []
    return sorted(
        p.name
        for p in VOLKS_DIR.iterdir()
        if p.is_dir() and not p.name.startswith(("_", "."))
    )


def render(profile: str | None = None) -> str:
    profile = profile or _profile_from_env()
    env = _env(profile)
    subnet = _PROFILE_SUBNETS.get(profile, f"<{profile}>")

    # Display generator (never connects): read with NO .get literal default per
    # RUL-003, and render an "(unset)"-style placeholder via boolean-`or` so the
    # doc still renders in a partially-configured env instead of raising.
    neo = _host(env.get("KG_ROUTING_URI") or env.get("KG_URI") or "")
    kg_database = env.get("KG_DATABASE") or "(unset)"
    redis = f"{env.get('REDIS_HOST') or '(unset)'}:{env.get('REDIS_PORT') or '(unset)'}"
    mg_single = _host(env.get("MEMGRAPH_URI") or "") or "(none)"
    cxp_host = env.get("CXP_GRPC_HOST") or "(unset)"
    gateway_probe = env.get("ORCHESTRATA_GATEWAY_PROBE_HOST") or "(unset)"
    prolog_hosts = env.get("PROLOG_HOSTS") or "(see INFRASTRUCTURE.md)"

    tools = _tool_names()
    volks = _volks()

    lines = [
        "# LEVEL 0 — always in context",
        "",
        f"> Generated by `scripts/gen_level0.py` from the active env profile (`{profile}`) + the tool registry.",
        "> **Do not hand-edit** — edit the sources and regenerate. Design: `plans/context-tiers-design.md`.",
        "> This is the foundation tier: where am I, what infra exists, what can I use. Detail is L1 (on demand).",
        "",
        "## Identity",
        "- **orchestrata-ultimate** — Python 3.14t free-threaded · `uv` (pip blocked) · absolute imports only.",
        "- Orchestrata is an agent orchestration engine. A human intent enters through Concierge."
        " Specialist volks each receive an LLM, injected context, and a persona. The graph is"
        " context. Redis is events and queues. Prolog computes what can be computed.",
        "- This checkout holds `packages/<name>/` beside `src/orchestrata_ultimate/`. Cut-over has not happened.",
        "- `origin/master` at `0faca7a7`, prod deployed 2026-10-19T11:41:39Z, is a backup-and-deploy of the"
        " in-progress tree (ADR-0113), not the cut-over. Cut-over remains: integration passes the fleet tier,"
        " the old `src/orchestrata_ultimate/*` trees are removed, one merge to master, and `/opt/orchestrata`"
        " ships from that tree.",
        "- **Session goal (SPEC-023):** injected live at session start"
        " (`project_context.sh` → `session_cli.py get-goal`), never baked into"
        " this committed file — `/goal show` for the current one.",
        "",
        f"## Infrastructure — Incus containers on `{subnet}` (`{profile}` profile) — NOT docker, NOT localhost",
        "`docker ps` / `ss -tlnp` show nothing locally; that is expected. `incus list` to inspect. Creds in `.env`; endpoints in `env/{dev,test,prod}.env`.",
        "",
        f"- **Profile:** `ORCHESTRATA_ENV_PROFILE={profile}` → `env/{profile}.env`.",
        f"- **Redis** `{redis}` — hot storage, queues, message broker.",
        f"- **Operational KG** `{neo}` database `{kg_database}` — Memgraph single node in this env (test/dev) or the prod HA orchgraph cluster.",
        f"- **Semantic / RAG graph** `{mg_single}` — shared Memgraph database; in the test/dev fleets this is the same single node as the operational KG, isolated by `database=`.",
        f"- **Prolog** `{prolog_hosts}` — policy engine / governance.",
        f"- **Kernel + Gateway (test fleet)** containerized in the `app` container at `{cxp_host}` (gRPC `{cxp_host}:50053`, WS probe `{gateway_probe}:8765`). See `deploy/orchestrata-test/`.",
        "",
        "### Environment profiles",
        "| Profile | Subnet | Incus project | Use |",
        "|---------|--------|---------------|-----|",
        "| `dev`   | `10.53.3.0/24` | `orchestrata-dev` | developer scratch |",
        "| `test`  | `10.53.2.0/24` | `orchestrata-test` | **authoritative suite target** |",
        "| `prod`  | `10.53.1.0/24` | `default` | live HA orchgraph cluster — NEVER a test target |",
        "",
        "Promotion: `python scripts/promote_env.py validate|test|promote --to <env>`. `tests/conftest.py` hard-pins the suite to `env/test.env` with a prod-subnet guard.",
        "",
        "## The two DATABASES — do not conflate them",
        f"- **`{kg_database}` database on `{neo.split(':')[0] if neo else '(unset)'}` = orchestrata's OPERATIONAL KG.** Sessions, gateway audit, CXP packets,",
        "  code intelligence (Function/Class/Method), specs, volks, ToolCall events (~39 labels).",
        "  Query: `scripts/graph_query.py`, `scripts/graph_browser.py`.",
        f"- **`memgraph` database on `{mg_single.split(':')[0] if mg_single != '(none)' else '(unset)'}` = the SHARED semantic/RAG graph.** BOTH the",
        "  `knowledge-wiki` project AND orchestrata's `semantic_graph` ingest (`scripts/ingest_orchestrata_ultimate.py`)",
        "  write here, so labels co-mingle (Entity, *Chunk, Module, Page, Tag, ADR, Tool). Query via the",
        "  `neo4j` driver (Bolt, auth=None) or `wiki-rag-query`.",
        f"- **Neither database models the tool registry** (`{kg_database}` has only `ToolCall` events; the semantic DB has ~2 wiki `Tool`",
        f"  nodes). Known data-quality debt: `{kg_database}` Session/audit are un-MERGEd/append-only; chunk labels are",
        "  split across `SemanticChunk`/`CodestralChunk`/`QwenChunk`/`Chunk`. See `plans/context-tiers-design.md`.",
        "",
        "## Tools & agents",
        f"- **{len(tools)} registered volk tools:** {', '.join(tools) if tools else '(see tools.registry)' }.",
        "  Live list + categories: `tools.registry` `ToolRegistry.metadata()/names()` (the SPEC-072 index).",
        f"- **Volks ({len(volks)}):** {', '.join(volks) if volks else '(see agents/volks/)' }.",
        "",
        "## Where the detail lives (L1 — pull on demand, don't preload)",
        "- **Code structure/symbols/BODIES** → AST toolkit `scripts/serialize_codebase_ast.py"
        " {symbols,refs,who-imports,source}` + `docs/CODEBASE_MAP.md` (`source <qname>` serves exact bytes).",
        "- **Impact/cycles/dead code/dependency path** → `scripts/code_intel.py"
        " {impact,cycles,unused,dep-path}` — deterministic Prolog code-intelligence queries.",
        "- **Requirement/test traceability, orphans, narrative** → `scripts/trace_intel.py"
        " {orphans,untraced-reqs,narrative}`.",
        "- **WHEN a symbol disappeared, and WHAT/HOW/WHY** → `scripts/history_intel.py"
        " {what,when,how,why,explain}` — the TIME dimension over the same KB."
        " `no-recorded-rationale` answers TRUE as a FINDING, not as a failure.",
        "- **Infra detail** → `INFRASTRUCTURE.md`.  **Specs/RTM** → `specs/` + `plans/`.  **Session state** → `session status`.",
        "",
        "## Non-negotiable rules",
        "- **The kernel is the beating heart — REQUIRED, never skip it.** Every suite run must START + gRPC-VERIFY the Agent Runtime kernel (`runtime_main`, `CXP_GRPC_PORT`/50053); a run that skips it is INVALID/false-green. `start_test_services.py` marks it `required=True`; opt out only via explicit `--skip kernel` (non-authoritative). See `docs/kernel-is-required.md`.",
        "- Tests: **no mock / no skip / no xdist**; real infra (Redis/graph auto). When a test catches a real bug, **fix the code, not the test**.",
        "- `uv` only (pip blocked) · absolute imports · `os.environ.get(...)` (no bare env, no monkeypatch) · declare deps in `pyproject.toml`.",
        "- **ZERO hand-reading of Python source** (user mandate 2026-08-02): structure via `symbols`/`refs`,"
        " content via `source <qname>` (exact bytes), refactors via the toolkit — never Read/grep .py files"
        " for what the AST layer serves. Ad-hoc Read is for non-Python artifacts only."
        " Review every source diff before committing. No AI attribution in commits.",
        "",
    ]
    return "\n".join(lines)


def main(argv: list[str]) -> int:
    content = render()
    if "--check" in argv:
        if not OUT.is_file():
            print(
                f"LEVEL0 drift: {OUT} does not exist — run scripts/gen_level0.py",
                file=sys.stderr,
            )
            return 1
        current = OUT.read_text(encoding="utf-8")
        if current != content:
            print(
                f"LEVEL0 drift: {OUT} is stale — run scripts/gen_level0.py",
                file=sys.stderr,
            )
            return 1
        print("LEVEL0 up to date.")
        return 0
    # Atomic: docs/LEVEL0.md is @import-ed into every session, so a reader can
    # be holding it open exactly when this regenerates. See dump_text_atomic.
    from orchestrata_ultimate.core.atomic_json import dump_text_atomic

    OUT.parent.mkdir(parents=True, exist_ok=True)
    dump_text_atomic(OUT, content)
    print(f"wrote {OUT} ({len(content)} bytes)")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))