75 lines
2.8 KiB
Python
75 lines
2.8 KiB
Python
"""Persist useful failure evidence even when a worker must be terminated."""
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import uuid
|
|
|
|
from artemis.config import TRACES_PATH
|
|
from artemis.runtime.case_result import resolve_case_result
|
|
|
|
|
|
def read_worker_outcome(session_id: str) -> dict:
|
|
"""Read a worker's terminal signal without waiting for hung SDK threads."""
|
|
try:
|
|
data = json.loads(
|
|
(TRACES_PATH / session_id / "run_outcome.json").read_text(encoding="utf-8")
|
|
)
|
|
if (
|
|
isinstance(data, dict)
|
|
and data.get("worker_finalized") is True
|
|
and data.get("task_status")
|
|
in {
|
|
"failed",
|
|
"completed",
|
|
"cancelled",
|
|
"interrupted",
|
|
}
|
|
):
|
|
return data
|
|
except (OSError, ValueError):
|
|
pass # Atomic outcome is absent until the task finalizes.
|
|
return {}
|
|
|
|
|
|
def save_failure_result(session_id: str, error: str, *, code: str = "execution_failed") -> None:
|
|
if not session_id or not all(c.isalnum() or c in "-_" for c in session_id):
|
|
raise ValueError("无效的任务编号")
|
|
directory = TRACES_PATH / session_id
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
payload: dict = {}
|
|
for name in ("result.json", "run_outcome.json"):
|
|
try:
|
|
previous = json.loads((directory / name).read_text(encoding="utf-8"))
|
|
if isinstance(previous, dict):
|
|
prior_result = resolve_case_result(payload, "failed")
|
|
prior_tests = payload.get("tests")
|
|
payload.update(previous)
|
|
if isinstance(prior_tests, dict) and isinstance(previous.get("tests"), dict):
|
|
payload["tests"] = {**prior_tests, **previous["tests"]}
|
|
if prior_result and prior_result.status == "failed":
|
|
payload["test_result"] = prior_result.model_dump()
|
|
except (OSError, ValueError):
|
|
pass # Missing/truncated results are expected after a killed worker.
|
|
payload.update(
|
|
{
|
|
"status": "failed",
|
|
"task_status": "failed",
|
|
"explanation": error,
|
|
"error": error,
|
|
"code": code,
|
|
"worker_finalized": True,
|
|
"tests": payload.get("tests") or {},
|
|
}
|
|
)
|
|
test_result = resolve_case_result(payload, "failed", error)
|
|
payload["test_result"] = test_result.model_dump() if test_result else None
|
|
for name in ("result.json", "run_outcome.json"):
|
|
target = directory / name
|
|
temporary = target.with_name(f".{name}.{uuid.uuid4().hex}.tmp")
|
|
try:
|
|
temporary.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
|
|
os.replace(temporary, target)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|