import json from pathlib import Path import pytest from investibook.core import workspace_evidence_store as store def _write_report(root: Path, workspace_id: str, *, message_id: str | None = None) -> Path: report_dir = root / "workspaces" / workspace_id / "evidences" report_dir.mkdir(parents=True) metadata = { "workspace_id": workspace_id, "cluster_id": "cluster-1", "evidence_dir": str(report_dir), } if message_id is not None: metadata["last_raw_evidence"] = { "source": "clinic-cli:metrics", "kind": "metric", "message_id": message_id, } else: metadata["last_raw_evidence"] = { "source": "clinic-cli:metrics", "kind": "metric", } report_path = report_dir / "workspace_evidence_report.json" report_path.write_text( json.dumps( { "type": "workspace_evidence_report", "metadata": metadata, "analysis": {}, "evidence": { "metrics": [ { "type": "metric", "name": "TiDB QPS", "metric_group_id": "tidb_qps", } ], "queries": [], "logs": [], }, } ), encoding="utf-8", ) return report_path def _load_json(path: Path) -> dict: return json.loads(path.read_text(encoding="utf-8")) def test_update_workspace_evidence_analysis_merges_patch_and_updates_state(tmp_path, monkeypatch): monkeypatch.setattr(store, "WORKSPACES_DIR", tmp_path / "workspaces") report_path = _write_report(tmp_path, "ws-4", message_id="assistant-1") result = store.update_workspace_evidence_analysis( "ws-4", { "severity": "Low", "add_key_evidence": [ { "type": "metric", "name": "TiDB QPS", "metric_ref": {"metric_group_id": "tidb_qps"}, "reason": "QPS around the incident window stays within the expected band.", } ], }, cluster_id="cluster-4", source="clinic-cli/post-pass", message_id="assistant-1", ) report = _load_json(report_path) assert result["status"] == "ok" assert report["analysis"]["severity"] == "Low" assert report["analysis"]["key_evidence"] == [ { "type": "metric", "name": "TiDB QPS", "metric_ref": {"metric_group_id": "tidb_qps"}, "reason": "QPS around the incident window stays within the expected band.", } ] assert report["metadata"]["last_raw_evidence"]["message_id"] == "assistant-1" assert report["metadata"]["last_analysis_update"] == { "at": result["last_analysis_update"]["at"], "message_id": "assistant-1", "source": "clinic-cli/post-pass", "kind": "analysis", } patch_file = _load_json(Path(result["analysis_patch_path"])) assert patch_file["severity"] == "Low" def test_update_workspace_evidence_analysis_requires_existing_report(tmp_path, monkeypatch): monkeypatch.setattr(store, "WORKSPACES_DIR", tmp_path / "workspaces") with pytest.raises(ValueError, match="workspace evidence report not found"): store.update_workspace_evidence_analysis( "missing", { "add_key_evidence": [ { "type": "metric", "name": "TiDB QPS", "reason": "QPS evidence exists.", } ] }, ) def test_get_workspace_evidence_postpass_state_detects_missing_analysis_for_same_message(tmp_path, monkeypatch): monkeypatch.setattr(store, "WORKSPACES_DIR", tmp_path / "workspaces") _write_report(tmp_path, "ws-5", message_id="assistant-5") state = store.get_workspace_evidence_postpass_state("ws-5", message_id="assistant-5") assert state["needs_postpass"] is True store.update_workspace_evidence_analysis( "ws-5", { "add_key_evidence": [ { "type": "metric", "name": "TiDB QPS", "metric_ref": {"metric_group_id": "tidb_qps"}, "reason": "QPS evidence was persisted for this turn.", } ] }, source="backend-postpass", message_id="assistant-5", ) state = store.get_workspace_evidence_postpass_state("ws-5", message_id="assistant-5") assert state["needs_postpass"] is False def test_get_workspace_evidence_postpass_state_allows_raw_evidence_without_message_id(tmp_path, monkeypatch): monkeypatch.setattr(store, "WORKSPACES_DIR", tmp_path / "workspaces") _write_report(tmp_path, "ws-no-message", message_id=None) state = store.get_workspace_evidence_postpass_state( "ws-no-message", message_id="assistant-no-message", ) assert state["needs_postpass"] is True store.update_workspace_evidence_analysis( "ws-no-message", { "add_key_evidence": [ { "type": "metric", "name": "TiDB QPS", "metric_ref": {"metric_group_id": "tidb_qps"}, "reason": "QPS evidence was persisted for this turn.", } ] }, source="backend-postpass", message_id="assistant-no-message", ) state = store.get_workspace_evidence_postpass_state( "ws-no-message", message_id="assistant-no-message", ) assert state["needs_postpass"] is False def test_write_standard_collector_handoff_builds_contract_file(tmp_path): evidence_dir = tmp_path / "evidences" evidence_dir.mkdir(parents=True) summary_path = evidence_dir / "evidence_summary_for_agent.json" summary_path.write_text( json.dumps( { "metadata": { "cluster_id": "cluster-fast", "cluster": { "cluster_id": "cluster-fast", "deployType": "premium", "parent_id": "parent-1", "keyspace_id": "ks-9", }, "time_range": { "start": "1775039760", "end": "1775043360", "start_iso": "2026-04-01T10:36:00Z", "end_iso": "2026-04-01T11:36:00Z", }, }, "errors": ["slow query returned empty rows"], "critical_evidence_hints": [ {"id": "primary_symptom_evidence_present", "summary": "CPU evidence exists."} ], } ), encoding="utf-8", ) (evidence_dir / "workspace_evidence_report.json").write_text( json.dumps({"type": "workspace_evidence_report", "metadata": {}, "analysis": {}}), encoding="utf-8", ) (evidence_dir / "metrics.json").write_text(json.dumps({"metrics": []}), encoding="utf-8") (evidence_dir / "collection_errors.json").write_text( json.dumps({"errors": []}), encoding="utf-8", ) (evidence_dir / "evidence_summary_for_agent.txt").write_text("summary", encoding="utf-8") result = store.write_standard_collector_handoff( evidence_dir=evidence_dir, summary_path=summary_path, ) payload = _load_json(evidence_dir / "agent-runs" / "evidence-agent.json") assert result["status"] == "ok" assert payload["agent"] == "evidence-agent" assert payload["status"] == "ok" assert payload["scope"] == { "cluster_id": "cluster-fast", "time_range": { "start": "1775039760", "end": "1775043360", "start_iso": "2026-04-01T10:36:00Z", "end_iso": "2026-04-01T11:36:00Z", }, "tier": "premium", "parent_id": "parent-1", "keyspace_id": "ks-9", } assert payload["warnings"] == ["slow query returned empty rows"] assert payload["critical_evidence_hints"] == [ {"id": "primary_symptom_evidence_present", "summary": "CPU evidence exists."} ] assert payload["evidence_files"] == [ "evidences/collection_errors.json", "evidences/evidence_summary_for_agent.json", "evidences/evidence_summary_for_agent.txt", "evidences/metrics.json", "evidences/workspace_evidence_report.json", ] def test_write_standard_collector_handoff_requires_all_root_artifacts(tmp_path): evidence_dir = tmp_path / "evidences" evidence_dir.mkdir(parents=True) (evidence_dir / "evidence_summary_for_agent.json").write_text( json.dumps({"metadata": {"cluster_id": "cluster-fast"}}), encoding="utf-8", ) (evidence_dir / "workspace_evidence_report.json").write_text( json.dumps({"type": "workspace_evidence_report", "metadata": {}, "analysis": {}}), encoding="utf-8", ) with pytest.raises(ValueError, match="collector artifacts missing"): store.write_standard_collector_handoff(evidence_dir=evidence_dir)