William Valentin
91 lines
3.5 KiB
Python
91 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from typing import Any
|
|
|
|
from .context_gate import (
|
|
DEFAULT_CLASSIFIER_URL,
|
|
ContextGateError,
|
|
build_plan,
|
|
classify_live,
|
|
classify_offline,
|
|
compact_json,
|
|
compact_line,
|
|
)
|
|
|
|
|
|
def _parse_context(raw_items: list[str]) -> dict[str, Any]:
|
|
context: dict[str, Any] = {}
|
|
for item in raw_items:
|
|
if "=" not in item:
|
|
raise ContextGateError(f"invalid_context_item:{item}")
|
|
key, value = item.split("=", 1)
|
|
if not key:
|
|
raise ContextGateError("invalid_context_key")
|
|
if value.lower() == "true":
|
|
parsed: Any = True
|
|
elif value.lower() == "false":
|
|
parsed = False
|
|
else:
|
|
parsed = value
|
|
context[key] = parsed
|
|
return context
|
|
|
|
|
|
def build_arg_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
description="Emit a local-only Atlas/Hermes advisory context bundle plan. No routing, retrieval, memory writes, sends, restarts, or vector mutations are performed.",
|
|
)
|
|
parser.add_argument("--query", required=True, help="Non-private query to plan for")
|
|
parser.add_argument("--format", choices=["compact", "compact-json", "json"], default="compact")
|
|
parser.add_argument("--context", action="append", default=[], metavar="KEY=VALUE", help="Optional compact request context, e.g. platform=kanban repo_path=/path")
|
|
parser.add_argument("--max-sources", type=int, default=4)
|
|
parser.add_argument("--trace-id")
|
|
parser.add_argument("--classifier-url", default=DEFAULT_CLASSIFIER_URL)
|
|
parser.add_argument("--classifier-timeout", type=float, default=8.0)
|
|
parser.add_argument("--offline", action="store_true", help="Use deterministic heuristic labels; makes no NPU claim")
|
|
parser.add_argument("--allow-offline-fallback", action="store_true", help="If live classifier is unavailable, emit an advisory fallback plan with npu_verified=false")
|
|
parser.add_argument("--no-require-npu-proof", action="store_true", help="Do not add npu_proof_inconclusive warning when running offline/fallback")
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = build_arg_parser()
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
context = _parse_context(args.context)
|
|
options = {
|
|
"dry_run": True,
|
|
"max_sources": args.max_sources,
|
|
"include_private_text": False,
|
|
"require_npu_proof": not args.no_require_npu_proof,
|
|
"trace_id": args.trace_id,
|
|
}
|
|
if args.offline:
|
|
classifier = classify_offline(args.query, context)
|
|
else:
|
|
try:
|
|
classifier = classify_live(args.query, context, classifier_url=args.classifier_url, timeout=args.classifier_timeout)
|
|
except ContextGateError as exc:
|
|
if not args.allow_offline_fallback:
|
|
raise
|
|
classifier = classify_offline(args.query, context, warning=str(exc))
|
|
plan = build_plan(args.query, context=context, options=options, classifier=classifier)
|
|
except ContextGateError as exc:
|
|
print(f"error={exc}", file=sys.stderr)
|
|
return 2
|
|
|
|
if args.format == "json":
|
|
print(json.dumps(plan, indent=2, sort_keys=True))
|
|
elif args.format == "compact-json":
|
|
print(compact_json(plan))
|
|
else:
|
|
print(compact_line(plan))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
raise SystemExit(main())
|