#!/usr/bin/env python3 """Dev functional test for workspace evidence file-first behavior. This FT drives Investibook through HTTP APIs only for product actions: workspace creation, Jira source preparation, chat, and file reads. kubectl is used only for final assertions against the dev backend pod filesystem. Required: INVESTIBOOK_DEV_TOKEN or INVESTIBOOK_API_KEY Example: INVESTIBOOK_DEV_TOKEN=ib_xxx \ python3 tests/ft/test_workspace_evidence_dev_api_ft.py \ --base-url https://dev.investibook.com Image attachment case: INVESTIBOOK_DEV_TOKEN=ib_xxx \ python3 tests/ft/test_workspace_evidence_dev_api_ft.py \ --case image \ --image-path /path/to/tikv-cpu.png """ from __future__ import annotations import argparse import json import mimetypes import os import socket import subprocess import sys import time import uuid from dataclasses import dataclass from typing import Any from urllib.error import HTTPError, URLError from urllib.parse import urlencode from urllib.request import Request, urlopen DEFAULT_DIRECT_PROMPT = ( "O11Y-2447801 查下 QPS 整体变化曲线。" ) DEFAULT_DIAGNOSIS_PROMPT = ( "查一下 10754652599988043412 集群 2026-04-05 12:00 UTC+8 前后 1 小时 TiKV 热点问题。" ) DEFAULT_IMAGE_PROMPT = ( "这是 10754652599988043412 集群的 TiKV CPU 截图。" "请结合图片摘要解释哪些 TiKV instance CPU 最高、时间段内曲线有什么变化，并说明不确定性。" ) @dataclass class ChatResult: events: list[dict[str, Any]] text: str def log(*args: object, **kwargs: object) -> None: print(*args, flush=True, **kwargs) class Api: def __init__(self, base_url: str, token: str, timeout_s: float) -> None: self.base_url = base_url.rstrip("/") self.token = token.strip() self.timeout_s = timeout_s def close(self) -> None: return None def headers(self) -> dict[str, str]: return {"Authorization": f"Bearer {self.token}"} def _request_json( self, method: str, path: str, *, body: dict[str, Any] | None = None, query: dict[str, Any] | None = None, ok_statuses: set[int] | None = None, ) -> tuple[int, Any]: status, text = self._request_text( method, path, body=body, query=query, ok_statuses=ok_statuses, ) return status, json.loads(text) if text else None def _request_text( self, method: str, path: str, *, body: dict[str, Any] | None = None, query: dict[str, Any] | None = None, ok_statuses: set[int] | None = None, ) -> tuple[int, str]: ok_statuses = ok_statuses or {200} url = f"{self.base_url}{path}" if query: url = f"{url}?{urlencode(query)}" data = None headers = self.headers() if body is not None: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" req = Request(url, data=data, headers=headers, method=method) attempts = 3 if method in {"GET", "POST"} else 1 last_exc: Exception | None = None for attempt in range(1, attempts + 1): try: with urlopen(req, timeout=self.timeout_s) as resp: status = int(resp.status) text = resp.read().decode("utf-8", errors="replace") break except HTTPError as exc: status = int(exc.code) text = exc.read().decode("utf-8", errors="replace") break except (TimeoutError, socket.timeout, URLError) as exc: last_exc = exc if attempt == attempts: raise log(f"[http] retry {attempt}/{attempts} {method} {path}: {exc}") time.sleep(float(attempt)) else: raise AssertionError(f"{method} {url} failed: {last_exc}") if status not in ok_statuses: raise AssertionError(f"{method} {url} failed status={status} body={text[:2000]}") return status, text def create_workspace(self, name: str) -> str: _, payload = self._request_json( "POST", "/api/workspaces", body={"name": name, "description": "Temporary dev API FT workspace", "is_example": False}, ) return str(payload["id"]) def delete_workspace(self, workspace_id: str) -> None: self._request_text( "DELETE", f"/api/workspaces/{workspace_id}", ok_statuses={200, 404}, ) def add_jira_source(self, workspace_id: str, issue_id: str) -> None: self._request_json( "POST", f"/api/workspace/{workspace_id}/sources", body={ "type": "jira_issue", "name": issue_id, "issueId": issue_id, "status": "loading", }, ) def prepare_source(self, workspace_id: str, issue_id: str, timeout_s: float) -> dict[str, Any]: _, payload = self._request_json( "POST", "/api/sources/prepare", body={"workspace_id": workspace_id, "issue_id": issue_id}, ) task_id = str(payload["task_id"]) deadline = time.monotonic() + timeout_s last: dict[str, Any] = {} while time.monotonic() < deadline: _, last = self._request_json( "GET", f"/api/sources/prepare/{task_id}", ) status = str(last.get("status") or "") log(f"[prepare] task={task_id} status={status}") if status == "completed": return last if status == "error": raise AssertionError(f"source prepare failed: {last}") time.sleep(2.0) raise TimeoutError(f"source prepare timed out: task={task_id} last={last}") def upload_image_attachment(self, workspace_id: str, image_path: str) -> dict[str, Any]: path = os.path.abspath(image_path) filename = os.path.basename(path) mime_type = mimetypes.guess_type(filename)[0] or "application/octet-stream" with open(path, "rb") as f: data = f.read() boundary = f"----investibook-ft-{uuid.uuid4().hex}" body = b"".join( [ f"--{boundary}\r\n".encode(), ( f'Content-Disposition: form-data; name="file"; filename="{filename}"\r\n' f"Content-Type: {mime_type}\r\n\r\n" ).encode(), data, b"\r\n", f"--{boundary}--\r\n".encode(), ] ) url = f"{self.base_url}/api/workspace/{workspace_id}/image-attachments" req = Request( url, data=body, headers={ **self.headers(), "Content-Type": f"multipart/form-data; boundary={boundary}", "Content-Length": str(len(body)), }, method="POST", ) try: with urlopen(req, timeout=self.timeout_s) as resp: payload = json.loads(resp.read().decode("utf-8", errors="replace")) except HTTPError as exc: text = exc.read().decode("utf-8", errors="replace") raise AssertionError(f"image upload failed status={exc.code} body={text[:2000]}") from exc return dict(payload["attachment"]) def image_attachments(self, workspace_id: str, attachment_ids: list[str]) -> list[dict[str, Any]]: _, payload = self._request_json( "GET", f"/api/workspace/{workspace_id}/image-attachments", query={"ids": ",".join(attachment_ids)}, ) return list(payload.get("attachments") or []) def wait_image_attachment_ready( self, workspace_id: str, attachment_id: str, timeout_s: float, ) -> dict[str, Any]: deadline = time.monotonic() + timeout_s last: dict[str, Any] = {} while time.monotonic() < deadline: items = self.image_attachments(workspace_id, [attachment_id]) if items: last = items[0] status = str(last.get("status") or "") log(f"[image] attachment={attachment_id} status={status} kind={last.get('kind') or ''}") if status == "ready": return last if status == "failed": raise AssertionError(f"image parse failed: {last}") time.sleep(2.0) raise TimeoutError(f"image parse timed out: attachment={attachment_id} last={last}") def stream_chat( self, workspace_id: str, message: str, timeout_s: float, *, attachment_ids: list[str] | None = None, ) -> ChatResult: events: list[dict[str, Any]] = [] text_chunks: list[str] = [] params = { "message": message, "workspace_id": workspace_id, "user_message_id": f"user-{int(time.time() * 1000)}-{uuid.uuid4().hex[:6]}", "token": self.token, } if attachment_ids: params["attachment_ids"] = ",".join(attachment_ids) url = f"{self.base_url}/api/chat/stream?{urlencode(params)}" req = Request( url, headers={ "Accept": "text/event-stream", "Cache-Control": "no-cache", }, method="GET", ) data_lines: list[str] = [] try: with urlopen(req, timeout=timeout_s) as resp: if int(resp.status) != 200: body = resp.read().decode("utf-8", errors="replace") raise AssertionError(f"chat stream failed status={resp.status} body={body[:2000]}") for raw_line in resp: line = raw_line.decode("utf-8", errors="replace").rstrip("\r\n") if not line: if data_lines: event = json.loads("\n".join(data_lines)) events.append(event) etype = event.get("type") if etype == "text": text_chunks.append(str(event.get("text") or "")) elif etype == "tool": log( "[chat] tool", event.get("status"), str(event.get("title") or "")[:120], ) elif etype == "error": raise AssertionError(f"chat error event: {event}") elif etype == "complete": log("[chat] complete") return ChatResult(events=events, text="".join(text_chunks)) data_lines = [] continue if line.startswith(":"): continue if line.startswith("data: "): data_lines.append(line[6:]) except HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") raise AssertionError(f"chat stream failed status={exc.code} body={body[:2000]}") from exc raise AssertionError(f"chat stream ended before complete event; events={len(events)}") def files(self, workspace_id: str) -> list[dict[str, Any]]: _, payload = self._request_json( "GET", f"/api/workspace/{workspace_id}/files", ) return list(payload["files"]) def file_text(self, workspace_id: str, path: str) -> str: _, text = self._request_text( "GET", f"/api/workspace/{workspace_id}/file", query={"path": path}, ) return text def messages(self, workspace_id: str) -> list[dict[str, Any]]: _, payload = self._request_json( "GET", f"/api/workspace/{workspace_id}/messages", query={"include_parts": "false"}, ) return list(payload.get("messages") or []) def flatten_file_paths(items: list[dict[str, Any]]) -> list[str]: out: list[str] = [] for item in items: path = str(item.get("path") or "") if item.get("type") == "file" and path: out.append(path) children = item.get("children") if isinstance(children, list): out.extend(flatten_file_paths(children)) return out def find_path(paths: list[str], suffix: str) -> str: matches = [path for path in paths if path.endswith(suffix)] if not matches: raise AssertionError(f"missing file ending with {suffix}; files={paths}") return sorted(matches)[-1] def file_entry(items: list[dict[str, Any]], path: str) -> dict[str, Any] | None: for item in items: if item.get("type") == "file" and item.get("path") == path: return item children = item.get("children") if isinstance(children, list): found = file_entry(children, path) if found is not None: return found return None def _jsonish(value: Any) -> Any: if isinstance(value, str): try: return json.loads(value) except json.JSONDecodeError: return value return value def _tool_output(event: dict[str, Any]) -> Any: for key in ("output", "content"): value = event.get(key) if value is not None: return _jsonish(value) state = event.get("state") if isinstance(state, dict): return _jsonish(state.get("output")) return None def _workspace_rel_path(workspace_id: str, path: str) -> str: marker = f"/workspaces/{workspace_id}/" if marker in path: return path.split(marker, 1)[1] data_marker = f"/data/workspaces/{workspace_id}/" if path.startswith(data_marker): return path[len(data_marker) :] return path def persisted_evidence_paths(events: list[dict[str, Any]], workspace_id: str) -> tuple[str | None, str | None]: metrics_path: str | None = None report_path: str | None = None for event in events: if event.get("type") != "tool": continue output = _tool_output(event) if not isinstance(output, dict): continue persisted = output.get("persisted_evidence") if not isinstance(persisted, dict): continue evidence_file = persisted.get("evidence_file_path") if isinstance(evidence_file, str) and evidence_file.endswith("/metrics.json"): metrics_path = _workspace_rel_path(workspace_id, evidence_file) report_file = persisted.get("report_path") if isinstance(report_file, str) and report_file.endswith("/workspace_evidence_report.json"): report_path = _workspace_rel_path(workspace_id, report_file) return metrics_path, report_path def assert_no_raw_payload_in_metric_tool(events: list[dict[str, Any]]) -> None: metric_tools = [] for event in events: if event.get("type") != "tool": continue raw = json.dumps(event, ensure_ascii=False) if "clinic_cli.py metrics" in raw or "clinic-cli:metrics" in raw or "persisted_evidence" in raw: metric_tools.append(raw) if not metric_tools: raise AssertionError("no metric tool event observed") for raw in metric_tools: if '"payload"' in raw: raise AssertionError(f"metric tool stdout still contains raw payload: {raw[:2000]}") if '"workspace_evidence"' in raw: raise AssertionError(f"metric tool stdout still contains workspace_evidence envelope: {raw[:2000]}") def assert_no_raw_payload_in_collect_tool(events: list[dict[str, Any]]) -> None: collect_tools = [] for event in events: if event.get("type") != "tool": continue raw = json.dumps(event, ensure_ascii=False) if "collect_evidence.py" in raw or "evidence_summary_for_agent" in raw: collect_tools.append(raw) if not collect_tools: raise AssertionError("no collect_evidence tool event observed") for raw in collect_tools: if '"payload"' in raw: raise AssertionError(f"collect_evidence stdout still contains raw payload: {raw[:2000]}") if '"workspace_evidence"' in raw: raise AssertionError(f"collect_evidence stdout still contains workspace_evidence envelope: {raw[:2000]}") def file_size_bytes(entry: dict[str, Any]) -> int: raw = entry.get("size") if isinstance(raw, (int, float)): return int(raw) text = str(raw or "").strip() if not text: return 0 parts = text.split() try: value = float(parts[0]) except (ValueError, IndexError): return 0 unit = parts[1].lower() if len(parts) > 1 else "b" multiplier = { "b": 1, "bytes": 1, "kb": 1024, "mb": 1024 * 1024, "gb": 1024 * 1024 * 1024, }.get(unit, 1) return int(value * multiplier) def kubectl_backend_pod(namespace: str) -> str: cmd = [ "kubectl", "get", "pods", "-n", namespace, "-l", "app=backend", "-o", "jsonpath={.items[0].metadata.name}", ] pod = subprocess.check_output(cmd, text=True).strip() if pod: return pod fallback = subprocess.check_output(["kubectl", "get", "pods", "-n", namespace, "-o", "name"], text=True) for line in fallback.splitlines(): name = line.removeprefix("pod/") if name.startswith("backend-"): return name raise AssertionError(f"backend pod not found in namespace {namespace}") def kubectl_assert_file(namespace: str, pod: str, workspace_id: str, rel_path: str) -> None: abs_path = f"/data/workspaces/{workspace_id}/{rel_path}" cmd = ["kubectl", "exec", "-n", namespace, pod, "--", "test", "-s", abs_path] last_exc: subprocess.CalledProcessError | None = None for attempt in range(1, 6): try: subprocess.check_call( cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) return except subprocess.CalledProcessError as exc: last_exc = exc log(f"[kubectl] retry {attempt}/5 file assert failed: {rel_path}") time.sleep(float(attempt)) raise AssertionError(f"kubectl file assert failed after retries: {abs_path}") from last_exc def kubectl_assert_image_attachment_db( namespace: str, pod: str, workspace_id: str, attachment_id: str, ) -> None: code = r""" import json import sqlite3 import sys workspace_id = sys.argv[1] attachment_id = sys.argv[2] conn = sqlite3.connect("/data/db/investibook.db") conn.row_factory = sqlite3.Row row = conn.execute( "SELECT id, status, kind, summary_path, analysis_path FROM workspace_image_attachments WHERE workspace_id = ? AND id = ?", (workspace_id, attachment_id), ).fetchone() if not row: raise SystemExit("missing workspace_image_attachments row") if row["status"] != "ready": raise SystemExit(f"attachment is not ready: {dict(row)}") msg = conn.execute( "SELECT attachments FROM messages_v2 WHERE workspace_id = ? AND role = 'user' AND attachments IS NOT NULL ORDER BY created_at DESC LIMIT 1", (workspace_id,), ).fetchone() if not msg: raise SystemExit("missing user message attachment metadata") attachments = json.loads(msg["attachments"]) if attachment_id not in {str(item.get("id")) for item in attachments if isinstance(item, dict)}: raise SystemExit(f"latest user message does not reference {attachment_id}: {attachments}") print(json.dumps(dict(row), ensure_ascii=False)) """ cmd = [ "kubectl", "exec", "-n", namespace, pod, "--", "python3", "-c", code, workspace_id, attachment_id, ] subprocess.check_call(cmd) def run_direct_case(api: Api, args: argparse.Namespace, pod: str | None) -> str: ws = api.create_workspace(f"FT direct evidence {uuid.uuid4().hex[:8]}") log(f"[direct] workspace={ws}") api.add_jira_source(ws, args.issue_id) api.prepare_source(ws, args.issue_id, args.prepare_timeout) result = api.stream_chat(ws, args.direct_prompt, args.chat_timeout) log("[direct] chat complete; checking tool output and persisted files") assert_no_raw_payload_in_metric_tool(result.events) files = api.files(ws) paths = flatten_file_paths(files) metrics_path, report_path = persisted_evidence_paths(result.events, ws) metrics_path = metrics_path or find_path(paths, "metrics.json") report_path = report_path or find_path(paths, "workspace_evidence_report.json") if metrics_path not in paths: raise AssertionError(f"persisted metrics path not listed by file API: {metrics_path}; files={paths}") if report_path not in paths: raise AssertionError(f"persisted report path not listed by file API: {report_path}; files={paths}") metrics_entry = file_entry(files, metrics_path) or {} report = json.loads(api.file_text(ws, report_path)) metrics_size = file_size_bytes(metrics_entry) if metrics_size <= 0: raise AssertionError(f"direct metrics file is empty in file API: {metrics_path}") log(f"[direct] file API sees metrics={metrics_path} size={metrics_size}") if not (report.get("evidence") or {}).get("metrics"): raise AssertionError(f"direct workspace report has no metric evidence: {report_path}") if pod: log("[direct] kubectl asserts persisted files are non-empty") kubectl_assert_file(args.kube_namespace, pod, ws, metrics_path) kubectl_assert_file(args.kube_namespace, pod, ws, report_path) log(f"[direct] PASS metrics={metrics_path} report={report_path}") return ws def run_diagnosis_case(api: Api, args: argparse.Namespace, pod: str | None) -> str: ws = api.create_workspace(f"FT diagnosis evidence {uuid.uuid4().hex[:8]}") log(f"[diagnosis] workspace={ws}") result = api.stream_chat(ws, args.diagnosis_prompt, args.chat_timeout) log("[diagnosis] chat complete; checking persisted files") assert_no_raw_payload_in_collect_tool(result.events) paths = flatten_file_paths(api.files(ws)) report_path = find_path(paths, "workspace_evidence_report.json") summary_path = find_path(paths, "evidence_summary_for_agent.txt") metrics_path = find_path(paths, "metrics.json") if any(path.endswith("/diagnosis_report.json") for path in paths): raise AssertionError(f"diagnosis_report.json should not be generated: {paths}") report = json.loads(api.file_text(ws, report_path)) summary = api.file_text(ws, summary_path) if report.get("type") != "workspace_evidence_report": raise AssertionError(f"unexpected workspace evidence report type: {report_path}") if not ((report.get("evidence") or {}).get("metrics")): raise AssertionError(f"workspace evidence report has no metric evidence: {report_path}") if "Evidence" not in summary and "证据" not in summary: raise AssertionError(f"summary does not look like an evidence summary: {summary_path}") if pod: log("[diagnosis] kubectl asserts persisted files are non-empty") kubectl_assert_file(args.kube_namespace, pod, ws, report_path) kubectl_assert_file(args.kube_namespace, pod, ws, summary_path) kubectl_assert_file(args.kube_namespace, pod, ws, metrics_path) log(f"[diagnosis] PASS report={report_path} summary={summary_path}") return ws def run_image_case(api: Api, args: argparse.Namespace, pod: str | None) -> str: if not args.image_path: raise SystemExit("--image-path is required for --case image") if not os.path.isfile(args.image_path): raise SystemExit(f"--image-path does not exist: {args.image_path}") ws = api.create_workspace(f"FT image attachment {uuid.uuid4().hex[:8]}") log(f"[image] workspace={ws}") uploaded = api.upload_image_attachment(ws, args.image_path) attachment_id = str(uploaded["id"]) log(f"[image] uploaded attachment={attachment_id} path={uploaded.get('rel_path')}") ready = api.wait_image_attachment_ready(ws, attachment_id, args.image_parse_timeout) summary_path = str(ready.get("summary_path") or "") analysis_path = str(ready.get("analysis_path") or "") if not summary_path or not analysis_path: raise AssertionError(f"ready attachment is missing output paths: {ready}") summary = api.file_text(ws, summary_path) analysis = json.loads(api.file_text(ws, analysis_path)) if analysis.get("type") != "image_attachment_analysis": raise AssertionError(f"unexpected image analysis type: {analysis_path}: {analysis}") if analysis.get("kind") not in {"metric_chart", "mixed", "text", "unknown"}: raise AssertionError(f"unexpected image kind: {analysis}") if "attachment_id" not in summary or attachment_id not in summary: raise AssertionError(f"summary does not identify attachment {attachment_id}: {summary[:1000]}") if "TiKV" not in summary and "CPU" not in summary and analysis.get("kind") == "unknown": raise AssertionError(f"image parser did not extract useful TiKV/CPU context: {summary[:1000]}") log("[image] parser ready; sending chat with attachment_ids") result = api.stream_chat( ws, args.image_prompt, args.chat_timeout, attachment_ids=[attachment_id], ) if not result.text.strip(): raise AssertionError("image chat produced no assistant text") lower_text = result.text.lower() if "tikv" not in lower_text and "cpu" not in lower_text: raise AssertionError(f"assistant did not appear to use image summary: {result.text[:2000]}") messages = api.messages(ws) user_with_attachment = [ msg for msg in messages if msg.get("role") == "user" and any(item.get("id") == attachment_id for item in (msg.get("attachments") or [])) ] if not user_with_attachment: raise AssertionError(f"messages API does not show attachment metadata: {messages}") files = flatten_file_paths(api.files(ws)) for path in (str(ready.get("rel_path") or ""), summary_path, analysis_path): if path not in files: raise AssertionError(f"image attachment file not listed by file API: {path}; files={files}") if pod: log("[image] kubectl asserts files and DB state") kubectl_assert_file(args.kube_namespace, pod, ws, str(ready.get("rel_path") or "")) kubectl_assert_file(args.kube_namespace, pod, ws, summary_path) kubectl_assert_file(args.kube_namespace, pod, ws, analysis_path) kubectl_assert_image_attachment_db(args.kube_namespace, pod, ws, attachment_id) log(f"[image] PASS attachment={attachment_id} summary={summary_path}") return ws def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Dev API FT for workspace evidence file-first flow") parser.add_argument("--base-url", default=os.getenv("INVESTIBOOK_DEV_BASE_URL", "https://dev.investibook.com")) parser.add_argument("--token", default=os.getenv("INVESTIBOOK_DEV_TOKEN") or os.getenv("INVESTIBOOK_API_KEY") or "") parser.add_argument("--issue-id", default="O11Y-2447801") parser.add_argument("--direct-prompt", default=DEFAULT_DIRECT_PROMPT) parser.add_argument("--diagnosis-prompt", default=DEFAULT_DIAGNOSIS_PROMPT) parser.add_argument("--case", choices=["direct", "diagnosis", "image", "both", "all"], default="both") parser.add_argument("--image-path", default=os.getenv("INVESTIBOOK_FT_IMAGE_PATH", "")) parser.add_argument("--image-prompt", default=DEFAULT_IMAGE_PROMPT) parser.add_argument("--image-parse-timeout", type=float, default=120.0) parser.add_argument("--chat-timeout", type=float, default=420.0) parser.add_argument("--prepare-timeout", type=float, default=120.0) parser.add_argument("--http-timeout", type=float, default=60.0) parser.add_argument("--kube-namespace", default="investibook") parser.add_argument("--skip-kubectl-assert", action="store_true") parser.add_argument("--cleanup-workspaces", action="store_true") return parser.parse_args() def main() -> int: args = parse_args() if not args.token: raise SystemExit("missing token: set INVESTIBOOK_DEV_TOKEN or INVESTIBOOK_API_KEY") api = Api(args.base_url, args.token, args.http_timeout) created: list[str] = [] pod: str | None = None try: if not args.skip_kubectl_assert: pod = kubectl_backend_pod(args.kube_namespace) log(f"[kubectl] backend_pod={pod}") if args.case in {"direct", "both", "all"}: created.append(run_direct_case(api, args, pod)) if args.case in {"diagnosis", "both", "all"}: created.append(run_diagnosis_case(api, args, pod)) if args.case in {"image", "all"}: created.append(run_image_case(api, args, pod)) log("[ft] PASS") return 0 finally: if args.cleanup_workspaces: for ws in created: try: api.delete_workspace(ws) log(f"[cleanup] deleted {ws}") except Exception as exc: log(f"[cleanup] failed to delete {ws}: {exc}", file=sys.stderr) api.close() if __name__ == "__main__": raise SystemExit(main())