From 594f1061467fac6d29677cd9b91f9dce466e3ce3 Mon Sep 17 00:00:00 2001 From: codex Date: Wed, 20 May 2026 03:24:27 -0300 Subject: [PATCH] triage: split local diagnosis module --- ariadne/services/testing_triage.py | 240 +----------------- ariadne/services/testing_triage_diagnosis.py | 253 +++++++++++++++++++ tests/test_testing_triage.py | 121 --------- tests/test_testing_triage_diagnosis.py | 152 +++++++++++ 4 files changed, 412 insertions(+), 354 deletions(-) create mode 100644 ariadne/services/testing_triage_diagnosis.py create mode 100644 tests/test_testing_triage_diagnosis.py diff --git a/ariadne/services/testing_triage.py b/ariadne/services/testing_triage.py index 904defe..d8f9345 100644 --- a/ariadne/services/testing_triage.py +++ b/ariadne/services/testing_triage.py @@ -11,12 +11,17 @@ from ..db.storage import Storage from ..settings import settings from ..utils.logging import get_logger from .cluster_state import collect_cluster_state +from .testing_triage_diagnosis import ( + TRIAGE_DIAGNOSIS_EVENT_TYPE, + diagnose_testing_triage, + latest_testing_triage_diagnosis, # noqa: F401 - re-exported for app routes. + model_diagnosis_enabled, +) logger = get_logger(__name__) TRIAGE_EVENT_TYPE = "testing_triage_bundle" -TRIAGE_DIAGNOSIS_EVENT_TYPE = "testing_triage_diagnosis" _SUCCESS_STATUS = "ok|passed|success|not_applicable|skipped|na|n/a" _JENKINS_TREE = ( "jobs[name,url,color,lastBuild[number,result,timestamp,duration,url]," @@ -26,15 +31,6 @@ _JENKINS_TREE = ( _MAX_JENKINS_LOG_LINES = 80 _MAX_JENKINS_LOG_CHARS = 12000 _MAX_EVIDENCE_ITEMS = 12 -_MAX_MODEL_EVIDENCE_CHARS = 24000 -_MAX_MODEL_OUTPUT_CHARS = 12000 -_DIAGNOSIS_SYSTEM_PROMPT = ( - "You are Ariadne's local testing triage model. Use only the supplied JSON evidence. " - "Return JSON only with keys: headline, root_cause, blast_radius, confidence, " - "needs_human, next_actions, evidence_refs. Confidence must be low, medium, or high. " - "Next actions must be read-only verification or Flux/IaC changes; never suggest " - "mutating kubectl commands or reading Kubernetes Secret values." -) @dataclass(frozen=True) @@ -56,12 +52,6 @@ def latest_testing_triage_bundle(storage: Storage) -> dict[str, Any] | None: return _latest_event_payload(storage, TRIAGE_EVENT_TYPE) -def latest_testing_triage_diagnosis(storage: Storage) -> dict[str, Any] | None: - """Return the most recent stored local-model testing diagnosis, if present.""" - - return _latest_event_payload(storage, TRIAGE_DIAGNOSIS_EVENT_TYPE) - - def _latest_event_payload(storage: Storage, event_type: str) -> dict[str, Any] | None: rows = storage.list_events(limit=1, event_type=event_type) if not rows: @@ -83,7 +73,7 @@ def run_testing_triage(storage: Storage) -> TestingTriageSummary: bundle = collect_testing_triage(storage) storage.record_event(TRIAGE_EVENT_TYPE, bundle) - if _model_diagnosis_enabled(): + if model_diagnosis_enabled(): diagnosis = diagnose_testing_triage(bundle) storage.record_event(TRIAGE_DIAGNOSIS_EVENT_TYPE, diagnosis) summary = bundle.get("summary") if isinstance(bundle.get("summary"), dict) else {} @@ -162,222 +152,6 @@ def collect_testing_triage(storage: Storage | None = None) -> dict[str, Any]: return bundle -def diagnose_testing_triage(bundle: dict[str, Any]) -> dict[str, Any]: - """Ask the configured local model to summarize a deterministic triage bundle.""" - - base_url = _model_url() - if not base_url: - return _diagnosis_unavailable(bundle, "model_url_not_configured") - - try: - with httpx.Client(timeout=_model_timeout()) as client: - response = client.post( - f"{base_url}/api/generate", - json={ - "model": _model_name(), - "system": _DIAGNOSIS_SYSTEM_PROMPT, - "prompt": _diagnosis_prompt(bundle), - "stream": False, - "format": "json", - "options": { - "temperature": 0.1, - "top_p": 0.9, - }, - }, - ) - response.raise_for_status() - payload = response.json() - except Exception as exc: - return _diagnosis_unavailable(bundle, f"model_request_failed: {exc}") - - raw = str(payload.get("response") or "") - parsed, parse_error = _parse_model_response(raw) - return _diagnosis_from_model(bundle, parsed, raw, parse_error) - - -def _model_diagnosis_enabled() -> bool: - return bool(_model_url()) - - -def _model_url() -> str: - return str(getattr(settings, "testing_triage_model_url", "") or "").strip().rstrip("/") - - -def _model_name() -> str: - return str(getattr(settings, "testing_triage_model", "qwen2.5:7b-instruct-q4_0") or "").strip() - - -def _model_timeout() -> float: - return float(getattr(settings, "testing_triage_model_timeout_sec", 180.0) or 180.0) - - -def _diagnosis_prompt(bundle: dict[str, Any]) -> str: - payload = { - "task": "Summarize testing and cluster evidence for tonight's debugging work.", - "required_output_schema": { - "headline": "one sentence", - "root_cause": "most likely cause, or say evidence is insufficient", - "blast_radius": "affected suites, namespaces, pods, nodes, or services", - "confidence": "low|medium|high", - "needs_human": True, - "next_actions": ["short, concrete actions"], - "evidence_refs": ["specific evidence keys or values used"], - }, - "bundle": _model_evidence_payload(bundle), - } - evidence = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) - if len(evidence) > _MAX_MODEL_EVIDENCE_CHARS: - evidence = evidence[:_MAX_MODEL_EVIDENCE_CHARS] + "\n[truncated]" - return evidence - - -def _model_evidence_payload(bundle: dict[str, Any]) -> dict[str, Any]: - summary = bundle.get("summary") if isinstance(bundle.get("summary"), dict) else {} - evidence = bundle.get("evidence") if isinstance(bundle.get("evidence"), dict) else {} - return { - "kind": bundle.get("kind"), - "generated_at": bundle.get("generated_at"), - "summary": summary, - "evidence": evidence, - "unknowns": bundle.get("unknowns") if isinstance(bundle.get("unknowns"), list) else [], - } - - -def _parse_model_response(raw: str) -> tuple[dict[str, Any], str | None]: - if not raw.strip(): - return {}, "empty_model_response" - try: - parsed = json.loads(raw) - except json.JSONDecodeError as exc: - return {}, f"model_json_parse_failed: {exc}" - return (parsed if isinstance(parsed, dict) else {}, None) - - -def _diagnosis_from_model( - bundle: dict[str, Any], - parsed: dict[str, Any], - raw: str, - parse_error: str | None, -) -> dict[str, Any]: - summary = bundle.get("summary") if isinstance(bundle.get("summary"), dict) else {} - unknowns = list(bundle.get("unknowns") or []) if isinstance(bundle.get("unknowns"), list) else [] - if parse_error: - unknowns.append(parse_error) - diagnosis = { - "headline": _text_value(parsed.get("headline"), "Testing triage needs review."), - "root_cause": _text_value(parsed.get("root_cause"), "Evidence is insufficient for a confident root cause."), - "blast_radius": _text_value(parsed.get("blast_radius"), _blast_radius_fallback(summary)), - "confidence": _confidence(parsed.get("confidence")), - "needs_human": _bool_value(parsed.get("needs_human"), bool(summary.get("problem_count"))), - "next_actions": _text_list(parsed.get("next_actions")) or _default_next_actions(summary), - "evidence_refs": _text_list(parsed.get("evidence_refs")) or _default_evidence_refs(summary), - } - return { - "kind": "testing_triage_diagnosis", - "generated_at": datetime.now(timezone.utc).isoformat(), - "evidence_generated_at": bundle.get("generated_at") or "", - "evidence_summary": summary, - "status": "needs_attention" if diagnosis["needs_human"] or summary.get("problem_count") else "ok", - "model": _model_name(), - "source": "local_ollama", - "diagnosis": diagnosis, - "openclaw": { - "ariadne_latest_url": "/api/internal/testing/triage/diagnosis/latest", - "ariadne_run_url": "/api/internal/testing/triage/diagnosis/run", - "evidence_url": "/api/internal/testing/triage/latest", - }, - "unknowns": unknowns, - "raw_model": raw[:_MAX_MODEL_OUTPUT_CHARS], - } - - -def _diagnosis_unavailable(bundle: dict[str, Any], reason: str) -> dict[str, Any]: - summary = bundle.get("summary") if isinstance(bundle.get("summary"), dict) else {} - return { - "kind": "testing_triage_diagnosis", - "generated_at": datetime.now(timezone.utc).isoformat(), - "evidence_generated_at": bundle.get("generated_at") or "", - "evidence_summary": summary, - "status": "unavailable", - "model": _model_name(), - "source": "local_ollama", - "diagnosis": { - "headline": "Local model diagnosis is unavailable.", - "root_cause": reason, - "blast_radius": _blast_radius_fallback(summary), - "confidence": "low", - "needs_human": True, - "next_actions": [ - "Use the stored evidence bundle for manual triage.", - "Verify the Ariadne testing triage model URL and OpenClaw Ollama service health.", - ], - "evidence_refs": _default_evidence_refs(summary), - }, - "openclaw": { - "ariadne_latest_url": "/api/internal/testing/triage/diagnosis/latest", - "ariadne_run_url": "/api/internal/testing/triage/diagnosis/run", - "evidence_url": "/api/internal/testing/triage/latest", - }, - "unknowns": [reason], - "raw_model": "", - } - - -def _text_value(value: Any, default: str) -> str: - if isinstance(value, str) and value.strip(): - return value.strip() - return default - - -def _text_list(value: Any) -> list[str]: - if isinstance(value, str) and value.strip(): - return [value.strip()] - if not isinstance(value, list): - return [] - return [str(item).strip() for item in value if str(item).strip()][:8] - - -def _confidence(value: Any) -> str: - confidence = str(value or "").strip().lower() - return confidence if confidence in {"low", "medium", "high"} else "low" - - -def _bool_value(value: Any, default: bool) -> bool: - if isinstance(value, bool): - return value - if isinstance(value, str): - lowered = value.strip().lower() - if lowered in {"true", "yes", "1"}: - return True - if lowered in {"false", "no", "0"}: - return False - return default - - -def _blast_radius_fallback(summary: dict[str, Any]) -> str: - failed_suites = summary.get("failed_suites") if isinstance(summary.get("failed_suites"), list) else [] - if failed_suites: - return ", ".join(str(item) for item in failed_suites[:6]) - return "No failed suite scope identified in the evidence bundle." - - -def _default_next_actions(summary: dict[str, Any]) -> list[str]: - if int(summary.get("problem_count") or 0) > 0: - return [ - "Review the evidence bundle sections with non-empty problem lists.", - "Check the named Jenkins build logs and Flux Kustomizations before changing manifests.", - ] - return ["No action required unless a fresh bundle changes the status."] - - -def _default_evidence_refs(summary: dict[str, Any]) -> list[str]: - refs = [f"summary.status={summary.get('status')}", f"summary.problem_count={summary.get('problem_count')}"] - failed_suites = summary.get("failed_suites") if isinstance(summary.get("failed_suites"), list) else [] - if failed_suites: - refs.append("summary.failed_suites=" + ",".join(str(item) for item in failed_suites[:6])) - return refs - - def _latest_cluster_snapshot(storage: Storage | None, errors: list[str]) -> dict[str, Any]: if storage is not None: try: diff --git a/ariadne/services/testing_triage_diagnosis.py b/ariadne/services/testing_triage_diagnosis.py new file mode 100644 index 0000000..e4657f6 --- /dev/null +++ b/ariadne/services/testing_triage_diagnosis.py @@ -0,0 +1,253 @@ +from __future__ import annotations + +from datetime import datetime, timezone +import json +from typing import Any + +import httpx + +from ..db.storage import Storage +from ..settings import settings + + +TRIAGE_DIAGNOSIS_EVENT_TYPE = "testing_triage_diagnosis" +_MAX_MODEL_EVIDENCE_CHARS = 24000 +_MAX_MODEL_OUTPUT_CHARS = 12000 +_DIAGNOSIS_SYSTEM_PROMPT = ( + "You are Ariadne's local testing triage model. Use only the supplied JSON evidence. " + "Return JSON only with keys: headline, root_cause, blast_radius, confidence, " + "needs_human, next_actions, evidence_refs. Confidence must be low, medium, or high. " + "Next actions must be read-only verification or Flux/IaC changes; never suggest " + "mutating kubectl commands or reading Kubernetes Secret values." +) + + +def latest_testing_triage_diagnosis(storage: Storage) -> dict[str, Any] | None: + """Return the most recent stored local-model testing diagnosis, if present.""" + + rows = storage.list_events(limit=1, event_type=TRIAGE_DIAGNOSIS_EVENT_TYPE) + if not rows: + return None + detail = rows[0].get("detail") + if isinstance(detail, dict): + return detail + if isinstance(detail, str): + try: + payload = json.loads(detail) + except json.JSONDecodeError: + return None + return payload if isinstance(payload, dict) else None + return None + + +def model_diagnosis_enabled() -> bool: + return bool(_model_url()) + + +def diagnose_testing_triage(bundle: dict[str, Any]) -> dict[str, Any]: + """Ask the configured local model to summarize a deterministic triage bundle.""" + + base_url = _model_url() + if not base_url: + return _diagnosis_unavailable(bundle, "model_url_not_configured") + + try: + with httpx.Client(timeout=_model_timeout()) as client: + response = client.post( + f"{base_url}/api/generate", + json={ + "model": _model_name(), + "system": _DIAGNOSIS_SYSTEM_PROMPT, + "prompt": _diagnosis_prompt(bundle), + "stream": False, + "format": "json", + "options": {"temperature": 0.1, "top_p": 0.9}, + }, + ) + response.raise_for_status() + payload = response.json() + except Exception as exc: + return _diagnosis_unavailable(bundle, f"model_request_failed: {exc}") + + raw = str(payload.get("response") or "") + parsed, parse_error = _parse_model_response(raw) + return _diagnosis_from_model(bundle, parsed, raw, parse_error) + + +def _model_url() -> str: + return str(getattr(settings, "testing_triage_model_url", "") or "").strip().rstrip("/") + + +def _model_name() -> str: + return str(getattr(settings, "testing_triage_model", "qwen2.5:7b-instruct-q4_0") or "").strip() + + +def _model_timeout() -> float: + return float(getattr(settings, "testing_triage_model_timeout_sec", 180.0) or 180.0) + + +def _diagnosis_prompt(bundle: dict[str, Any]) -> str: + payload = { + "task": "Summarize testing and cluster evidence for tonight's debugging work.", + "required_output_schema": { + "headline": "one sentence", + "root_cause": "most likely cause, or say evidence is insufficient", + "blast_radius": "affected suites, namespaces, pods, nodes, or services", + "confidence": "low|medium|high", + "needs_human": True, + "next_actions": ["short, concrete actions"], + "evidence_refs": ["specific evidence keys or values used"], + }, + "bundle": _model_evidence_payload(bundle), + } + evidence = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + if len(evidence) > _MAX_MODEL_EVIDENCE_CHARS: + evidence = evidence[:_MAX_MODEL_EVIDENCE_CHARS] + "\n[truncated]" + return evidence + + +def _model_evidence_payload(bundle: dict[str, Any]) -> dict[str, Any]: + summary = bundle.get("summary") if isinstance(bundle.get("summary"), dict) else {} + evidence = bundle.get("evidence") if isinstance(bundle.get("evidence"), dict) else {} + return { + "kind": bundle.get("kind"), + "generated_at": bundle.get("generated_at"), + "summary": summary, + "evidence": evidence, + "unknowns": bundle.get("unknowns") if isinstance(bundle.get("unknowns"), list) else [], + } + + +def _parse_model_response(raw: str) -> tuple[dict[str, Any], str | None]: + if not raw.strip(): + return {}, "empty_model_response" + try: + parsed = json.loads(raw) + except json.JSONDecodeError as exc: + return {}, f"model_json_parse_failed: {exc}" + return (parsed if isinstance(parsed, dict) else {}, None) + + +def _diagnosis_from_model( + bundle: dict[str, Any], + parsed: dict[str, Any], + raw: str, + parse_error: str | None, +) -> dict[str, Any]: + summary = bundle.get("summary") if isinstance(bundle.get("summary"), dict) else {} + unknowns = list(bundle.get("unknowns") or []) if isinstance(bundle.get("unknowns"), list) else [] + if parse_error: + unknowns.append(parse_error) + diagnosis = { + "headline": _text_value(parsed.get("headline"), "Testing triage needs review."), + "root_cause": _text_value(parsed.get("root_cause"), "Evidence is insufficient for a confident root cause."), + "blast_radius": _text_value(parsed.get("blast_radius"), _blast_radius_fallback(summary)), + "confidence": _confidence(parsed.get("confidence")), + "needs_human": _bool_value(parsed.get("needs_human"), bool(summary.get("problem_count"))), + "next_actions": _text_list(parsed.get("next_actions")) or _default_next_actions(summary), + "evidence_refs": _text_list(parsed.get("evidence_refs")) or _default_evidence_refs(summary), + } + return { + "kind": "testing_triage_diagnosis", + "generated_at": datetime.now(timezone.utc).isoformat(), + "evidence_generated_at": bundle.get("generated_at") or "", + "evidence_summary": summary, + "status": "needs_attention" if diagnosis["needs_human"] or summary.get("problem_count") else "ok", + "model": _model_name(), + "source": "local_ollama", + "diagnosis": diagnosis, + "openclaw": { + "ariadne_latest_url": "/api/internal/testing/triage/diagnosis/latest", + "ariadne_run_url": "/api/internal/testing/triage/diagnosis/run", + "evidence_url": "/api/internal/testing/triage/latest", + }, + "unknowns": unknowns, + "raw_model": raw[:_MAX_MODEL_OUTPUT_CHARS], + } + + +def _diagnosis_unavailable(bundle: dict[str, Any], reason: str) -> dict[str, Any]: + summary = bundle.get("summary") if isinstance(bundle.get("summary"), dict) else {} + return { + "kind": "testing_triage_diagnosis", + "generated_at": datetime.now(timezone.utc).isoformat(), + "evidence_generated_at": bundle.get("generated_at") or "", + "evidence_summary": summary, + "status": "unavailable", + "model": _model_name(), + "source": "local_ollama", + "diagnosis": { + "headline": "Local model diagnosis is unavailable.", + "root_cause": reason, + "blast_radius": _blast_radius_fallback(summary), + "confidence": "low", + "needs_human": True, + "next_actions": [ + "Use the stored evidence bundle for manual triage.", + "Verify the Ariadne testing triage model URL and OpenClaw Ollama service health.", + ], + "evidence_refs": _default_evidence_refs(summary), + }, + "openclaw": { + "ariadne_latest_url": "/api/internal/testing/triage/diagnosis/latest", + "ariadne_run_url": "/api/internal/testing/triage/diagnosis/run", + "evidence_url": "/api/internal/testing/triage/latest", + }, + "unknowns": [reason], + "raw_model": "", + } + + +def _text_value(value: Any, default: str) -> str: + if isinstance(value, str) and value.strip(): + return value.strip() + return default + + +def _text_list(value: Any) -> list[str]: + if isinstance(value, str) and value.strip(): + return [value.strip()] + if not isinstance(value, list): + return [] + return [str(item).strip() for item in value if str(item).strip()][:8] + + +def _confidence(value: Any) -> str: + confidence = str(value or "").strip().lower() + return confidence if confidence in {"low", "medium", "high"} else "low" + + +def _bool_value(value: Any, default: bool) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, str): + lowered = value.strip().lower() + if lowered in {"true", "yes", "1"}: + return True + if lowered in {"false", "no", "0"}: + return False + return default + + +def _blast_radius_fallback(summary: dict[str, Any]) -> str: + failed_suites = summary.get("failed_suites") if isinstance(summary.get("failed_suites"), list) else [] + if failed_suites: + return ", ".join(str(item) for item in failed_suites[:6]) + return "No failed suite scope identified in the evidence bundle." + + +def _default_next_actions(summary: dict[str, Any]) -> list[str]: + if int(summary.get("problem_count") or 0) > 0: + return [ + "Review the evidence bundle sections with non-empty problem lists.", + "Check the named Jenkins build logs and Flux Kustomizations before changing manifests.", + ] + return ["No action required unless a fresh bundle changes the status."] + + +def _default_evidence_refs(summary: dict[str, Any]) -> list[str]: + refs = [f"summary.status={summary.get('status')}", f"summary.problem_count={summary.get('problem_count')}"] + failed_suites = summary.get("failed_suites") if isinstance(summary.get("failed_suites"), list) else [] + if failed_suites: + refs.append("summary.failed_suites=" + ",".join(str(item) for item in failed_suites[:6])) + return refs diff --git a/tests/test_testing_triage.py b/tests/test_testing_triage.py index 4ab67dd..fe75800 100644 --- a/tests/test_testing_triage.py +++ b/tests/test_testing_triage.py @@ -91,26 +91,6 @@ def test_run_testing_triage_stores_latest(monkeypatch) -> None: assert latest["summary"]["status"] == "ok" -def test_run_testing_triage_stores_diagnosis_when_model_enabled(monkeypatch) -> None: - storage = DummyStorage() - bundle = {"summary": {"status": "needs_attention", "problem_count": 1, "failed_suites": ["ariadne"]}} - diagnosis = {"kind": "testing_triage_diagnosis", "status": "needs_attention"} - - monkeypatch.setattr(testing_triage, "settings", SettingsStub(testing_triage_model_url="http://ollama")) - monkeypatch.setattr(testing_triage, "collect_testing_triage", lambda _storage: bundle) - monkeypatch.setattr(testing_triage, "diagnose_testing_triage", lambda _bundle: diagnosis) - - summary = testing_triage.run_testing_triage(storage) - latest = testing_triage.latest_testing_triage_diagnosis(storage) - - assert summary.status == "needs_attention" - assert [event[0] for event in storage.events] == [ - testing_triage.TRIAGE_EVENT_TYPE, - testing_triage.TRIAGE_DIAGNOSIS_EVENT_TYPE, - ] - assert latest == diagnosis - - def test_latest_testing_triage_bundle_handles_json_strings() -> None: class JsonStorage: def list_events(self, limit: int = 1, event_type: str | None = None): # type: ignore[no-untyped-def] @@ -142,107 +122,6 @@ def test_latest_testing_triage_bundle_ignores_bad_payloads() -> None: assert testing_triage.latest_testing_triage_bundle(EmptyStorage()) is None # type: ignore[arg-type] -def test_diagnose_testing_triage_calls_local_ollama(monkeypatch) -> None: - captured = {} - model_response = json.dumps( - { - "headline": "Ariadne has one failing suite.", - "root_cause": "Jenkins reported a failed ariadne run.", - "blast_radius": "ariadne", - "confidence": "medium", - "needs_human": True, - "next_actions": ["Inspect the failed Jenkins build log."], - "evidence_refs": ["summary.failed_suites=ariadne"], - } - ) - - class FakeResponse: - def raise_for_status(self) -> None: - return None - - def json(self): # type: ignore[no-untyped-def] - return {"response": model_response} - - class FakeClient: - def __init__(self, *, timeout) -> None: # type: ignore[no-untyped-def] - captured["timeout"] = timeout - - def __enter__(self): - return self - - def __exit__(self, *args) -> None: # type: ignore[no-untyped-def] - return None - - def post(self, url, json=None): # type: ignore[no-untyped-def] - captured["url"] = url - captured["request"] = json - return FakeResponse() - - monkeypatch.setattr( - testing_triage, - "settings", - SettingsStub( - testing_triage_model_url="http://ollama/", - testing_triage_model="tiny-model", - testing_triage_model_timeout_sec=3.0, - ), - ) - monkeypatch.setattr(testing_triage.httpx, "Client", FakeClient) - - diagnosis = testing_triage.diagnose_testing_triage( - { - "kind": "testing_triage_bundle", - "generated_at": "now", - "summary": {"status": "needs_attention", "problem_count": 1, "failed_suites": ["ariadne"]}, - "evidence": {"jenkins": {"failed_builds": [{"job": "ariadne"}]}}, - "unknowns": [], - } - ) - - assert captured["url"] == "http://ollama/api/generate" - assert captured["timeout"] == 3.0 - assert captured["request"]["model"] == "tiny-model" - assert captured["request"]["format"] == "json" - assert diagnosis["kind"] == "testing_triage_diagnosis" - assert diagnosis["status"] == "needs_attention" - assert diagnosis["diagnosis"]["confidence"] == "medium" - assert diagnosis["diagnosis"]["next_actions"] == ["Inspect the failed Jenkins build log."] - - -def test_diagnose_testing_triage_handles_disabled_and_bad_json(monkeypatch) -> None: - monkeypatch.setattr(testing_triage, "settings", SettingsStub(testing_triage_model_url="")) - disabled = testing_triage.diagnose_testing_triage({"summary": {"status": "ok", "problem_count": 0}}) - assert disabled["status"] == "unavailable" - assert disabled["diagnosis"]["root_cause"] == "model_url_not_configured" - - class BadResponse: - def raise_for_status(self) -> None: - return None - - def json(self): # type: ignore[no-untyped-def] - return {"response": "not json"} - - class BadClient: - def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] - return None - - def __enter__(self): - return self - - def __exit__(self, *args) -> None: # type: ignore[no-untyped-def] - return None - - def post(self, url, json=None): # type: ignore[no-untyped-def] - return BadResponse() - - monkeypatch.setattr(testing_triage, "settings", SettingsStub(testing_triage_model_url="http://ollama")) - monkeypatch.setattr(testing_triage.httpx, "Client", BadClient) - diagnosis = testing_triage.diagnose_testing_triage({"summary": {"status": "ok", "problem_count": 0}}) - - assert diagnosis["status"] == "ok" - assert "model_json_parse_failed" in diagnosis["unknowns"][0] - - def test_latest_cluster_snapshot_falls_back_to_live_collect(monkeypatch) -> None: class BrokenStorage: def latest_cluster_state(self): # type: ignore[no-untyped-def] diff --git a/tests/test_testing_triage_diagnosis.py b/tests/test_testing_triage_diagnosis.py new file mode 100644 index 0000000..a63dc97 --- /dev/null +++ b/tests/test_testing_triage_diagnosis.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +import json + +from ariadne.services import testing_triage +from ariadne.services import testing_triage_diagnosis + + +class DummyStorage: + def __init__(self) -> None: + self.events: list[tuple[str, dict]] = [] + + def record_event(self, event_type: str, detail: dict) -> None: + self.events.append((event_type, detail)) + + def list_events(self, limit: int = 1, event_type: str | None = None): # type: ignore[no-untyped-def] + matching = [ + {"detail": detail} + for stored_type, detail in self.events + if event_type is None or stored_type == event_type + ] + return matching[-limit:][::-1] + + +class SettingsStub: + def __init__(self, **overrides) -> None: # type: ignore[no-untyped-def] + self.testing_triage_model_url = "" + self.testing_triage_model = "qwen2.5:7b-instruct-q4_0" + self.testing_triage_model_timeout_sec = 1.0 + for key, value in overrides.items(): + setattr(self, key, value) + + +def test_run_testing_triage_stores_diagnosis_when_model_enabled(monkeypatch) -> None: + storage = DummyStorage() + bundle = {"summary": {"status": "needs_attention", "problem_count": 1, "failed_suites": ["ariadne"]}} + diagnosis = {"kind": "testing_triage_diagnosis", "status": "needs_attention"} + + monkeypatch.setattr(testing_triage, "model_diagnosis_enabled", lambda: True) + monkeypatch.setattr(testing_triage, "collect_testing_triage", lambda _storage: bundle) + monkeypatch.setattr(testing_triage, "diagnose_testing_triage", lambda _bundle: diagnosis) + + summary = testing_triage.run_testing_triage(storage) # type: ignore[arg-type] + latest = testing_triage.latest_testing_triage_diagnosis(storage) # type: ignore[arg-type] + + assert summary.status == "needs_attention" + assert [event[0] for event in storage.events] == [ + testing_triage.TRIAGE_EVENT_TYPE, + testing_triage.TRIAGE_DIAGNOSIS_EVENT_TYPE, + ] + assert latest == diagnosis + + +def test_diagnose_testing_triage_calls_local_ollama(monkeypatch) -> None: + captured = {} + model_response = json.dumps( + { + "headline": "Ariadne has one failing suite.", + "root_cause": "Jenkins reported a failed ariadne run.", + "blast_radius": "ariadne", + "confidence": "medium", + "needs_human": True, + "next_actions": ["Inspect the failed Jenkins build log."], + "evidence_refs": ["summary.failed_suites=ariadne"], + } + ) + + class FakeResponse: + def raise_for_status(self) -> None: + return None + + def json(self): # type: ignore[no-untyped-def] + return {"response": model_response} + + class FakeClient: + def __init__(self, *, timeout) -> None: # type: ignore[no-untyped-def] + captured["timeout"] = timeout + + def __enter__(self): + return self + + def __exit__(self, *args) -> None: # type: ignore[no-untyped-def] + return None + + def post(self, url, json=None): # type: ignore[no-untyped-def] + captured["url"] = url + captured["request"] = json + return FakeResponse() + + monkeypatch.setattr( + testing_triage_diagnosis, + "settings", + SettingsStub( + testing_triage_model_url="http://ollama/", + testing_triage_model="tiny-model", + testing_triage_model_timeout_sec=3.0, + ), + ) + monkeypatch.setattr(testing_triage_diagnosis.httpx, "Client", FakeClient) + + diagnosis = testing_triage_diagnosis.diagnose_testing_triage( + { + "kind": "testing_triage_bundle", + "generated_at": "now", + "summary": {"status": "needs_attention", "problem_count": 1, "failed_suites": ["ariadne"]}, + "evidence": {"jenkins": {"failed_builds": [{"job": "ariadne"}]}}, + "unknowns": [], + } + ) + + assert captured["url"] == "http://ollama/api/generate" + assert captured["timeout"] == 3.0 + assert captured["request"]["model"] == "tiny-model" + assert captured["request"]["format"] == "json" + assert diagnosis["kind"] == "testing_triage_diagnosis" + assert diagnosis["status"] == "needs_attention" + assert diagnosis["diagnosis"]["confidence"] == "medium" + assert diagnosis["diagnosis"]["next_actions"] == ["Inspect the failed Jenkins build log."] + + +def test_diagnose_testing_triage_handles_disabled_and_bad_json(monkeypatch) -> None: + monkeypatch.setattr(testing_triage_diagnosis, "settings", SettingsStub(testing_triage_model_url="")) + disabled = testing_triage_diagnosis.diagnose_testing_triage({"summary": {"status": "ok", "problem_count": 0}}) + assert disabled["status"] == "unavailable" + assert disabled["diagnosis"]["root_cause"] == "model_url_not_configured" + + class BadResponse: + def raise_for_status(self) -> None: + return None + + def json(self): # type: ignore[no-untyped-def] + return {"response": "not json"} + + class BadClient: + def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] + return None + + def __enter__(self): + return self + + def __exit__(self, *args) -> None: # type: ignore[no-untyped-def] + return None + + def post(self, url, json=None): # type: ignore[no-untyped-def] + return BadResponse() + + monkeypatch.setattr(testing_triage_diagnosis, "settings", SettingsStub(testing_triage_model_url="http://ollama")) + monkeypatch.setattr(testing_triage_diagnosis.httpx, "Client", BadClient) + diagnosis = testing_triage_diagnosis.diagnose_testing_triage({"summary": {"status": "ok", "problem_count": 0}}) + + assert diagnosis["status"] == "ok" + assert "model_json_parse_failed" in diagnosis["unknowns"][0]