147 lines
5.7 KiB
Python
147 lines
5.7 KiB
Python
"""Flash's exit-only acceptance check uses the same read-only Checker as Pro."""
|
|
|
|
from pathlib import Path
|
|
|
|
from artemis.agents.checker.checker import run_final_check
|
|
from artemis.context import ArtemisContext
|
|
from artemis.llm.diagnostics import final_review_error_text
|
|
from artemis.graph.checkpoints import (
|
|
RunOutcome,
|
|
TestSummary,
|
|
final_check_enabled,
|
|
publish_checker_event,
|
|
verdict_payloads,
|
|
write_run_outcome,
|
|
)
|
|
from artemis.runtime.case_result import CaseTestResult, resolve_case_result
|
|
from artemis.services.llm import final_review_budget
|
|
from artemis.utils.logger import get_logger
|
|
from artemis.utils.notes import save_verified_note
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
async def settle_flash_outcome(
|
|
ctx: ArtemisContext,
|
|
goal: str,
|
|
execution_report: dict,
|
|
*,
|
|
final_report_received: bool = False,
|
|
) -> dict:
|
|
"""Check once at exit, without operating the device or rerunning the task."""
|
|
stop_capture = getattr(ctx, "on_execution_review", None)
|
|
if stop_capture:
|
|
await stop_capture()
|
|
status = str(execution_report.get("status") or "failed")
|
|
result = None
|
|
delivery = None
|
|
report_review = None
|
|
if status == "completed" or final_report_received:
|
|
body = execution_report.get("explanation")
|
|
directory = getattr(getattr(ctx, "data_engine", None), "base_dir", None)
|
|
if isinstance(body, str) and body.strip() and isinstance(directory, (str, Path)):
|
|
try:
|
|
receipt = save_verified_note(directory, "output", body)
|
|
delivery = {
|
|
**receipt,
|
|
"report_path": receipt["path"],
|
|
"allow_report_corrections": True,
|
|
}
|
|
except (OSError, ValueError) as exc:
|
|
logger.warning(f"Flash 最终答复保存失败：{exc}")
|
|
delivery = {"ok": False, "verified": False, "error": str(exc)}
|
|
if final_check_enabled(ctx):
|
|
publish_checker_event(
|
|
ctx,
|
|
{
|
|
"event": "attempt_started",
|
|
"phase": "final",
|
|
"attempt_id": "final#1",
|
|
"checkpoint_id": "final",
|
|
"check_items": [],
|
|
"status": "running",
|
|
},
|
|
)
|
|
try:
|
|
report = await final_review_budget(ctx, attempt_id="final#1").run(
|
|
lambda: run_final_check(
|
|
ctx,
|
|
goal=goal,
|
|
plan_text="",
|
|
ledger=[],
|
|
check_items=[],
|
|
attempt_id="final#1",
|
|
execution_report=execution_report,
|
|
report_delivery=delivery,
|
|
),
|
|
)
|
|
result = report.test_result
|
|
if report.report_corrections and directory:
|
|
from artemis.agents.checker.report_review import apply_report_review
|
|
|
|
try:
|
|
report_review = apply_report_review(
|
|
directory, execution_report, report.report_corrections
|
|
)
|
|
except (OSError, ValueError) as exc:
|
|
report_review = {"applied": False, "reason": str(exc)}
|
|
logger.warning(f"报告复核修改未保存：{exc}")
|
|
publish_checker_event(
|
|
ctx,
|
|
{
|
|
"event": "attempt_finished",
|
|
"phase": "final",
|
|
"attempt_id": "final#1",
|
|
"checkpoint_id": "final",
|
|
"status": "done",
|
|
"verdicts": verdict_payloads(report.verdicts),
|
|
"test_result": result.model_dump() if result else None,
|
|
"route": "end",
|
|
},
|
|
)
|
|
except Exception as exc:
|
|
detail = final_review_error_text(exc)
|
|
logger.warning(f"Flash 测试验收未完成：{detail}")
|
|
result = CaseTestResult(
|
|
status="inconclusive", reason=f"操作已结束，但测试验收未完成：{detail}。"
|
|
)
|
|
publish_checker_event(
|
|
ctx,
|
|
{
|
|
"event": "attempt_finished",
|
|
"phase": "final",
|
|
"attempt_id": "final#1",
|
|
"checkpoint_id": "final",
|
|
"status": "error",
|
|
"error": result.reason,
|
|
"route": "end",
|
|
},
|
|
)
|
|
else:
|
|
result = CaseTestResult(
|
|
status="inconclusive", reason="已关闭最终检查，未进行测试结果验收。"
|
|
)
|
|
|
|
outcome = RunOutcome(
|
|
task_status="completed" if status == "completed" else "blocked", tests=TestSummary()
|
|
)
|
|
payload = outcome.model_dump()
|
|
if report_review:
|
|
payload["report_review"] = report_review
|
|
if result:
|
|
payload["test_result"] = result.model_dump()
|
|
resolved = resolve_case_result(payload, status, str(execution_report.get("explanation") or ""))
|
|
payload["test_result"] = resolved.model_dump() if resolved else None
|
|
if ctx.data_engine:
|
|
write_run_outcome(
|
|
ctx.data_engine.base_dir,
|
|
outcome,
|
|
{
|
|
"test_result": payload["test_result"],
|
|
**({"report_review": report_review} if report_review else {}),
|
|
},
|
|
)
|
|
publish_checker_event(ctx, {"event": "run_outcome", "phase": "outcome", **payload})
|
|
ctx.run_outcome = payload
|
|
return payload
|