Add single-command LLM connection probe

src/viano_bug_analysis/connection_test.py

@@ -0,0 +1,42 @@
+"""One small LLM request, without creating or changing case/workspace files."""
+import argparse
+import asyncio
+import json
+
+from .llm import GatewayUnavailable, LLMError, PromptBuilder
+from .trial import preflight
+from .telemetry.safety import safe_code
+
+
+async def probe(gateway):
+    request = PromptBuilder().build(
+        task='Return exactly the JSON object {"answer": "ok"}.',
+        data={}, privacy="configured",
+        output_schema={"type": "object", "properties": {"answer": {"const": "ok"}},
+                       "required": ["answer"], "additionalProperties": False},
+    )
+    try:
+        result = await gateway.complete_round(request)
+        return {"status": "SUCCESS", "answer": result["document"], "metadata": result["metadata"]}
+    except GatewayUnavailable as error:
+        return {"status": "UNAVAILABLE", "attempts": error.attempts}
+    except LLMError as error:
+        return {"status": "ERROR", "code": safe_code(error.code), "http_status": error.http_status}
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--llm-config", default="config/llm.local.toml")
+    args = parser.parse_args(argv)
+    try:
+        result = asyncio.run(probe(preflight(args.llm_config)))
+    except KeyboardInterrupt:
+        result = {"status": "INTERRUPTED"}
+    except Exception as error:
+        result = {"status": "ERROR", "code": safe_code(getattr(error, "code", None))}
+    print(json.dumps(result))
+    return 0 if result["status"] == "SUCCESS" else 130 if result["status"] == "INTERRUPTED" else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_connection_test.py

@@ -0,0 +1,24 @@
+import json
+import pytest
+import viano_bug_analysis.connection_test as connection
+from viano_bug_analysis.llm import ProviderResponse, ProviderUnavailable
+from test_llm_gateway import Adapter, provider
+from viano_bug_analysis.llm import GatewayConfig, LLMGateway
+
+
+@pytest.mark.parametrize("response,status", [
+    (ProviderResponse({"answer": "ok"}, "fixture", 1, 1), "SUCCESS"),
+    (ProviderUnavailable("LLM_HTTP_TEMPORARY", http_status=503), "UNAVAILABLE"),
+])
+def test_one_round_no_workspace_files(tmp_path, monkeypatch, capsys, response, status):
+    adapter = Adapter({"first": response})
+    gateway = LLMGateway(GatewayConfig((provider(),)), {"fixture": adapter})
+    monkeypatch.setattr(connection, "preflight", lambda path: gateway)
+    monkeypatch.chdir(tmp_path)
+    exit_code = connection.main([])
+    result = json.loads(capsys.readouterr().out)
+    assert result["status"] == status
+    assert exit_code == (0 if status == "SUCCESS" else 2)
+    assert adapter.calls == ["first"] and not list(tmp_path.iterdir())
+    if status == "UNAVAILABLE":
+        assert result["attempts"][0]["http_status"] == 503