skills/conform/scripts/conform_packets.py

#!/usr/bin/env python3
"""Build, dispatch, and expand conform's mapping-stage packets; manage its workspace."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any

from _runtime_gate import require_supported_python

require_supported_python()

from conform.packets import (  # noqa: E402
    PacketError,
    build_mapping_packets,
    canonical_response_path,
    expand_responses,
    prefill_response,
    validate_response,
)
from conform.prompts import render_mapping_dispatch_prompt  # noqa: E402
from conform.workspace import (  # noqa: E402
    WorkspaceError,
    cleanup_workspace,
    create_workspace,
    require_within_workspace,
)


def _read_json(path: Path) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8-sig"))
    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
        raise PacketError(f"unable to read JSON artifact {path}") from exc


def _write_json(path: Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8"
    )


def _cmd_workspace_create(args: argparse.Namespace) -> dict[str, Any]:
    root, marker = create_workspace(
        fingerprint=args.fingerprint, workspace_root=args.workspace_root
    )
    return {"work_dir": str(root), "marker": marker}


def _cmd_workspace_cleanup(args: argparse.Namespace) -> dict[str, Any]:
    root = cleanup_workspace(args.work_dir)
    return {"cleaned_up": str(root)}


def _cmd_build(args: argparse.Namespace) -> dict[str, Any]:
    source_ledger = _read_json(args.source_ledger)
    core_ledger = _read_json(args.core_ledger)
    selected_text = None
    if args.selected_text is not None:
        payload = _read_json(args.selected_text)
        if not isinstance(payload, dict) or not isinstance(payload.get("text"), str):
            raise PacketError(
                "--selected-text file must contain a JSON object with a text field"
            )
        selected_text = payload["text"]

    packets, manifest = build_mapping_packets(
        source_ledger,
        core_ledger,
        mode=args.mode,
        selected_text=selected_text,
        target_items_per_packet=args.target_items_per_packet,
    )

    work_dir = args.work_dir.resolve()
    packet_dir = require_within_workspace(
        work_dir / "packets" / "mapping", work_dir, label="packet directory"
    )
    packet_dir.mkdir(parents=True, exist_ok=True)
    packet_paths = []
    for packet in packets:
        packet_path = packet_dir / f"packet-{packet['packet']:03d}.json"
        if packet_path.exists():
            raise PacketError(f"refusing to overwrite existing packet: {packet_path}")
        _write_json(packet_path, packet)
        packet_paths.append(str(packet_path))

    manifest_path = require_within_workspace(
        work_dir / "private" / "mapping-manifest.json", work_dir, label="manifest path"
    )
    if manifest_path.exists():
        raise PacketError(f"refusing to overwrite existing manifest: {manifest_path}")
    _write_json(manifest_path, manifest)

    return {
        "manifest": str(manifest_path),
        "packets": packet_paths,
        "item_count": len(manifest["items"]),
        "packet_count": len(packets),
    }


def _cmd_render_dispatch(args: argparse.Namespace) -> dict[str, Any]:
    work_dir = args.work_dir.resolve()
    packet_path = require_within_workspace(args.packet, work_dir, label="packet")
    packet = _read_json(packet_path)
    response_path = prefill_response(work_dir, packet)
    validation_command = (
        "python conform_packets.py validate-response --work-dir "
        f"{work_dir} --manifest {work_dir / 'private' / 'mapping-manifest.json'} "
        f"--packet-number {packet['packet']} --response {response_path}"
    )
    prompt = render_mapping_dispatch_prompt(
        packet_path=packet_path,
        response_path=response_path,
        validation_command=validation_command,
    )
    return {"response_path": str(response_path), "prompt": prompt}


def _cmd_validate_response(args: argparse.Namespace) -> dict[str, Any]:
    work_dir = args.work_dir.resolve()
    manifest = _read_json(args.manifest)
    packet_path = (
        work_dir / "packets" / "mapping" / f"packet-{args.packet_number:03d}.json"
    )
    packet = _read_json(packet_path)
    response = _read_json(args.response)
    validate_response(packet, manifest, response)
    return {"valid": True, "packet": args.packet_number}


def _cmd_expand(args: argparse.Namespace) -> dict[str, Any]:
    work_dir = args.work_dir.resolve()
    manifest = _read_json(args.manifest)
    packets_by_number = {}
    responses_by_number = {}
    for entry in manifest.get("packets", []):
        number = entry["packet"]
        packet_path = work_dir / "packets" / "mapping" / f"packet-{number:03d}.json"
        packets_by_number[number] = _read_json(packet_path)
        response_path = canonical_response_path(work_dir, number)
        if not response_path.is_file():
            raise PacketError(
                f"missing canonical mapping response for packet {number}: {response_path}"
            )
        responses_by_number[number] = _read_json(response_path)

    bundle = expand_responses(manifest, packets_by_number, responses_by_number)
    if args.base_bundle is not None and args.base_bundle.is_file():
        base = _read_json(args.base_bundle)
        if isinstance(base, dict) and isinstance(base.get("decisions"), list):
            bundle["decisions"] = base["decisions"] + bundle["decisions"]

    output_path = args.output.resolve()
    _write_json(output_path, bundle)
    return {"output": str(output_path), "decision_count": len(bundle["decisions"])}


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Build conform mapping-stage packets or expand compact worker responses."
    )
    commands = parser.add_subparsers(dest="command", required=True)

    create = commands.add_parser(
        "workspace-create", help="Create a marked run workspace"
    )
    create.add_argument("--fingerprint")
    create.add_argument("--workspace-root", type=Path)

    cleanup = commands.add_parser(
        "workspace-cleanup", help="Delete one exact marked temporary workspace"
    )
    cleanup.add_argument("--work-dir", type=Path, required=True)

    build = commands.add_parser(
        "build", help="Build mapping packets and a private manifest"
    )
    build.add_argument("--stage", choices=("mapping",), required=True)
    build.add_argument("--work-dir", type=Path, required=True)
    build.add_argument("--source-ledger", type=Path, required=True)
    build.add_argument("--core-ledger", type=Path, required=True)
    build.add_argument("--selected-text", type=Path)
    build.add_argument(
        "--mode",
        choices=("conform_selected_text", "precedent_leakage_check"),
        default="conform_selected_text",
    )
    build.add_argument("--target-items-per-packet", type=int, default=6)

    dispatch = commands.add_parser(
        "render-dispatch", help="Render the standardized generic-worker task"
    )
    dispatch.add_argument("--stage", choices=("mapping",), required=True)
    dispatch.add_argument("--work-dir", type=Path, required=True)
    dispatch.add_argument("--packet", type=Path, required=True)

    validate = commands.add_parser(
        "validate-response", help="Validate one compact worker response"
    )
    validate.add_argument("--stage", choices=("mapping",), required=True)
    validate.add_argument("--work-dir", type=Path, required=True)
    validate.add_argument("--manifest", type=Path, required=True)
    validate.add_argument("--packet-number", type=int, required=True)
    validate.add_argument("--response", type=Path, required=True)

    expand = commands.add_parser(
        "expand", help="Expand validated responses into a mapping bundle"
    )
    expand.add_argument("--stage", choices=("mapping",), required=True)
    expand.add_argument("--work-dir", type=Path, required=True)
    expand.add_argument("--manifest", type=Path, required=True)
    expand.add_argument("--output", type=Path, required=True)
    expand.add_argument("--base-bundle", type=Path)

    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    handlers = {
        "workspace-create": _cmd_workspace_create,
        "workspace-cleanup": _cmd_workspace_cleanup,
        "build": _cmd_build,
        "render-dispatch": _cmd_render_dispatch,
        "validate-response": _cmd_validate_response,
        "expand": _cmd_expand,
    }
    try:
        result = handlers[args.command](args)
    except (PacketError, WorkspaceError) as exc:
        print(
            json.dumps(
                {"status": "failed", "error": str(exc)}, indent=2, sort_keys=True
            ),
            file=sys.stderr,
        )
        return 1
    print(json.dumps(result, indent=2, sort_keys=True))
    return 0


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