GEWU/scripts/export_agent_contract.py

"""Bind the GEWU agent pack to the MCP tools actually exposed by the service."""

from __future__ import annotations

import asyncio
import hashlib
import json
import shutil
import tempfile
from pathlib import Path
from typing import Any

from gewu.adapters.git_repository import GitCanonicalRepository
from gewu.adapters.sqlite_identity import SQLiteIdentityStore
from gewu.adapters.sqlite_search import SQLiteSearchIndex
from gewu.adapters.text_extractor import TextMarkdownExtractor
from gewu.application.acceptance import AcceptanceService
from gewu.application.context import ValidationContext
from gewu.application.reading import CanonicalNodeReader, CanonicalSourceReader
from gewu.application.search import CanonicalSearchService
from gewu.application.uploads import LocalArtifactStore
from gewu.domain.operations import Transport, operations_for
from gewu.transports.mcp_server import create_mcp_server

ROOT = Path(__file__).resolve().parents[1]
PACK = ROOT / "agents" / "gewu"
CONTRACT_PATH = PACK / "contracts" / "service-v1.json"
SCHEMA_OUTPUT = PACK / "contracts" / "schemas"
CATALOG_OUTPUT = PACK / "contracts" / "catalog-v1.json"


def _sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


async def _tool_inventory(data_root: Path) -> list[dict[str, Any]]:
    repository = GitCanonicalRepository(data_root / "knowledge.git")
    identities = SQLiteIdentityStore(data_root / "identity.sqlite3")
    artifact_store = LocalArtifactStore(data_root / "artifacts")

    def context_provider() -> ValidationContext:
        return ValidationContext({}, {}, {}, {}, {})

    acceptance_service = AcceptanceService(
        repository=repository,
        artifact_store=artifact_store,
        workflow_root=data_root / "workflow",
        context_provider=context_provider,
        actor_is_current=lambda _actor: True,
    )
    search_service = CanonicalSearchService(
        index=SQLiteSearchIndex(data_root / "index.sqlite3", repository),
        project_ids_provider=lambda: (),
        canonical_head=repository.head,
    )
    server = create_mcp_server(
        identity_provider=identities,
        node_reader=CanonicalNodeReader(repository),
        source_reader=CanonicalSourceReader(repository),
        search_service=search_service,
        artifact_store=artifact_store,
        extractor=TextMarkdownExtractor(),
        acceptance_service=acceptance_service,
        context_provider=context_provider,
    )
    tools = await server.list_tools()
    return [
        {
            "name": tool.name,
            "description": tool.description,
            "input_schema": tool.input_schema,
        }
        for tool in tools
    ]


def build_contract(data_root: Path, *, schema_output: Path | None = None) -> dict[str, Any]:
    catalog_path = ROOT / "schema" / "v1" / "catalog.json"
    catalog = json.loads(catalog_path.read_text(encoding="utf-8"))
    tools = asyncio.run(_tool_inventory(data_root))
    expected_names = {item.name for item in operations_for(Transport.MCP)}
    actual_names = {tool["name"] for tool in tools}
    if actual_names != expected_names:
        raise RuntimeError("MCP tool inventory does not match the explicit operation catalog")
    if schema_output is not None:
        schema_output.mkdir(parents=True, exist_ok=True)
    record_schemas = []
    for source in sorted((ROOT / "schema" / "v1" / "records").glob("*.schema.json")):
        destination = schema_output / source.name if schema_output is not None else source
        if schema_output is not None:
            shutil.copyfile(source, destination)
        record_schemas.append({"path": f"schemas/{source.name}", "sha256": _sha256(destination)})
    return {
        "contract_id": "gewu.service.v1",
        "contract_version": 1,
        "api_version": catalog["api_version"],
        "catalog_sha256": _sha256(catalog_path),
        "operations": [item for item in catalog["operations"] if "mcp" in item["transports"]],
        "record_schemas": record_schemas,
        "tools": tools,
    }


def main() -> None:
    CONTRACT_PATH.parent.mkdir(parents=True, exist_ok=True)
    shutil.copyfile(ROOT / "schema" / "v1" / "catalog.json", CATALOG_OUTPUT)
    with tempfile.TemporaryDirectory(prefix="gewu-agent-contract-") as temporary:
        contract = build_contract(Path(temporary), schema_output=SCHEMA_OUTPUT)
    CONTRACT_PATH.write_text(
        json.dumps(contract, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )


if __name__ == "__main__":
    main()