feat(npu): add local context gate advisory

2026-06-05 15:52:42 -07:00
parent 24d620e9c9
commit b7b4edf0f5
7 changed files with 884 additions and 0 deletions
@@ -0,0 +1,90 @@
+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())