feat: add shared test acceptance and release 1.5.0

artemis/runtime/case_result.py

@@ -0,0 +1,132 @@
+"""A test verdict is independent of the task's execution/lifecycle status."""
+
+import json
+from pathlib import Path
+from typing import Literal
+
+from pydantic import BaseModel, Field, ValidationError
+
+from artemis.config import TRACES_PATH
+
+
+class CaseTestResult(BaseModel):
+    status: Literal["passed", "failed", "inconclusive", "blocked"] = Field(
+        description="Judge the system under test against the user's expectations. A correctly executed inspection that reveals an expectation mismatch is failed, even if the executor completed and reported the mismatch correctly."
+    )
+    reason: str = Field(description="用简体中文说明提示词中的要求是否满足，以及不能判定的原因。")
+    evidence: list[str] = Field(
+        default_factory=list,
+        description="支撑结论的具体步骤、截图内容或只读探测结果；执行者声称完成不属于独立证据。",
+    )
+
+
+RESULT_LABELS = {
+    "passed": "测试通过",
+    "failed": "测试不通过",
+    "inconclusive": "未验证",
+    "blocked": "测试阻塞",
+}
+_TERMINAL = {"completed", "success", "failed", "cancelled", "interrupted", "skipped"}
+
+
+def normalize_case_result(result: CaseTestResult) -> CaseTestResult:
+    evidence = [item.strip() for item in result.evidence if item.strip()]
+    if result.status in {"passed", "failed"} and not evidence:
+        return CaseTestResult(
+            status="inconclusive",
+            reason="验收结论缺少具体证据，不能确认测试是否通过。",
+        )
+    return result.model_copy(update={"evidence": evidence})
+
+
+def resolve_case_result(
+    outcome: dict | None, execution_status: str, error: str = ""
+) -> CaseTestResult | None:
+    """Preserve confirmed defects; unfinished execution never becomes a pass."""
+    outcome = outcome if isinstance(outcome, dict) else {}
+    tests = outcome.get("tests") or {}
+    tests = tests if isinstance(tests, dict) else {}
+    candidate = None
+    if isinstance(outcome.get("test_result"), dict):
+        try:
+            candidate = normalize_case_result(CaseTestResult.model_validate(outcome["test_result"]))
+        except ValidationError:
+            candidate = None
+
+    failed_items = tests.get("failed_items") or []
+    if not isinstance(failed_items, list):
+        failed_items = []
+    failures = [
+        item
+        for item in failed_items
+        if isinstance(item, dict)
+        and item.get("kind", "assert") == "assert"
+        and item.get("evidence")
+    ]
+    if failures:
+        return CaseTestResult(
+            status="failed",
+            reason="；".join(str(item.get("item_text") or "测试断言不满足") for item in failures),
+            evidence=[str(item["evidence"]) for item in failures],
+        )
+    if candidate and candidate.status == "failed":
+        return candidate
+    if execution_status not in _TERMINAL:
+        return None
+    if execution_status not in {"completed", "success"} or outcome.get("task_status") in {
+        "blocked",
+        "partial",
+        "failed",
+        "cancelled",
+        "interrupted",
+    }:
+        return CaseTestResult(
+            status="blocked",
+            reason=error
+            or str(
+                outcome.get("error")
+                or outcome.get("explanation")
+                or "任务未完整执行，无法完成验收。"
+            ),
+        )
+    if candidate:
+        if candidate.status == "passed" and (tests.get("inconclusive") or tests.get("unchecked")):
+            return CaseTestResult(
+                status="inconclusive",
+                reason="部分检查项缺少证据或尚未验证。",
+                evidence=candidate.evidence,
+            )
+        return candidate
+    # Existing Pro runs already have a verdict ledger. A bare 'completed' from
+    # Flash (or a process exit code) is never used as a positive test result.
+    if tests.get("failed"):
+        return CaseTestResult(
+            status="blocked", reason="执行检查未满足；没有证据确认被测功能不符合预期。"
+        )
+    if tests.get("inconclusive") or tests.get("unchecked"):
+        return CaseTestResult(status="inconclusive", reason="部分检查项缺少证据或尚未验证。")
+    if tests.get("passed"):
+        return CaseTestResult(
+            status="passed",
+            reason="已记录的验证项全部通过。",
+            evidence=[f"已有检查结果：{tests['passed']} 项通过。"],
+        )
+    return CaseTestResult(
+        status="inconclusive", reason="没有保存测试验收结论；执行完成不等于测试通过。"
+    )
+
+
+def read_case_result(
+    session_id: str, execution_status: str, error: str = "", *, root: Path | None = None
+) -> dict | None:
+    if not session_id or not all(c.isalnum() or c in "-_" for c in session_id):
+        raise ValueError("无效的任务编号")
+    outcome = {}
+    try:
+        outcome = json.loads(
+            ((root or TRACES_PATH) / session_id / "run_outcome.json").read_text(encoding="utf-8")
+        )
+    except (OSError, ValueError):
+        pass  # Old tasks and active tasks may not have an outcome yet.
+    result = resolve_case_result(outcome, execution_status, error)
+    return result.model_dump() if result else None