| """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.") |
| |
| 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 |
|
|
