src/yaml_edit/cli.py

"""Local-only CLI. Never contacts Kubernetes or overwrites an existing file."""

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

from . import _publication
from .engine import inspect, preview, preview_batch, propose, propose_batch
from .inference import InferenceBackend, InferenceConfig, InferenceEngine, InferenceError
from .models import (
    MAX_BATCH_ARTIFACT_BYTES,
    MAX_BATCH_CANDIDATE_BYTES,
    MAX_BATCH_OPERATIONS,
    MAX_BATCH_SCALAR_BYTES,
    BatchRequest,
    ErrorCode,
    YamlEditError,
    is_scalar,
    serialize_batch_proposal,
)
from .parsing import MAX_SOURCE_BYTES
from .suggestion import SuggestionDisposition


def _read(path: Path, *, limit: int = MAX_SOURCE_BYTES) -> str:
    with path.open("rb") as file:
        raw = file.read(limit + 1)
    if len(raw) > limit:
        raise ValueError(f"{path} exceeds the {limit}-byte limit.")
    # read_text() normalizes CRLF. Reading bytes preserves exact source text.
    return raw.decode("utf-8")


def _pairs(pairs: list[tuple[str, object]]) -> dict[str, object]:
    result: dict[str, object] = {}
    for key, value in pairs:
        if key in result:
            raise ValueError(f"Duplicate JSON key: {key!r}.")
        result[key] = value
    return result


def _constant(value: str) -> Never:
    raise ValueError(f"Not a finite JSON value: {value}.")


def _json(text: str) -> object:
    try:
        decoded: object = json.loads(text, object_pairs_hook=_pairs, parse_constant=_constant)
    except RecursionError as error:
        raise ValueError("JSON nesting is too deep.") from error
    else:
        return decoded


def _write_new(path: Path, text: str) -> None:
    raw = text.encode("utf-8")
    _publication.publish_new(path, raw, staging_prefix=".yaml-edit-")


def _arguments() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Precise, local-only edits to Deployment YAML.")
    commands = parser.add_subparsers(dest="command", required=True)
    discover = commands.add_parser(
        "inspect", help="List existing allowlisted editable fields as JSON"
    )
    discover.add_argument("source", type=Path)
    natural = commands.add_parser(
        "suggest", help="Interpret one instruction; deterministic baseline by default"
    )
    natural.add_argument("source", type=Path)
    natural.add_argument("--instruction", required=True, help="One complete bounded instruction")
    natural.add_argument(
        "--output", type=Path, help="New proposal file, written only for a supported result"
    )
    natural.add_argument(
        "--backend", choices=tuple(InferenceBackend), default=InferenceBackend.BASELINE
    )
    natural.add_argument(
        "--allow-experimental",
        action="store_true",
        help="Acknowledge learned inference is experimental and may misunderstand intent",
    )
    natural.add_argument("--config", dest="checkpoint_config", type=Path)
    natural.add_argument("--weights", type=Path)
    natural.add_argument("--device", choices=("cpu", "mps"))
    create = commands.add_parser("propose", help="Create a validated, snapshot-bound proposal")
    create.add_argument("source", type=Path)
    create.add_argument(
        "--target", required=True, help="Candidate ID from inspect (a JSON Pointer)"
    )
    create.add_argument("--value", required=True, help="JSON scalar: 3, or '\"512Mi\"'")
    create.add_argument("--output", type=Path, help="New proposal file (default: JSON to stdout)")
    create_batch = commands.add_parser(
        "propose-batch", help="Create one atomic proposal for explicit existing fields"
    )
    create_batch.add_argument("source", type=Path)
    create_batch.add_argument(
        "--set",
        dest="sets",
        action="append",
        nargs=2,
        required=True,
        metavar=("TARGET", "JSON_SCALAR"),
    )
    create_batch.add_argument(
        "--output", type=Path, help="New batch proposal file (default: JSON to stdout)"
    )
    for command in ("preview", "apply", "preview-batch", "apply-batch"):
        action = commands.add_parser(command, help=f"{command.title()} a saved proposal")
        action.add_argument("source", type=Path)
        action.add_argument("--proposal", type=Path, required=True)
        if command in ("apply", "apply-batch"):
            action.add_argument("--output", type=Path, required=True, help="New YAML output file")
            action.add_argument(
                "--yes", action="store_true", required=True, help="Confirm the edit"
            )
    return parser


def _suggest_config(args: argparse.Namespace) -> InferenceConfig:
    if args.backend == InferenceBackend.BASELINE and args.device is not None:
        raise ValueError("--device requires --backend learned.")
    device = args.device if args.device is not None else "cpu"
    try:
        return InferenceConfig(
            backend=InferenceBackend(args.backend),
            allow_experimental=args.allow_experimental,
            config_path=args.checkpoint_config,
            weights_path=args.weights,
            device=device,
        )
    except InferenceError as error:
        message = str(error).replace("allow_experimental=True", "--allow-experimental")
        raise ValueError(message) from error


def _suggest_command(
    source: str, instruction: str, output: Path | None, config: InferenceConfig
) -> int:
    if config.backend == InferenceBackend.LEARNED:
        print(
            "Warning: learned inference is experimental; review the proposal before applying.",
            file=sys.stderr,
        )
    result = InferenceEngine(config).suggest(source, instruction)
    if output is not None and result.proposal is not None:
        proposal = json.dumps(result.proposal.to_dict(), indent=2, ensure_ascii=False) + "\n"
        _write_new(output, proposal)
        print(f"Saved proposal to {output}; preview it before applying.", file=sys.stderr)
    print(json.dumps(result.to_dict(), indent=2, ensure_ascii=False))
    return {
        SuggestionDisposition.SUPPORTED: 0,
        SuggestionDisposition.NEEDS_CLARIFICATION: 3,
        SuggestionDisposition.UNSUPPORTED: 4,
    }[result.disposition]


def _propose_command(source: str, target: str, value_text: str, output: Path | None) -> None:
    try:
        value = _json(value_text)
    except ValueError as error:
        raise YamlEditError(ErrorCode.INVALID_VALUE, str(error)) from error
    proposal = propose(source, target, value)
    serialized = json.dumps(proposal.to_dict(), indent=2, ensure_ascii=False) + "\n"
    if output is None:
        sys.stdout.write(serialized)
        return
    _write_new(output, serialized)
    print(f"Saved proposal to {output}; preview it before applying.", file=sys.stderr)


def _bounded_cli_token(value: str, maximum: int) -> int:
    if len(value) > maximum:
        raise YamlEditError(ErrorCode.INVALID_PROPOSAL, "Batch argument exceeds its size limit.")
    try:
        size = len(value.encode("utf-8"))
    except UnicodeEncodeError as error:
        raise YamlEditError(
            ErrorCode.INVALID_PROPOSAL, "Batch argument is not valid UTF-8."
        ) from error
    if size > maximum:
        raise YamlEditError(ErrorCode.INVALID_PROPOSAL, "Batch argument exceeds its size limit.")
    return size


def _batch_requests_from_cli(sets: list[list[str]]) -> list[BatchRequest]:
    if not 1 <= len(sets) <= MAX_BATCH_OPERATIONS:
        raise YamlEditError(
            ErrorCode.INVALID_PROPOSAL,
            f"A batch requires 1 through {MAX_BATCH_OPERATIONS} --set arguments.",
        )
    total = 0
    requests: list[BatchRequest] = []
    for candidate_id, value_text in sets:
        total += _bounded_cli_token(candidate_id, MAX_BATCH_CANDIDATE_BYTES)
        total += _bounded_cli_token(value_text, MAX_BATCH_SCALAR_BYTES)
        if total > MAX_BATCH_ARTIFACT_BYTES:
            raise YamlEditError(
                ErrorCode.INVALID_PROPOSAL, "Batch arguments exceed their aggregate size limit."
            )
        try:
            value = _json(value_text)
        except ValueError as error:
            raise YamlEditError(ErrorCode.INVALID_VALUE, str(error)) from error
        if not is_scalar(value):
            raise YamlEditError(
                ErrorCode.INVALID_PROPOSAL, "Batch values must be finite JSON scalars."
            )
        requests.append(BatchRequest(candidate_id, value))
    return requests


def _propose_batch_command(source: str, sets: list[list[str]], output: Path | None) -> None:
    proposal = propose_batch(source, _batch_requests_from_cli(sets))
    serialized = serialize_batch_proposal(proposal)
    if output is None:
        sys.stdout.write(serialized)
        return
    _write_new(output, serialized)
    print(f"Saved batch proposal to {output}; preview it before applying.", file=sys.stderr)


def _review_command(source: str, proposal_path: Path, output: Path | None) -> None:
    try:
        data = _json(_read(proposal_path, limit=65_536))
    except ValueError as error:
        raise YamlEditError(ErrorCode.INVALID_PROPOSAL, str(error)) from error
    result = preview(source, data)
    if output is None:
        sys.stdout.write(result.diff)
        if not result.changed:
            print("No change needed.", file=sys.stderr)
        return
    _write_new(output, result.source)
    print(f"Wrote {output}; original file unchanged.", file=sys.stderr)


def _batch_review_command(source: str, proposal_path: Path, output: Path | None) -> None:
    try:
        data = _json(_read(proposal_path, limit=MAX_BATCH_ARTIFACT_BYTES))
    except ValueError as error:
        raise YamlEditError(ErrorCode.INVALID_PROPOSAL, str(error)) from error
    result = preview_batch(source, data)
    if output is None:
        sys.stdout.write(result.diff)
        if not result.changed:
            print("No change needed.", file=sys.stderr)
        return
    _write_new(output, result.source)
    print(f"Wrote {output}; original file unchanged.", file=sys.stderr)


def main(argv: list[str] | None = None) -> int:
    """Run the local CLI, returning a documented status for each outcome."""
    args = _arguments().parse_args(argv)
    try:
        source = _read(args.source)
        if args.command == "inspect":
            print(json.dumps(inspect(source).to_dict(), indent=2, ensure_ascii=False))
        elif args.command == "suggest":
            return _suggest_command(source, args.instruction, args.output, _suggest_config(args))
        elif args.command == "propose":
            _propose_command(source, args.target, args.value, args.output)
        elif args.command == "propose-batch":
            _propose_batch_command(source, args.sets, args.output)
        elif args.command == "preview":
            _review_command(source, args.proposal, None)
        elif args.command == "apply":
            _review_command(source, args.proposal, args.output)
        elif args.command == "preview-batch":
            _batch_review_command(source, args.proposal, None)
        else:
            _batch_review_command(source, args.proposal, args.output)
    except YamlEditError as error:
        print(f"{error.code}: {error}", file=sys.stderr)
        return 2
    except (OSError, ValueError) as error:
        print(f"error: {error}", file=sys.stderr)
        return 2
    return 0