feat: NVIDIA NIM model discovery (issue #86 scaffold)

CHANGELOG.md

@@ -0,0 +1,36 @@
+# Changelog
+
+All notable changes to this project are documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## [Unreleased]
+
+### Fixed
+- NIM discovery catalog body bound + dry-run call budget uses max_steps;
+  offline cost-quality rejects malformed scripted answers and zeros failed-cell usage.
+
+### Added
+- Offline NIM capability probe plan + fixture classification (issue #86).
+- Offline cost-quality `--use-mock-orchestrator` path: Fugu `route_once` and
+  Conductor/TRINITY `conduct` via `mock://` agents (issue #86 paper-path exercise).
+
+### Added
+- Offline NIM cost-quality comparison harness (`nim_cost_quality` +
+  `nim-cost-quality-offline` CLI) for issue #86 post-discovery: locked task
+  manifest scorers, honest unknown actual/hypothetical cost, policy summaries,
+  and quality-latency / quality-cost Pareto frontiers without live egress.
+- Offline NIM capability inventory + dry-run benchmark plan (issue #86).
+- `discover-nim-models` CLI and `nim_discovery` module (issue #86): allowlisted
+  NVIDIA HTTPS `/v1/models` only; offline fixture status; unique agent ids on
+  slug collision; live tests require `RUN_LIVE_NIM_TESTS=1`.
+- Role-differentiated sampling temperatures for paper-role ablation.
+
+### Security
+- Semgrep nosemgrep hygiene for audited SQL placeholders / TLS opt-out / urllib.
+
+## [0.1.0] - 2026-07-13
+
+### Added
+- Initial OpenAI-compatible orchestration gateway.

README.md

@@ -285,3 +285,35 @@ python tests/test_commercial_purchase_approval_packet.py
 python tests/test_commercial_due_diligence_room.py
 python tests/test_commercial_investment_committee_memo.py
 ```
+
+### NIM model discovery
+
+```bash
+python -m contextual_orchestrator discover-nim-models
+python -m contextual_orchestrator discover-nim-models --as-agent-pool
+python -m contextual_orchestrator discover-nim-models --benchmark-dry-run
+python -m contextual_orchestrator discover-nim-models --capability-probe-plan
+python -m contextual_orchestrator discover-nim-models --capability-probe-dry-run examples/nim_capability_probe_fixtures.json
+```
+
+Requires `NVIDIA_NIM_API_KEY` in the KV (`register-credential`) for live catalog
+listing. Offline dry-run / capability inventory paths stay secret-free.
+
+### Offline NIM cost-quality (issue #86)
+
+```bash
+python -m contextual_orchestrator nim-cost-quality-offline \
+  --task-manifest examples/nim_task_manifest_offline.json
+python -m contextual_orchestrator nim-cost-quality-offline \
+  --task-manifest examples/nim_task_manifest_offline.json \
+  --pricing-scenario examples/nim_pricing_scenario_offline.json \
+  --markdown
+python -m contextual_orchestrator nim-cost-quality-offline \
+  --use-mock-orchestrator --agents examples/agents.mock.json --markdown
+python tests/test_nim_cost_quality.py
+```
+
+Compares route/conduct/direct policies on a locked task set with honest
+`unknown` costs unless a pricing scenario covers every model. Optional
+`--use-mock-orchestrator` exercises Fugu `route_once` and Conductor/TRINITY
+`conduct` on `mock://` agents. Never uses `COPILOT_GITHUB_TOKEN` for model calls.

contextual_orchestrator/__main__.py

@@ -55,11 +55,238 @@ def _register_credential_command(argv: list[str]) -> None:
     print(json.dumps({"registered": args.name, "backend": "kv"}, ensure_ascii=False))
 
 
+
+def _discover_nim_models_command(argv: list[str]) -> None:
+    """List NIM model IDs via KV credential and print agent-pool JSON candidates."""
+    from .nim_discovery import (
+        DEFAULT_NIM_MODELS_URL,
+        NimDiscoveryError,
+        build_benchmark_plan_dry_run,
+        build_capability_inventory,
+        build_capability_probe_plan,
+        discover_nim_models,
+        models_to_agent_pool_entries,
+        run_capability_probes_dry_run,
+        validate_nim_models_url,
+    )
+
+    parser = argparse.ArgumentParser(
+        prog="python -m contextual_orchestrator discover-nim-models",
+        description="Discover NVIDIA NIM model IDs using the KV credential NVIDIA_NIM_API_KEY.",
+    )
+    parser.add_argument(
+        "--models-url",
+        default=DEFAULT_NIM_MODELS_URL,
+        help=(
+            "HTTPS NVIDIA catalog URL (default: integrate.api.nvidia.com/v1/models). "
+            "Only allowlisted NVIDIA hosts with path /v1/models are accepted; "
+            "the API key is never sent to other origins."
+        ),
+    )
+    parser.add_argument(
+        "--as-agent-pool",
+        action="store_true",
+        help="Emit agent-pool JSON entries instead of the discovery report.",
+    )
+    parser.add_argument(
+        "--capability-inventory",
+        action="store_true",
+        help="Emit offline capability-hint inventory for discovered model ids (issue #86 dry path).",
+    )
+    parser.add_argument(
+        "--benchmark-dry-run",
+        action="store_true",
+        help="Emit a fail-closed dry-run benchmark plan with unknown costs (issue #86).",
+    )
+    parser.add_argument(
+        "--hard-request-budget",
+        type=int,
+        default=100,
+        help="Hard call budget for dry-run admission (default: 100).",
+    )
+    parser.add_argument(
+        "--capability-probe-plan",
+        action="store_true",
+        help="Emit offline capability probe plan (models x probe kinds) without network.",
+    )
+    parser.add_argument(
+        "--capability-probe-dry-run",
+        metavar="FIXTURE_JSON",
+        default=None,
+        help=(
+            "Classify offline probe fixtures from JSON list of "
+            "{model_id, probe_kind, status_code|error_class, body?} rows."
+        ),
+    )
+    args = parser.parse_args(argv)
+    try:
+        models_url = validate_nim_models_url(args.models_url)
+    except NimDiscoveryError as exc:
+        parser.error(str(exc))
+    if args.capability_probe_dry_run:
+        try:
+            with open(args.capability_probe_dry_run, encoding="utf-8") as handle:
+                fixtures = json.load(handle)
+            plan = run_capability_probes_dry_run(
+                fixtures if isinstance(fixtures, list) else fixtures.get("probe_rows") or fixtures.get("fixtures") or [],
+                hard_request_budget=args.hard_request_budget,
+            )
+        except (NimDiscoveryError, OSError, ValueError, TypeError) as exc:
+            parser.error(str(exc))
+        print(json.dumps(plan, ensure_ascii=False, indent=2))
+        return
+
+    report = discover_nim_models(models_url=models_url)
+    model_ids = report.get("model_ids") or []
+    if args.capability_probe_plan:
+        try:
+            plan = build_capability_probe_plan(
+                model_ids, hard_request_budget=args.hard_request_budget
+            )
+        except NimDiscoveryError as exc:
+            parser.error(str(exc))
+        print(json.dumps(plan, ensure_ascii=False, indent=2))
+        return
+    if args.benchmark_dry_run:
+        try:
+            plan = build_benchmark_plan_dry_run(
+                model_ids, hard_request_budget=args.hard_request_budget
+            )
+        except NimDiscoveryError as exc:
+            parser.error(str(exc))
+        print(json.dumps(plan, ensure_ascii=False, indent=2))
+    elif args.capability_inventory:
+        print(json.dumps(build_capability_inventory(model_ids), ensure_ascii=False, indent=2))
+    elif args.as_agent_pool:
+        print(json.dumps(models_to_agent_pool_entries(model_ids), ensure_ascii=False, indent=2))
+    else:
+        print(json.dumps(report, ensure_ascii=False, indent=2))
+
+
+def _nim_cost_quality_offline_command(argv: list[str]) -> None:
+    """Run the offline cost-quality harness against a locked task manifest (issue #86)."""
+    from .nim_cost_quality import (
+        CostQualityContractError,
+        build_orchestrator_policy_runners,
+        build_scripted_policy_runners,
+        load_pricing_scenario,
+        load_task_manifest,
+        locked_evaluation_tasks,
+        render_cost_quality_markdown,
+        run_offline_cost_quality,
+        validate_scripted_answers,
+    )
+    from .orchestrator import TaskOrchestrator, load_agents
+
+    parser = argparse.ArgumentParser(
+        prog="python -m contextual_orchestrator nim-cost-quality-offline",
+        description=(
+            "Offline cost-quality comparison for issue #86 (post-discovery). "
+            "Uses scripted answers by default so CI never needs NVIDIA_NIM_API_KEY. "
+            "Pass --use-mock-orchestrator to drive Fugu route_once vs Conductor "
+            "conduct through mock:// agents. Never invents prices."
+        ),
+    )
+    parser.add_argument(
+        "--task-manifest",
+        default="examples/nim_task_manifest_offline.json",
+        help="Path to the versioned task manifest (locked split only).",
+    )
+    parser.add_argument(
+        "--pricing-scenario",
+        default=None,
+        help="Optional USD-per-million-token scenario JSON; omit to keep costs unknown.",
+    )
+    parser.add_argument(
+        "--scripted-answers",
+        default=None,
+        help=(
+            "Optional JSON map {task_id: {policy_name: answer}}. "
+            "When omitted, answers are empty (scores zero) for structural dry-run only."
+        ),
+    )
+    parser.add_argument(
+        "--use-mock-orchestrator",
+        action="store_true",
+        help=(
+            "Run policies via TaskOrchestrator route_once/conduct on --agents "
+            "(default examples/agents.mock.json). Mutually exclusive with "
+            "--scripted-answers."
+        ),
+    )
+    parser.add_argument(
+        "--agents",
+        default="examples/agents.mock.json",
+        help="Agent pool JSON for --use-mock-orchestrator (mock:// recommended).",
+    )
+    parser.add_argument(
+        "--model-id",
+        default="mock-scripted",
+        help="Model id recorded on cells and used for pricing lookups (default: mock-scripted).",
+    )
+    parser.add_argument(
+        "--markdown",
+        action="store_true",
+        help="Emit a short markdown summary instead of the full JSON report.",
+    )
+    args = parser.parse_args(argv)
+    if args.use_mock_orchestrator and args.scripted_answers:
+        parser.error("--use-mock-orchestrator cannot be combined with --scripted-answers")
+    try:
+        manifest = load_task_manifest(args.task_manifest)
+        tasks = locked_evaluation_tasks(manifest)
+        pricing = load_pricing_scenario(args.pricing_scenario)
+        if args.use_mock_orchestrator:
+            agents = load_agents(args.agents)
+            if not agents:
+                parser.error("--agents pool is empty")
+            non_mock = [a.id for a in agents if not str(a.base_url).startswith("mock://")]
+            if non_mock:
+                parser.error(
+                    "--use-mock-orchestrator requires mock:// agents only; "
+                    f"non-mock: {non_mock}"
+                )
+            orchestrator = TaskOrchestrator(agents)
+            runners = build_orchestrator_policy_runners(orchestrator)
+            model_id = args.model_id if args.model_id != "mock-scripted" else "mock-orchestrator"
+        else:
+            answers: dict = {}
+            if args.scripted_answers:
+                with open(args.scripted_answers, encoding="utf-8") as handle:
+                    raw_answers = json.load(handle)
+                answers = validate_scripted_answers(raw_answers)
+            runners = build_scripted_policy_runners(answers, model_id=args.model_id)
+            model_id = args.model_id
+        report = run_offline_cost_quality(
+            tasks=tasks,
+            policy_runners=runners,
+            model_id=model_id,
+            pricing_scenario=pricing,
+        )
+        if args.use_mock_orchestrator:
+            report["runner_backend"] = "mock_orchestrator"
+            report["agent_pool_path"] = args.agents
+        else:
+            report["runner_backend"] = "scripted_answers"
+    except (CostQualityContractError, OSError, ValueError) as exc:
+        parser.error(str(exc))
+    if args.markdown:
+        print(render_cost_quality_markdown(report))
+    else:
+        print(json.dumps(report, ensure_ascii=False, indent=2))
+
+
 def main() -> None:
     """Parse CLI options and run bootstrap, prompt completion, or the HTTP server."""
     if len(sys.argv) > 1 and sys.argv[1] == "register-credential":
         _register_credential_command(sys.argv[2:])
         return
+    if len(sys.argv) > 1 and sys.argv[1] == "discover-nim-models":
+        _discover_nim_models_command(sys.argv[2:])
+        return
+    if len(sys.argv) > 1 and sys.argv[1] == "nim-cost-quality-offline":
+        _nim_cost_quality_offline_command(sys.argv[2:])
+        return
 
     parser = argparse.ArgumentParser(description="Route or conduct chat requests across model agents.")
     parser.add_argument("prompt", nargs="?", help="User prompt for CLI mode.")

contextual_orchestrator/cost_ledger.py

@@ -583,12 +583,12 @@ def _seed_dimension_catalog(self) -> None:
         ph = self._placeholder()
         cur = self._conn.cursor()
         for order, (name, label, _column) in enumerate(ATTRIBUTION_DIMENSION_CATALOG):
-            cur.execute(
+            cur.execute(  # nosemgrep -- sqlalchemy-execute-raw-query FP: only the DB-API placeholder char is interpolated; the value is bound.
                 f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}",  # nosec B608 - ph is a DB-API placeholder.
                 (name,),
             )
             if cur.fetchone() is None:
-                cur.execute(
+                cur.execute(  # nosemgrep -- sqlalchemy-execute-raw-query FP: only DB-API placeholder chars are interpolated; values are bound.
                     "INSERT INTO cost_attribution_dimensions "
                     f"(dimension_name, dimension_label, dimension_order) VALUES ({ph}, {ph}, {ph})",  # nosec B608 - ph is a DB-API placeholder.
                     (name, label, order),
@@ -602,7 +602,7 @@ def append(self, record: UsageRecord) -> None:
         placeholders = ", ".join(ph for _ in _USAGE_COLUMNS)
         columns = ", ".join(_USAGE_COLUMNS)
         cur = self._conn.cursor()
-        cur.execute(
+        cur.execute(  # nosemgrep -- sqlalchemy-execute-raw-query FP: columns are the fixed _USAGE_COLUMNS constant; values are bound.
             f"INSERT INTO llm_usage_records ({columns}) VALUES ({placeholders})",  # nosec B608 - columns are fixed _USAGE_COLUMNS.
             tuple(row.get(column) for column in _USAGE_COLUMNS),
         )
@@ -622,7 +622,7 @@ def query(self, start: Optional[int] = None, end: Optional[int] = None) -> List[
         where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
         columns = ", ".join(_USAGE_COLUMNS)
         cur = self._conn.cursor()
-        cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params))  # nosec B608 - columns and clauses are fixed.
+        cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params))  # nosec B608 - columns and clauses are fixed.  # nosemgrep -- sqlalchemy-execute-raw-query FP: fixed columns and clause templates; all values are bound.
         return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()]