test: cover result card evidence disclosure

apps/admin_console/routers/automation.py

@@ -13,6 +13,7 @@ from apps.admin_console.services import automation_service as service
 from apps.admin_console.services import batch_archive as archive_service
 from apps.admin_console.services import report_export, task_import
 from artemis.config.settings import settings
+from artemis.runtime.case_result import read_case_result
 
 router = APIRouter(prefix="/api/automation", tags=["automation"])
 
@@ -122,7 +123,12 @@ def batch_by_request(request_id: str):
 
 @router.get("/batches/{batch_id}")
 def batch_detail(batch_id: str):
-    return {**require("batch", batch_id), "jobs": store.jobs(batch_id)}
+    jobs = store.jobs(batch_id)
+    for job in jobs:
+        job["test_result"] = read_case_result(
+            job["session_id"], job["status"], job.get("error") or ""
+        )
+    return {**require("batch", batch_id), "jobs": jobs}
 
 
 @router.delete("/batches/{batch_id}")
@@ -260,6 +266,9 @@ def history(
     items = [dict(row) for row in rows]
     jobs = store.jobs_for_sessions([item["session_id"] for item in items])
     for item in items:
+        item["test_result"] = read_case_result(
+            item["session_id"], item["status"], item.get("error") or ""
+        )
         if job := jobs.get(item["session_id"]):
             item["name"] = task_import.execution_title(job)
             item["iteration"] = job.get("iteration", 1)

apps/admin_console/routers/sessions.py

@@ -20,6 +20,7 @@ from fastapi import APIRouter, HTTPException
 
 from artemis.config import DB_PATH, TRACES_PATH
 from artemis.runtime import trace_store
+from artemis.runtime.case_result import read_case_result
 
 try:
     from admin_console.core.state import state
@@ -130,6 +131,9 @@ def _list_sessions_sync():
         else:
             row_dict["video_url"] = None
         row_dict["recording_status"] = recording_status
+        row_dict["test_result"] = read_case_result(
+            s_id, row_dict.get("status") or "", row_dict.get("error") or "", root=TRACES_PATH
+        )
 
         agent_names = agent_names_by_session.get(s_id, [])
         sess_profile = model_service.resolve_session_profile(
@@ -189,7 +193,11 @@ async def get_session_details(session_id: str):
     row = session_repo.get_session_by_id(session_id)
     if not row:
         raise HTTPException(status_code=404, detail=f"Session {session_id} not found")
-    return dict(row)
+    result = dict(row)
+    result["test_result"] = read_case_result(
+        session_id, result.get("status") or "", root=TRACES_PATH
+    )
+    return result
 
 
 @router.get("/api/sessions/{session_id}/usage")

apps/admin_console/services/report_export.py

@@ -23,6 +23,7 @@ from apps.admin_console.database.repositories.session_repository import SessionR
 from apps.admin_console.database.repositories.step_repository import StepRepository
 from apps.admin_console.services.automation_store import store, identifier
 from apps.admin_console.services.task_import import execution_title
+from artemis.runtime.case_result import RESULT_LABELS, resolve_case_result
 
 EXPORT_ROOT = WORKSPACE_ROOT / "artifacts" / "reports"
 EXPORT_LOCK = threading.RLock()
@@ -62,6 +63,9 @@ def text(value) -> str:
 
 
 def verdict(outcome: dict | None) -> str:
+    result = (outcome or {}).get("test_result")
+    if isinstance(result, dict) and result.get("status") in RESULT_LABELS:
+        return RESULT_LABELS[result["status"]]
     tests = (outcome or {}).get("tests") or {}
     if tests.get("failed", 0):
         return "验证失败"
@@ -179,6 +183,9 @@ def session_report(session_id: str, db_path=None) -> dict:
     case = meta.get("case") or {}
     status = meta.get("terminal_status") or meta.get("status") or session.get("status", "unknown")
     error = failure_reason(directory, status, meta.get("error"))
+    test_result = resolve_case_result(outcome, status, error)
+    if test_result:
+        outcome = {**outcome, "test_result": test_result.model_dump()}
     verification = verdict(outcome)
     if status in {"failed", "interrupted"} and verification.startswith("未验证"):
         verification = "未验证（任务异常终止）"
@@ -197,6 +204,7 @@ def session_report(session_id: str, db_path=None) -> dict:
         "expected": case.get("expected", ""),
         "status": status,
         "verdict": verification,
+        "test_result": test_result.model_dump() if test_result else None,
         "outcome": outcome,
         "content": content,
         "steps": steps,
@@ -233,6 +241,11 @@ def metadata(report: dict) -> list[tuple]:
         ("预期结果", report["expected"] or "未指定"),
         ("执行状态", STATUS.get(report["status"], report["status"])),
         ("验证结论", report["verdict"]),
+        ("判定依据", (report.get("test_result") or {}).get("reason") or "未记录"),
+        (
+            "验收证据",
+            "\n".join((report.get("test_result") or {}).get("evidence") or []) or "未记录",
+        ),
         ("开始时间", time_text(report["start"])),
         ("结束时间", time_text(report["end"])),
         ("设备", report["device"]),
@@ -434,6 +447,11 @@ def word_report(report: dict, path: Path):
     if report["error"]:
         doc.add_heading("异常原因", 1)
         doc.add_paragraph(text(report["error"]))
+    if result := report.get("test_result"):
+        doc.add_heading("测试结论与依据", 1)
+        doc.add_paragraph(text(result.get("reason") or "未记录"))
+        for evidence in result.get("evidence") or []:
+            doc.add_paragraph(text(evidence), style="List Bullet")
     doc.add_heading("任务信息", 1)
     details = [
         (label, value)
@@ -441,10 +459,11 @@ def word_report(report: dict, path: Path):
         if label in {"任务编号", "用例编号", "检测项目", "测试子项", "开始时间", "结束时间", "设备"}
     ]
     word_table(doc, details, (3.2, 13.3))
-    doc.add_heading("任务目标", 2)
+    doc.add_heading("任务目标与预期" if not report["expected"] else "任务目标", 2)
     doc.add_paragraph(report["goal"] or "未记录")
+    if report["expected"]:
         doc.add_heading("预期结果", 2)
-    doc.add_paragraph(report["expected"] or "未指定")
+        doc.add_paragraph(report["expected"])
     if report.get("preconditions"):
         doc.add_paragraph("前置约束：" + text(report["preconditions"]))
     if report.get("task_config"):

apps/showcase_ui/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "artemis-showcase-ui",
-  "version": "1.4.0",
+  "version": "1.5.0",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "artemis-showcase-ui",
-      "version": "1.4.0",
+      "version": "1.5.0",
       "dependencies": {
         "@angular/cdk": "^22.1.5",
         "@angular/common": "^22.1.4",

apps/showcase_ui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "artemis-showcase-ui",
-  "version": "1.4.0",
+  "version": "1.5.0",
   "scripts": {
     "ng": "ng",
     "start": "ng serve --proxy-config proxy.conf.json",

apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.html

@@ -86,6 +86,7 @@ limitations under the License.
                 }
                 {{ getTaskStatus(session) | uiLabel }}
               </span>
+              <app-test-result-badge [result]="session.test_result" [executionStatus]="getTaskStatus(session)" />
               <div class="item-header-actions">
                 @if (getDeviceSerial(session)) {
                   <span class="item-device" [title]="'Device: ' + getDeviceSerial(session)">
@@ -549,7 +550,7 @@ limitations under the License.
         }
         @for (phase of phases(); track trackPhase($index, phase); let first = $first) {
           <div class="phase-container">
-            @if (hasVisibleBlocks(phase)) {
+            @if (hasVisibleBlocks(phase) && phase.blocks[0]?.data?.phase !== 'outcome') {
               <div
                 class="phase-header"
                 [class.first-phase]="first && startupWorkItems().length === 0"
@@ -584,6 +585,9 @@ limitations under the License.
                              and how each is judged; the chronological stream follows, and the
                              result rows close the block (see check-result below). -->
                         @if (isCheckerBlock(block)) {
+                          @if (block.data.phase === 'outcome' && block.data.test_result) {
+                            <app-test-result-card [result]="block.data.test_result" />
+                          } @else {
                           <div class="check-lane" [ngClass]="'tone-' + getCheckerTone(block)">
                             <div class="check-header">
                               <span class="stream-status-dot" [class.streaming]="isBlockRunning(block)"></span>
@@ -606,6 +610,7 @@ limitations under the License.
                             }
                           </div>
                           }
+                        }
                         <!-- Text, tool calls, and actions share one chronological timeline. -->
                         @if (getSortedStepEvents(block.data).length > 0) {
                           <div class="step-actions-container">
@@ -1268,7 +1273,7 @@ limitations under the License.
                           </div>
                         }
                         <!-- Check block result: one row per check item, keyed by its number above -->
-                        @if (hasCheckerResult(block)) {
+                        @if (hasCheckerResult(block) && !(block.data.phase === 'outcome' && block.data.test_result)) {
                           <div class="check-result" [ngClass]="'tone-' + getCheckerTone(block)">
                             @for (verdict of getCheckerVerdicts(block); track $index) {
                               <div class="check-verdict" [ngClass]="verdict.status">

apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.ts

@@ -25,6 +25,8 @@ import { RouterLink } from '@angular/router';
 import { HttpClient } from '@angular/common/http';
 import { AgentService, GuidanceEvent, StartupProgressEvent } from '../../services/agent.service';
 import { Session, ModelInfo, SessionUsage } from '../../core/models/session.model';
+import { TestResultBadgeComponent, TEST_RESULT_LABELS } from '../test-result-badge/test-result-badge.component';
+import { TestResultCardComponent } from '../test-result-badge/test-result-card.component';
 import { MarkdownSegment, MarkdownLine, NoteMilestone, ParsedNote } from '../../core/models/markdown.model';
 import { OverlayModule, ConnectedPosition } from '@angular/cdk/overlay';
 import { StepBlock, PhaseBlock, StepEvent, ActionParam, CheckerResult, StreamResetNotice, DEFAULT_STREAM_RESET_MESSAGE } from '../../core/models/stream.model';
@@ -237,7 +239,7 @@ export type { MarkdownSegment, MarkdownLine, NoteMilestone, ParsedNote, StreamRe
 @Component({
   selector: 'app-agent-stream',
   standalone: true,
-  imports: [TaskIdCopyComponent, UiLabelPipe, CommonModule, FormsModule, OverlayModule, TaskHistoryComponent, AutomationPanelComponent],
+  imports: [TaskIdCopyComponent, UiLabelPipe, CommonModule, FormsModule, OverlayModule, TaskHistoryComponent, AutomationPanelComponent, TestResultBadgeComponent, TestResultCardComponent],
   templateUrl: './agent-stream.component.html',
   styleUrl: './agent-stream.component.scss',
   changeDetection: ChangeDetectionStrategy.OnPush
@@ -1754,7 +1756,9 @@ export class AgentStreamComponent implements AfterViewInit {
   /** Follow-ups not already visible in the verdict rows (the findings text repeats them). */
   public getCheckerNotes(block: any): string[] {
     const d = block?.data || {};
-    if (d.phase === 'outcome') return Array.isArray(d.last_findings) ? d.last_findings : [];
+    if (d.phase === 'outcome') {
+      return [d.test_result?.reason, ...(d.test_result?.evidence || []), ...(Array.isArray(d.last_findings) ? d.last_findings : [])].filter(Boolean);
+    }
     const notes: string[] = [];
     if (Array.isArray(d.unmet_subgoals)) {
       for (const text of d.unmet_subgoals) notes.push(`尚未完成：${text}`);
@@ -1796,7 +1800,9 @@ export class AgentStreamComponent implements AfterViewInit {
   public getCheckerTone(block: any): string {
     const d = block?.data || {};
     if (d.phase === 'outcome') {
-      return d.task_status === 'completed' ? 'passed' : (d.task_status === 'blocked' ? 'failed' : 'inconclusive');
+      if (d.test_result) return d.test_result.status === 'blocked' ? 'inconclusive' : d.test_result.status;
+      if (d.tests?.failed) return 'failed';
+      return d.tests?.passed && !d.tests?.unchecked && !d.tests?.inconclusive ? 'passed' : 'inconclusive';
     }
     if (d.isCompleted === false) return 'running';
     if (d.status === 'superseded' || d.status === 'unchecked') return 'muted';
@@ -1811,12 +1817,12 @@ export class AgentStreamComponent implements AfterViewInit {
     const d = block?.data || {};
     if (d.phase === 'outcome') {
       const t = d.tests || {};
-      const label = d.task_status === 'completed' ? "目标已完成"
+      const label = d.test_result ? TEST_RESULT_LABELS[d.test_result.status as keyof typeof TEST_RESULT_LABELS]
-        : (d.task_status === 'blocked' ? "受阻" : "部分完成");
+        : (d.task_status === 'completed' ? "执行完成" : (d.task_status === 'blocked' ? "执行受阻" : "部分完成"));
       const counts = [
-        [t.passed, 'passed'],
+        [t.passed, '项通过'],
-        [t.failed, 'failed'],
+        [t.failed, '项不通过'],
-        [t.inconclusive, 'inconclusive'],
+        [t.inconclusive, '项无法确认'],
         [t.unchecked, "未检查"]
       ]
         .filter(([n]) => Number(n) > 0)

apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.html

@@ -14,10 +14,11 @@
     @if (report(); as r) {
       <div class="report-overview">
         <strong class="report-title">{{r.name}}</strong>
-        <div class="report-badges"><span [class.failed]="r.status === 'failed'">{{label(r.status)}}</span><span>{{r.verdict}}</span></div>
+        <div class="report-badges"><span [class.failed]="r.status === 'failed'">{{label(r.status)}}</span><app-test-result-badge [result]="r.test_result" [executionStatus]="r.status" /></div>
+        @if(r.test_result; as result){<app-test-result-card [result]="result" />}
         @if (r.error) { <div class="report-error" role="status"><strong>异常原因</strong><p>{{r.error}}</p></div> }
       </div>
-      <div class="report-meta"><div><strong>任务目标</strong><p>{{r.goal || '未记录'}}</p></div><div><strong>预期结果</strong><p>{{r.expected || '未指定'}}</p></div></div>
+      <div class="report-meta"><div><strong>任务目标与预期</strong><p>{{r.goal || '未记录'}}</p></div>@if(r.expected){<div><strong>预期结果</strong><p>{{r.expected}}</p></div>}</div>
       <div class="actions"><a [href]="'/api/automation/reports/'+r.id+'/docx'">下载 Word</a><a [href]="'/api/automation/reports/'+r.id+'/xlsx'">下载 Excel</a>
         <a [href]="'/api/automation/reports/'+r.id+'/screenshots'">步骤截图 ZIP</a>
         <a [href]="'/api/automation/reports/'+r.id+'/video'">分段录屏 ZIP</a>
@@ -102,6 +103,6 @@
        </div>
      }
     @if(b.report_error){<p class="error">报告生成失败：{{b.report_error}}</p>}
-      @for(j of b.jobs; track j.session_id){<div class="job"><button (click)="agent.selectSession(j.session_id,true); tab.set('reports')" [title]="j.case.goal">{{j.case.case_id ? j.case.case_id + ' ' : ''}}{{j.case.name}}@if(j.repeat_count > 1){ · 第 {{j.iteration}} / {{j.repeat_count}} 轮}</button><span>{{label(j.status)}}</span>@if(j.error){<small>{{j.error}}</small>}</div>}
+      @for(j of b.jobs; track j.session_id){<div class="job"><button (click)="agent.selectSession(j.session_id,true); tab.set('reports')" [title]="j.case.goal">{{j.case.case_id ? j.case.case_id + ' ' : ''}}{{j.case.name}}@if(j.repeat_count > 1){ · 第 {{j.iteration}} / {{j.repeat_count}} 轮}</button><span>{{label(j.status)}}</span><app-test-result-badge [result]="j.test_result" [executionStatus]="j.status" />@if(j.error){<small>{{j.error}}</small>}</div>}
   </section>}
 }

apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.ts

@@ -6,6 +6,8 @@ import { firstValueFrom } from 'rxjs';
 import { AgentService } from '../../services/agent.service';
 import { SystemService } from '../../services/system.service';
 import { renderMarkdownToHtml } from '../../utils/markdown-parser.util';
+import { TestResultBadgeComponent } from '../test-result-badge/test-result-badge.component';
+import { TestResultCardComponent } from '../test-result-badge/test-result-card.component';
 
 interface ArchiveProgress {
   status: 'idle' | 'preparing' | 'ready' | 'failed';
@@ -18,7 +20,7 @@ interface ArchiveProgress {
 }
 
 @Component({
-  selector: 'app-automation-panel', standalone: true, imports: [CommonModule, FormsModule],
+  selector: 'app-automation-panel', standalone: true, imports: [CommonModule, FormsModule, TestResultBadgeComponent, TestResultCardComponent],
   templateUrl: './automation-panel.component.html', styleUrl: './automation-panel.component.scss'
 })
 export class AutomationPanelComponent {

apps/showcase_ui/src/app/components/chat-interface/chat-interface.component.html

@@ -82,6 +82,7 @@ limitations under the License.
               }
               {{ getTaskStatus(session) | uiLabel }}
             </span>
+            <app-test-result-badge [result]="session.test_result" [executionStatus]="getTaskStatus(session)" />
             <div class="task-card-header-actions">
               @if (getDeviceSerial(session)) {
                 <span class="task-device" [title]="'Device: ' + getDeviceSerial(session)">

apps/showcase_ui/src/app/components/chat-interface/chat-interface.component.ts

@@ -23,6 +23,7 @@ import { CommonModule } from '@angular/common';
 import { FormsModule } from '@angular/forms';
 import { AgentService } from '../../services/agent.service';
 import { Session } from '../../core/models/session.model';
+import { TestResultBadgeComponent } from '../test-result-badge/test-result-badge.component';
 import { MarkdownSegment, MarkdownLine, NoteMilestone, ParsedNote } from '../../core/models/markdown.model';
 import { parseNote, parseNoteLines } from '../../utils/markdown-parser.util';
 
@@ -31,7 +32,7 @@ export type { MarkdownSegment, MarkdownLine, NoteMilestone, ParsedNote };
 @Component({
   selector: 'app-chat-interface',
   standalone: true,
-  imports: [TaskIdCopyComponent, UiLabelPipe, CommonModule, FormsModule, AutomationPanelComponent, TaskHistoryComponent],
+  imports: [TaskIdCopyComponent, UiLabelPipe, CommonModule, FormsModule, AutomationPanelComponent, TaskHistoryComponent, TestResultBadgeComponent],
   templateUrl: './chat-interface.component.html',
   styleUrl: './chat-interface.component.scss',
   changeDetection: ChangeDetectionStrategy.OnPush

apps/showcase_ui/src/app/components/task-history/task-history.component.spec.ts

@@ -15,6 +15,23 @@ describe('历史任务检索', () => {
   });
   afterEach(() => http.verify());
 
+  it('执行完成的任务仍能显示测试不通过，旧任务不会自动显示通过', () => {
+    const fixture = TestBed.createComponent(TaskHistoryComponent);
+    http.expectOne(r => r.url === '/api/automation/history').flush({total: 3, items: [
+      {session_id:'failed-test', name:'预期校验', status:'completed', start_time:100, test_result:{status:'failed', reason:'预期出现成功，实际出现错误', evidence:['步骤 3 显示错误']}},
+      {session_id:'passed-test', name:'已验收', status:'completed', start_time:100, test_result:{status:'passed', reason:'实际与预期一致', evidence:['步骤 3 显示成功']}},
+      {session_id:'old-task', name:'旧任务', status:'completed', start_time:100}
+    ]});
+    fixture.detectChanges();
+    const rows = fixture.nativeElement.querySelectorAll('button.row');
+    expect(rows[0].textContent).toContain('执行完成');
+    expect(rows[0].querySelector('app-test-result-badge').textContent).toContain('测试不通过');
+    expect(rows[0].querySelector('.test-result-badge').title).toContain('实际出现错误');
+    expect(rows[1].querySelector('app-test-result-badge').textContent).toContain('测试通过');
+    expect(rows[2].querySelector('app-test-result-badge').textContent).toContain('未验证');
+    fixture.destroy();
+  });
+
   it('历史 ID 独立于打开任务按钮，点击复制不会打开任务', async () => {
     spyOn(navigator.clipboard, 'writeText').and.rejectWith(new Error('blocked'));
     const fixture = TestBed.createComponent(TaskHistoryComponent);

apps/showcase_ui/src/app/components/task-history/task-history.component.ts

@@ -1,4 +1,5 @@
 import { TaskIdCopyComponent } from '../task-id-copy/task-id-copy.component';
+import { TestResultBadgeComponent } from '../test-result-badge/test-result-badge.component';
 import { Component, DestroyRef, inject, signal } from '@angular/core';
 import { CommonModule } from '@angular/common';
 import { FormsModule } from '@angular/forms';
@@ -6,14 +7,14 @@ import { HttpClient, HttpParams } from '@angular/common/http';
 import { AgentService } from '../../services/agent.service';
 
 @Component({
-  selector:'app-task-history', standalone:true, imports:[CommonModule,FormsModule,TaskIdCopyComponent],
+  selector:'app-task-history', standalone:true, imports:[CommonModule,FormsModule,TaskIdCopyComponent,TestResultBadgeComponent],
   template:`<section aria-label="历史任务"><h3>历史任务 <small>{{total()}} 条</small></h3>
     <input aria-label="搜索历史任务" placeholder="搜索名称、目标、任务编号或标签" [(ngModel)]="query" (ngModelChange)="search()">
     <div class="filters"><label>起始日期<input type="date" [(ngModel)]="start" (ngModelChange)="search()"></label><label>结束日期<input type="date" [(ngModel)]="end" (ngModelChange)="search()"></label><label>状态<select [(ngModel)]="status" (ngModelChange)="search()"><option value="">全部状态</option><option value="completed">执行完成</option><option value="failed">失败</option><option value="cancelled">已取消</option><option value="interrupted">中断</option><option value="skipped">跳过</option></select></label></div>
     @if(error()){<p role="alert">{{error()}}</p>}
     @if(loading()){<small>正在查询…</small>}
     @for(group of groups(); track group.key){<details [open]="!collapsed.has(group.key)" (toggle)="toggle(group.key,$event)"><summary>{{group.title}} <small>本页 {{group.rows.length}} 条</small></summary>
-      @for(row of group.rows; track row.session_id){<div class="history-item" [class]="'history-item status-'+row.status"><button class="row" [class.selected]="agent.currentSessionId() === row.session_id" (click)="agent.selectSession(row.session_id,true)" [title]="row.initial_goal || ''"><strong>{{row.name || row.initial_goal}}</strong><span class="history-meta"><b class="history-status">{{label(row.status)}}</b> <span>{{row.end_time ? '完成于 '+(row.end_time*1000 | date:'MM-dd HH:mm:ss') : '开始于 '+(row.start_time*1000 | date:'MM-dd HH:mm:ss')}}</span><span>总耗时 {{duration(row.start_time,row.end_time)}}</span></span>@if(row.error){<span class="history-error" [title]="row.error">{{row.error}}</span>}</button><app-task-id-copy [taskId]="row.session_id" /><button class="delete-history" (click)="remove(row.session_id)" aria-label="删除此历史任务"><span class="material-symbols-outlined">delete</span></button></div>}
+      @for(row of group.rows; track row.session_id){<div class="history-item" [class]="'history-item status-'+row.status"><button class="row" [class.selected]="agent.currentSessionId() === row.session_id" (click)="agent.selectSession(row.session_id,true)" [title]="row.initial_goal || ''"><strong>{{row.name || row.initial_goal}}</strong><span class="history-meta"><b class="history-status">{{label(row.status)}}</b><app-test-result-badge [result]="row.test_result" [executionStatus]="row.status" /> <span>{{row.end_time ? '完成于 '+(row.end_time*1000 | date:'MM-dd HH:mm:ss') : '开始于 '+(row.start_time*1000 | date:'MM-dd HH:mm:ss')}}</span><span>总耗时 {{duration(row.start_time,row.end_time)}}</span></span>@if(row.error){<span class="history-error" [title]="row.error">{{row.error}}</span>}</button><app-task-id-copy [taskId]="row.session_id" /><button class="delete-history" (click)="remove(row.session_id)" aria-label="删除此历史任务"><span class="material-symbols-outlined">delete</span></button></div>}
     </details>}
     @if(!loading() && !total()){<p>没有符合条件的历史任务。</p>}
     <div class="pager"><button [disabled]="page === 1 || loading()" (click)="page=page-1;load()">上一页</button><span>{{page}} / {{pages()}}</span><button [disabled]="page >= pages() || loading()" (click)="page=page+1;load()">下一页</button></div>

apps/showcase_ui/src/app/components/test-result-badge/test-result-badge.component.ts

@@ -0,0 +1,31 @@
+import { Component, computed, input } from '@angular/core';
+import { CaseTestResult } from '../../core/models/session.model';
+
+export const TEST_RESULT_LABELS: Record<CaseTestResult['status'], string> = {
+  passed: '测试通过', failed: '测试不通过', inconclusive: '未验证', blocked: '测试阻塞'
+};
+
+@Component({
+  selector: 'app-test-result-badge',
+  standalone: true,
+  template: `<span [class]="'test-result-badge ' + status()" [class.prominent]="prominent()" [title]="result()?.reason || '执行状态与测试结论分别记录'"><span class="material-symbols-outlined" aria-hidden="true">{{icon()}}</span>{{label()}}</span>`,
+  styles: [`
+    :host{display:inline-flex;vertical-align:middle;max-width:100%}
+    .test-result-badge{display:inline-flex;align-items:center;gap:4px;white-space:nowrap;border-radius:6px;border:1px solid #cbd5e1;padding:3px 8px;font-size:12px;font-weight:700;line-height:1.5;background:#f1f5f9;color:#475569}
+    .material-symbols-outlined{font-size:16px;line-height:1}
+    .passed{background:#15803d;color:#fff;border-color:#15803d}
+    .failed{background:#b91c1c;color:#fff;border-color:#b91c1c}
+    .inconclusive{background:#fff7ed;color:#9a3412;border-color:#fdba74}
+    .blocked{background:#fef3c7;color:#92400e;border-color:#fbbf24}
+    .prominent{padding:4px 9px;font-size:14px;gap:5px}
+    .prominent .material-symbols-outlined{font-size:18px}
+  `]
+})
+export class TestResultBadgeComponent {
+  result = input<CaseTestResult | null | undefined>();
+  executionStatus = input('');
+  prominent = input(false);
+  status = computed(() => this.result()?.status || (['completed', 'success', 'failed', 'cancelled', 'interrupted', 'skipped'].includes(this.executionStatus()) ? 'inconclusive' : 'pending'));
+  label = computed(() => this.status() === 'pending' ? '待判定' : TEST_RESULT_LABELS[this.status() as CaseTestResult['status']]);
+  icon = computed(() => ({passed: 'check_circle', failed: 'cancel', blocked: 'block', inconclusive: 'help', pending: 'hourglass_empty'})[this.status()]);
+}

apps/showcase_ui/src/app/components/test-result-badge/test-result-card.component.spec.ts

@@ -0,0 +1,19 @@
+import { TestBed } from '@angular/core/testing';
+import { TestResultCardComponent } from './test-result-card.component';
+
+describe('用例测试结果卡片', () => {
+  it('显式展示不通过及原因，允许展开证据', () => {
+    const fixture = TestBed.createComponent(TestResultCardComponent);
+    fixture.componentRef.setInput('result', {status: 'failed', reason: '预期英文，实际为中文。', evidence: ['截图显示搜索设置']});
+    fixture.detectChanges();
+    const view: HTMLElement = fixture.nativeElement;
+    expect(view.querySelector('[aria-label="用例测试结果"]')).not.toBeNull();
+    expect(view.querySelector('.prominent')?.textContent).toContain('测试不通过');
+    expect(view.querySelector('.reason')?.textContent).toContain('预期英文，实际为中文。');
+    const details = view.querySelector('details')!;
+    expect(details.open).toBeFalse();
+    details.querySelector('summary')!.click();
+    expect(details.open).toBeTrue();
+    expect(details.textContent).toContain('截图显示搜索设置');
+  });
+});

apps/showcase_ui/src/app/components/test-result-badge/test-result-card.component.ts

@@ -0,0 +1,31 @@
+import { Component, input } from '@angular/core';
+import { CaseTestResult } from '../../core/models/session.model';
+import { TestResultBadgeComponent } from './test-result-badge.component';
+
+@Component({
+  selector: 'app-test-result-card',
+  standalone: true,
+  imports: [TestResultBadgeComponent],
+  template: `<section [class]="'test-result-card ' + result().status" aria-label="用例测试结果" role="status">
+    <header><span class="caption">用例测试结果</span><app-test-result-badge [result]="result()" [prominent]="true" /></header>
+    <p class="reason">{{result().reason}}</p>
+    @if(result().evidence.length){<details><summary>查看验收证据 · {{result().evidence.length}} 条</summary><ol>@for(item of result().evidence; track $index){<li>{{item}}</li>}</ol></details>}
+  </section>`,
+  styles: [`
+    :host{display:block;margin:8px 0}
+    .test-result-card{border:1px solid #cbd5e1;border-left:3px solid #64748b;border-radius:10px;padding:12px 16px;background:#f8fafc;color:#334155}
+    .passed{border-color:#bbf7d0;border-left-color:#15803d;background:#f0fdf4}
+    .failed{border-color:#fecaca;border-left-color:#b91c1c;background:#fef2f2}
+    .inconclusive,.blocked{border-color:#fde68a;border-left-color:#b45309;background:#fffbeb}
+    header{display:flex;align-items:center;justify-content:space-between;gap:8px;flex-wrap:wrap}
+    .caption{font-size:12px;font-weight:600;color:#475569}
+    .reason{font-size:13px;line-height:1.7;margin:8px 0 0;white-space:pre-wrap;overflow-wrap:anywhere;color:#1e293b}
+    details{margin-top:6px;font-size:12px;line-height:1.7;color:#475569}
+    summary{cursor:pointer;width:fit-content;padding:2px 0;font-weight:600}
+    ol{padding-left:22px;margin:8px 0 0}li{padding:3px 0;white-space:pre-wrap;overflow-wrap:anywhere}
+    @media(max-width:480px){.test-result-card{padding:10px 12px}}
+  `]
+})
+export class TestResultCardComponent {
+  result = input.required<CaseTestResult>();
+}

apps/showcase_ui/src/app/core/models/session.model.ts

@@ -21,6 +21,12 @@ export interface ModelInfo {
   architecture?: string;
 }
 
+export interface CaseTestResult {
+  status: 'passed' | 'failed' | 'inconclusive' | 'blocked';
+  reason: string;
+  evidence: string[];
+}
+
 export interface TaskQueueItem {
   session_id: string;
   goal: string;
@@ -41,6 +47,7 @@ export interface Session {
   display_name?: string;
   case_id?: string;
   error?: string;
+  test_result?: CaseTestResult | null;
   video_url?: string;
   recording_status?: 'recording' | 'finalizing' | 'processing' | 'ready' | 'failed' | 'unavailable';
   model_info?: ModelInfo;

apps/showcase_ui/src/app/core/models/stream.model.ts

@@ -203,6 +203,7 @@ export interface CheckerBlockData {
   route?: string;
   error?: string;
   task_status?: 'completed' | 'partial' | 'blocked' | string;
+  test_result?: import('./session.model').CaseTestResult | null;
   tests?: { passed: number; failed: number; inconclusive: number; unchecked: number };
   last_findings?: string[];
   started_at?: number;

artemis/agents/checker/checker.json

@@ -5,5 +5,6 @@
   "assert_semantics": "# `assert` items (test assertions)\n- An `assert` item is a test expectation, not an instruction. A failed assertion is a legitimate, valuable test result: record it truthfully with evidence. Nothing will (or should) try to repair it.\n- Never mark an assert `failed` merely because evidence is absent — absence of captured evidence is `inconclusive` with an explicit note that the evidence is missing.\n- Precondition first: before judging an assert, confirm from the execution history (step replay, notes, anchor evidence) that the action(s) meant to trigger the asserted behavior were actually and correctly performed by the executor (the toggle was really tapped, the intended screen was really reached).\n- If that precondition is not established — the executor never performed the trigger, performed the wrong action, or the evidence cannot show it — output `inconclusive` and state explicitly that the precondition failed / the triggering action is not confirmed as executed. Never output `failed` here: a failed assert is attributed to the device under test, and an execution problem must not be reported as a device defect.\n- Output `failed` only when the precondition is established AND the device state, probe output, or recorded evidence still contradicts the expectation. The evidence must cite the concrete step(s) that performed the trigger plus the probe output / observation that contradicts the expectation, so the verdict reads as a device-defect diagnosis.\n- `suggestion` is not needed for assert verdicts.",
   "anchor_guide": "# Evidence anchoring (checkpoint audit)\n- You are auditing a subgoal at its completion moment. The authoritative evidence is the execution history around the anchor step (the anchor step's pre-action screenshot and the preceding step's post-action state) plus persistently probeable device state.\n- For transient expectations (toasts, dialogs, momentary states) and ordering expectations, you MUST inspect the recorded history near the anchor step — the live device has already moved on and is NOT valid evidence for them.\n- Prefer persistent state probes and recorded notes over inference from summaries.",
   "final_guide": "# Final review (task exit audit)\n- Audit against the user's original goal as amended by the user guidance received during the run (listed in the prompt), not merely the plan's checkboxes. Guidance outranks the plan and its check lines: judge a check line the guidance retired or changed by the guidance, not by its original wording.\n- `@end` items: judge them using the FINAL device state provided to you.\n- `on_complete` items: cite the verdict ledger — an already-recorded PASS/FAIL from the item's checkpoint attempt is authoritative (an assert that ever failed stays failed). For items never checked, judge from the historical evidence near their subgoal's completion; if that evidence was never captured, output `inconclusive` stating the evidence is missing. NEVER re-judge an `on_complete` item from the current screen: the current screen only proves the present, not the past moment the item was declared for.\n- `unmet_subgoals`: list, verbatim, any plan subgoal marked complete whose verify criteria are demonstrably not met. Leave it empty otherwise; failed asserts alone do NOT make a subgoal unmet.",
-  "probe_guide": "# Device probes (read-only)\n- `probe_device` runs a fixed, enumerated set of read-only state probes. Prefer probing persistent state over inferring it: alarms (`alarms`), battery status (`battery`), foreground activities (`foreground`), notifications (`notifications`), a settings value (`setting` with `namespace` + `key`), a content provider query (`content` with `uri`), installed packages (`packages`), a system property (`prop` with `key`).\n- Probes outside this table do not exist; you cannot run arbitrary commands."
+  "probe_guide": "# Device probes (read-only)\n- `probe_device` runs a fixed, enumerated set of read-only state probes. Prefer probing persistent state over inferring it: alarms (`alarms`), battery status (`battery`), foreground activities (`foreground`), notifications (`notifications`), a settings value (`setting` with `namespace` + `key`), a content provider query (`content` with `uri`), installed packages (`packages`), a system property (`prop` with `key`).\n- Probes outside this table do not exist; you cannot run arbitrary commands.",
+  "case_result_guide": "# Overall test result (final review, for Flash and Pro)\nAlways produce `test_result`, even when no plan/check items exist. Infer acceptance criteria directly from the user's original task prompt and subsequent guidance; do not ask for an extra expected-results form. Evaluate explicit expectations and the observable outcome of requested actions. Do not invent requirements, turn examples into expectations, or relax an expectation because the executor says it completed the task.\nCompare the executor's final explanation with actual screenshots, UI elements, step history, notes, probes and the verdict ledger. The explanation is a claim to verify, never proof by itself. A final screen only proves the final state: use recorded history for intermediate/transient results and required sequences. Use read-only tools only. Do not repair a failed test or repeat device actions.\n- passed: all requirements in the prompt/guidance are supported by concrete evidence, and no active assertion failed. Cite the specific step/screenshot/probe observations.\n- failed: the triggering actions and prerequisites are confirmed as correctly executed, but concrete evidence contradicts at least one stated expectation. Cite both the trigger and the contradictory result. A confirmed failure stays failed even when other requirements are unverified.\n- inconclusive: evidence is missing/ambiguous, the task has no discernible acceptance criterion, or the conclusion cannot be checked. Explain which requirement/evidence is missing; never claim passed or failed merely from the report or an absent exception.\n- blocked: an execution or environment obstacle prevented the required actions/validation. This is not evidence of a product defect.\nWrite `reason` in concise Simplified Chinese, explaining the expected and actual result. `evidence` contains concrete observations with step numbers when available. Preserve user-supplied values, UI text and logs exactly.\nThe test result evaluates the system under test, NOT whether the testing agent obeyed the instruction to inspect/report. Explicit expectations (预期, 应当, 必须, should, expected) are test assertions even without plan checkboxes. Reporting a mismatch correctly does not satisfy that expectation. Example: 'Check the search box; expected Search settings; if mismatched report the actual text and finish.' If the box visibly says 搜索设置, output test_result.status=failed, not passed: the inspection completed but the expected product result was not met. 'Report any failures and stop' never converts a detected failure into a pass. Likewise, confirming a missing required entry is a failed test when the correct page was reached. First list the expected product behavior in reason, then compare the actual evidence; judge execution/reporting separately. If no product expectation is given, judge the observable result of the requested actions."
 }

artemis/agents/checker/checker.py

@@ -46,6 +46,7 @@ from pydantic import BaseModel, Field, ValidationError
 
 from artemis.core.tool_failure import ToolFailure, is_tool_failure
 from artemis.constants import CHECKER_MAX_ITERATIONS
+from artemis.runtime.case_result import CaseTestResult, normalize_case_result
 from artemis.context import ArtemisContext
 from artemis.data_engine.context_vars import CURRENT_TRACE_ID
 from artemis.data_engine.trace import trace, trace_langchain_tool
@@ -98,6 +99,10 @@ class CheckVerdict(BaseModel):
 
 class CheckReport(BaseModel):
     verdicts: list[CheckVerdict] = Field(default_factory=list)
+    test_result: CaseTestResult | None = Field(
+        default=None,
+        description="Final review only: overall test verdict against the user's prompt and guidance, independent of execution completion.",
+    )
     unmet_subgoals: list[str] = Field(
         default_factory=list,
         description=(
@@ -287,6 +292,7 @@ def assemble_checker_prompt_segments(
         segments.append(prompts.get("anchor_guide", ""))
     else:
         segments.append(prompts.get("final_guide", ""))
+        segments.append(prompts.get("case_result_guide", ""))
     if probe_tool_registered:
         segments.append(prompts.get("probe_guide", ""))
     return "\n\n".join(s for s in segments if s)
@@ -458,7 +464,11 @@ def _normalize_report(report: CheckReport, check_items: list) -> CheckReport:
                     evidence="no verdict produced for this item",
                 )
             )
-    return CheckReport(verdicts=verdicts, unmet_subgoals=list(report.unmet_subgoals))
+    return CheckReport(
+        verdicts=verdicts,
+        unmet_subgoals=list(report.unmet_subgoals),
+        test_result=normalize_case_result(report.test_result) if report.test_result else None,
+    )
 
 
 async def _run_check_loop(
@@ -796,6 +806,7 @@ async def run_final_check(
     ledger: list[dict],
     check_items,
     attempt_id: str | None = None,
+    execution_report: dict | str | None = None,
 ) -> CheckReport:
     """Final review at task exit: audits the user's original goal and all
     declared check items, with the final screen state available."""
@@ -828,6 +839,8 @@ async def run_final_check(
     human_text = (
         f"# User's Original Goal\n{goal}\n\n"
         f"# User guidance received during the run\n{_format_user_guidance(ctx, steps)}\n\n"
+        f"# Executor's Final Report (claims to verify, not independent evidence)\n"
+        f"{json.dumps(execution_report, ensure_ascii=False, default=str) if execution_report is not None else '(not provided)'}\n\n"
         f"# Final Task Plan\n{plan_text or '(no plan)'}\n\n"
         f"# Declared Check Items\n{_format_check_items(items)}\n\n"
         f"# Verdict Ledger (checkpoint results, authoritative for on_complete"
@@ -852,7 +865,15 @@ async def run_final_check(
     ]
     with CheckerStreamCapture(ctx) as capture:
         try:
-            return await _run_check_loop(ctx, messages, tools, items)
+            report = await _run_check_loop(ctx, messages, tools, items)
+            report.test_result = (
+                normalize_case_result(report.test_result)
+                if report.test_result
+                else CaseTestResult(
+                    status="inconclusive", reason="验收未生成测试结论，不能确认用例是否通过。"
+                )
+            )
+            return report
         finally:
             _persist_stream(
                 ctx, capture, attempt_id=attempt_id, phase="final", checkpoint_id="final"

artemis/agents/checker/flash.py

@@ -0,0 +1,113 @@
+"""Flash's exit-only acceptance check uses the same read-only Checker as Pro."""
+
+import asyncio
+
+from artemis.agents.checker.checker import run_final_check
+from artemis.context import ArtemisContext
+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.utils.logger import get_logger
+
+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."""
+    status = str(execution_report.get("status") or "failed")
+    result = None
+    if status == "completed" or final_report_received:
+        if final_check_enabled(ctx):
+            timeout = float(getattr(ctx.execution_setup, "checkpoint_timeout", 180.0))
+            publish_checker_event(
+                ctx,
+                {
+                    "event": "attempt_started",
+                    "phase": "final",
+                    "attempt_id": "final#1",
+                    "checkpoint_id": "final",
+                    "check_items": [],
+                    "status": "running",
+                },
+            )
+            try:
+                report = await asyncio.wait_for(
+                    run_final_check(
+                        ctx,
+                        goal=goal,
+                        plan_text="",
+                        ledger=[],
+                        check_items=[],
+                        attempt_id="final#1",
+                        execution_report=execution_report,
+                    ),
+                    timeout=timeout,
+                )
+                result = report.test_result
+                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 = (
+                    f"超过 {timeout:g} 秒"
+                    if isinstance(exc, TimeoutError)
+                    else f"{type(exc).__name__}: {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 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"]}
+        )
+    publish_checker_event(ctx, {"event": "run_outcome", "phase": "outcome", **payload})
+    ctx.run_outcome = payload
+    return payload

artemis/agents/flash/runner.py

@@ -145,6 +145,7 @@ class FlashRunner:
     def __init__(self, ctx: ArtemisContext, goal: str, max_turns: int | None = None):
         self.ctx = ctx
         self.goal = goal
+        self.final_report_received = False
         try:
             cfg = load_agent_config()
             self.max_turns = max_turns if max_turns is not None else cfg.flash.max_turns
@@ -570,6 +571,7 @@ class FlashRunner:
     ) -> dict:
         """Records the final step, acknowledges the tool call, and flushes."""
         final_report = args
+        self.final_report_received = True
         if self.ctx.data_engine:
             try:
                 if self.ctx.data_engine.current_step_id is None:

artemis/clients/ui_automator_client.py

@@ -247,6 +247,48 @@ def _ensure_maestro_not_installed(device_id: str) -> None:
         _uninstall_package(device_id, MAESTRO_PACKAGE)
 
 
+class UIAutomatorConnectionError(RuntimeError):
+    """Terminal connection failure with device context and the original cause."""
+
+    def __init__(self, device_id: str, attempts: int, cause: Exception) -> None:
+        # The library appends its whole output buffer on every readiness poll.
+        # Keep the original exception as __cause__, but do not repeat those
+        # identical startup lines dozens of times in the user-facing failure.
+        lines = [line for arg in cause.args for line in str(arg).splitlines()]
+        detail = "\n".join(dict.fromkeys(lines)) or str(cause)
+        super().__init__(
+            f"UIAutomator2 连接失败（设备：{device_id}；已尝试 {attempts} 次）。"
+            f"\n底层原因：{type(cause).__name__}: {detail}"
+        )
+
+
+class _ManagedUIAutomatorDevice(u2.Device):
+    """Release a failed launch before the constructor can lose its handle.
+
+    UIAutomator2 registers exit cleanup only after startup succeeds. Its
+    partially constructed Device is not returned when readiness fails, so
+    the caller cannot stop that launch. Intercept the startup boundary here
+    instead of modifying the installed library or killing device processes.
+    """
+
+    def start_uiautomator(self) -> None:
+        try:
+            super().start_uiautomator()
+        except BaseException:
+            # _process is set only for a server launched by this instance;
+            # attaching to a server already on the device leaves it None.
+            if self._process is not None:
+                try:
+                    super().stop_uiautomator(wait=False)
+                except Exception as cleanup_error:
+                    # Cleanup must never hide the startup error or interrupt.
+                    logger.warning(
+                        f"Failed to stop this UIAutomator2 launch on {self._dev.serial}: "
+                        f"{cleanup_error}"
+                    )
+            raise
+
+
 class UIAutomatorClient:
     """UIAutomator2 client for Android screen data retrieval.
 
@@ -297,7 +339,7 @@ class UIAutomatorClient:
             if attempt:
                 time.sleep(1.0 * attempt)
             try:
-                self._device = u2.connect(self._device_id)
+                self._device = _ManagedUIAutomatorDevice(self._device_id)
                 break
             except Exception as exc:
                 last_error = exc
@@ -307,7 +349,9 @@ class UIAutomatorClient:
                 )
         else:
             self._awake_strategy = None
-            raise last_error
+            raise UIAutomatorConnectionError(
+                self._device_id, attempt + 1, last_error
+            ) from last_error
         logger.info("UIAutomator2 connected successfully")
         return self._device
 

artemis/graph/graph.py

@@ -284,6 +284,7 @@ async def exit_settlement_node(state: State, ctx: ArtemisContext):
     max_attempts = int(getattr(setup, "final_check_max_attempts", 3) or 3) if setup else 3
 
     verify_blocked = False
+    case_result = None
     blocked_findings: list[str] = []
     if final_check_enabled(ctx):
         attempt_no = ctx.final_check_attempts + 1
@@ -320,9 +321,20 @@ async def exit_settlement_node(state: State, ctx: ArtemisContext):
                 ),
                 timeout=checkpoint_timeout,
             )
+            case_result = report.test_result
         except Exception as e:
             # Fail-open: release, but the verdict value stays inconclusive.
             logger.warning(f"Final check errored ({e}); releasing fail-open.")
+            from artemis.runtime.case_result import CaseTestResult
+
+            detail = (
+                f"超过 {checkpoint_timeout:g} 秒"
+                if isinstance(e, TimeoutError)
+                else f"{type(e).__name__}: {e}"
+            )
+            case_result = CaseTestResult(
+                status="inconclusive", reason=f"最终验收未完成：{detail}。"
+            )
             error_verdicts: list[dict] = []
             for ci in snapshot.all_check_items:
                 append_ledger_record(
@@ -444,10 +456,20 @@ async def exit_settlement_node(state: State, ctx: ArtemisContext):
         verify_blocked=verify_blocked,
         retired=retired if isinstance(retired, set) else None,
     )
-    if snapshot.all_check_items or records or verify_blocked:
+    if snapshot.all_check_items or records or verify_blocked or case_result is not None:
-        update["run_outcome"] = outcome.model_dump()
+        from artemis.runtime.case_result import resolve_case_result
+
+        payload = outcome.model_dump()
+        if case_result:
+            payload["test_result"] = case_result.model_dump()
+        resolved = resolve_case_result(
+            payload, "completed" if outcome.task_status == "completed" else "failed"
+        )
+        payload["test_result"] = resolved.model_dump() if resolved else None
+        update["run_outcome"] = payload
         # BLOCKED/partial wrap-ups carry the last findings in the metadata file.
         extra: dict = {"last_findings": blocked_findings} if blocked_findings else {}
+        extra["test_result"] = payload["test_result"]
         # Which UI-hierarchy source served the run, so a report reader can tell a
         # helper run from a UIAutomator2 fallback without reading logs.
         from artemis.clients.screen_client_factory import hierarchy_backend_summary
@@ -461,7 +483,7 @@ async def exit_settlement_node(state: State, ctx: ArtemisContext):
             {
                 "event": "run_outcome",
                 "phase": "outcome",
-                **outcome.model_dump(),
+                **payload,
                 "last_findings": list(blocked_findings),
             },
         )

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

artemis/runtime/task_result.py

@@ -6,6 +6,7 @@ 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:
@@ -37,10 +38,17 @@ def save_failure_result(session_id: str, error: str, *, code: str = "execution_f
     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 / "result.json").read_text(encoding="utf-8"))
+            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(
@@ -54,6 +62,8 @@ def save_failure_result(session_id: str, error: str, *, code: str = "execution_f
             "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")

artemis/sdk/agent.py

@@ -701,7 +701,16 @@ class Agent:
                             runner = FlashRunner(context, goal=request.goal, max_turns=max_turns)
                             flash_result = await runner.run(state)
 
-                            output = flash_result
+                            from artemis.agents.checker.flash import settle_flash_outcome
+
+                            run_outcome = await settle_flash_outcome(
+                                context,
+                                request.goal,
+                                flash_result,
+                                final_report_received=runner.final_report_received,
+                            )
+                            output = attach_test_summary(flash_result, run_outcome)
+                            state.run_outcome = run_outcome
                             last_state_snapshot = state.model_dump()
 
                             status = flash_result.get("status")
@@ -1545,17 +1554,21 @@ def attach_test_summary(output, run_outcome: dict | None):
         + int(tests.get("inconclusive", 0))
         + int(tests.get("unchecked", 0))
     )
-    if total <= 0:
+    test_result = run_outcome.get("test_result")
+    if total <= 0 and not test_result:
         return output
     summary = {"task_status": run_outcome.get("task_status"), **tests}
+    additions = {"test_summary": summary} if total > 0 else {}
+    if test_result:
+        additions["test_result"] = test_result
     if isinstance(output, dict):
         merged = dict(output)
-        merged.setdefault("test_summary", summary)
+        merged.update(additions)
         return merged
     if output is None:
-        return {"test_summary": summary}
+        return additions
     if isinstance(output, str):
-        return {"result": output, "test_summary": summary}
+        return {"result": output, **additions}
     # Typed/structured outputs keep their shape; the summary stays available in
     # run_outcome.json and the session state.
     return output
@@ -1566,6 +1579,9 @@ def resolve_trace_suffix(task_status: str, run_outcome: dict | None) -> str:
     if task_status != "completed":
         return "_FAIL"
     tests = (run_outcome or {}).get("tests") or {}
-    if int(tests.get("failed", 0)) > 0:
+    if (
+        int(tests.get("failed", 0)) > 0
+        or ((run_outcome or {}).get("test_result") or {}).get("status") == "failed"
+    ):
         return "_TESTFAIL"
     return "_PASS"

docs/ai-automation.md

@@ -52,6 +52,10 @@ MCP 同步仅检查 `$CODEX_HOME/config.toml`；未设置或为空时使用当
 
 执行保护默认总时限 1800 秒，单任务、批次和 MCP 都适用，排队及用户暂停不计时。设备观测默认 60 秒、最多 3 次，ADB 命令默认 10 秒；观测失败不会继续向模型发送占位图或旧元素索引。Flash 连续 3 次动作失败先重新观察再切换兜底，兜底仍失败则结束任务并保存原因。不能以提高步数或反复重提任务代替设备诊断。
 
+UIAutomator2 启动失败时，先释放本次连接启动的服务，再按现有策略重试；最终错误保留设备编号、尝试次数和底层原因。初始化失败且没有操作步骤时，用例目标属于未执行、未验证，应先检查设备诊断和启动日志，不能据此认定被测功能失败。
+
+Flash 和 Pro 默认复用只读 Checker，以任务提示词和用户指导中的要求核对执行结论、步骤截图及探测证据，无需另填预期结果。`test_result` 包含 `status`（`passed`、`failed`、`inconclusive`、`blocked`）、中文 `reason` 和 `evidence`；任务列表、报告与 MCP 使用同一结论。`completed` 仍只代表执行结束，可同时存在测试不通过。Flash 只在退出时验收，不添加计划或中途检查；沿用 Checker 的超时和开关，关闭、超时或缺少证据时显示未验证，不重跑用例。
+
 `mobile_run_task(runtime_overrides={"timeout_seconds":1800,"observation_timeout_seconds":60,"observation_max_attempts":3,"device_command_timeout_seconds":10,"action_failure_attempts":3})` 可为单任务覆盖执行保护；其中 `model` 是 API 模型名，外层 `model` 仍是 Flash/Pro 模式。相同字段也支持批次的 `task_config`。全局默认通过网页功能配置或明确授权的 `mobile_config(section="capabilities", changes={"task_timeout_seconds":1800}, persist_defaults=true)` 保存，对后续任务生效，不接受密钥。
 
 任务已经产生最终结果，但 SDK 线程或资源清理仍卡住时，默认最多再等待 30 秒后回收工作进程，避免继续占用队列。全局可用 `ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS` 或功能配置的 `worker_cleanup_timeout_seconds` 调整，任务 JSON 同名字段可覆盖。原有结果与截图保留，未完成保存的录屏尽力恢复。

docs/automation-workspace.md

@@ -8,7 +8,7 @@
 
 报告包含目标、预期结果、执行状态、验证结论、实际输出和步骤证据。Word 内嵌关键截图，Excel 包含结果与证据工作表；任务输出中的 Markdown 表格会转换为真正的文档表格和 Excel 工作表。
 
-“执行完成”与“验证通过”分开记录。没有实际检查记录时显示“未验证”，不会把模型结束执行当作业务测试通过。正在执行的任务也可以下载，但文件只代表下载时的快照。
+“执行完成”与测试结论分开记录。预期直接写在任务提示词中即可；Flash 和 Pro 默认用只读终检核对提示词、用户指导、执行结论及实际证据，不增加必填表单。任务列表显式标注“测试通过 / 测试不通过 / 未验证 / 测试阻塞”，结果区域和 Word/Excel 同时列出判定依据及证据。终检关闭、出错或缺少证据时显示“未验证”，设备或执行异常不当作产品缺陷；已确认的断言失败会保留。正在执行的任务也可以下载，但文件只代表下载时的快照。
 
 报告由项目自己的 `python-docx`、`openpyxl` 生成，不依赖 Codex 客户端的私有工具。输出保存在 `artifacts/reports/`。
 

mcp_server/tools/task_manager.py

@@ -271,7 +271,9 @@ def mobile_manage_task(
       setups), `task_desc`, `model`, `elapsed_seconds`, a `test_summary` (when
       the run declared verification check items: machine-readable
       passed/failed/inconclusive/unchecked counts plus failed-item details —
-      no need to parse report prose), and `progress` (Flash:
+      no need to parse report prose), `test_result` (Flash and Pro: status
+      passed/failed/inconclusive/blocked, reason and evidence; completed only
+      means execution ended), and `progress` (Flash:
       current turn, latest thought/action — intervene if turns climb without
       progress or the thought indicates it is stuck; Pro: the active task plan
       — check it still aligns with your goal).
@@ -419,11 +421,23 @@ def mobile_manage_task(
                         "task_status": run_outcome.get("task_status"),
                         **tests,
                     }
+                    from artemis.runtime.case_result import resolve_case_result
+
+                    test_result = resolve_case_result(
+                        run_outcome, current_status, response.get("error") or ""
+                    )
+                    response["test_result"] = test_result.model_dump() if test_result else None
                 except (OSError, ValueError, TypeError):
                     # Unreadable or malformed run_outcome.json: omit the summary.
                     pass
                 break
 
+        if "test_result" not in response:
+            from artemis.runtime.case_result import resolve_case_result
+
+            test_result = resolve_case_result(None, current_status, response.get("error") or "")
+            response["test_result"] = test_result.model_dump() if test_result else None
+
         if current_status in ("running", "pending", "task_paused", "resuming") and is_alive:
             progress: dict[str, Any] = {}
 

mcp_server/tools/task_runner.py

@@ -279,7 +279,9 @@ def mobile_run_task(
       UI path. No step cap by default (history is compressed, and the agent
       can recall earlier steps and query the session recording), but no ADB
       shell, no pre-execution safety net, no persistent plan or notes, and
-      no verification or written report — unsuitable for exploration,
+      no midway checkpoints or persistent written notes. An exit-only read-only
+      Checker judges the prompt's expectations and returns test_result;
+      execution completion is separate from test pass. Unsuitable for exploration,
       multi-branch recovery, structured monitoring/polling, or tasks that
       must report back large amounts of detail.
     - **Pro**: complex, exploratory, or multi-branch tasks; continuous

pyproject.toml

@@ -14,7 +14,7 @@
 
 [project]
 name = "artemis"
-version = "1.4.0"
+version = "1.5.0"
 description = "Autonomous AI Assistant for Mobile Automation"
 
 authors = [

skills/artemis-automation/SKILL.md

@@ -22,7 +22,7 @@ Windows 启动脚本只提示源码更新。升级使用 `update.bat`（国内
 
 ## 结果、证据与交付
 
-- 执行状态与业务判定分开：`completed` 只说明执行结束，不等于测试通过。检查工具返回的验证摘要、预期/实际结果，以及 `mobile_inspect_trace` 的步骤/截图。证据不足标记“未验证”；前置条件不满足说明阻塞原因，不能编造通过结论。
+- 执行状态与业务判定分开：`completed` 只说明执行结束，不等于测试通过。Flash 和 Pro 默认根据任务提示词、用户指导、执行结论和实际证据进行只读终检，不需要另填预期结果。读取 `test_result.status`（`passed` 通过、`failed` 不通过、`inconclusive` 未验证、`blocked` 阻塞）、`reason` 与 `evidence`，结合验证摘要和 `mobile_inspect_trace` 的步骤/截图交付结果。执行失败不能直接认定产品缺陷；已确认的断言失败不得被后续成功覆盖。终检关闭、出错或证据不足时不得宣称通过。
 - 用 `mobile_export_report(session_id=...)` 获取单任务报告；`formats=["screenshots"]` 获取排序截图，`["video"]` 获取保存的录屏分段，`["all"]` 获取 Word、Excel 和所有可用资料的 ZIP。批次传 `batch_id` 获取汇总、各项任务报告、步骤截图及已有录屏的 ZIP。未指定格式时使用保存的 Word/Excel 偏好。执行中的导出只是当前快照。
 - 工具返回 MCP 主机上的绝对文件路径和下载 URL。向用户交付真实生成的文件链接，并说明完成、失败、未验证、跳过数量及重要限制。不要只说“报告已生成”而不交付文件。
 - 用 `mobile_search_history` 按名称、目标、编号或标签搜索全部历史；日期先按用户时区转换为 Unix 秒，`start` 含边界、`end` 不含边界。按返回的 `total/page/size` 翻页，不能把第一页当作全部历史。

tests/unit/admin_console/test_automation.py

@@ -780,7 +780,9 @@ def test_failed_report_exposes_logged_reason_and_formats_chinese_markdown(isolat
     )
     report = reports.session_report(job["session_id"])
     assert "连续 3 次" in report["error"]
-    assert report["verdict"] == "未验证（任务异常终止）"
+    assert report["verdict"] == "测试阻塞"
+    assert report["test_result"]["status"] == "blocked"
+    assert "连续 3 次" in report["test_result"]["reason"]
     assert "abc123secret" not in reports.failure_reason(
         folder, "failed", "认证失败 OPENAI_API_KEY=abc123secret Authorization: Bearer abc123secret"
     )
@@ -796,7 +798,10 @@ def test_failed_report_exposes_logged_reason_and_formats_chinese_markdown(isolat
         cell.text == "分类" for table in doc.tables for row in table.rows for cell in row.cells
     )
     wb = load_workbook(next(path for path in paths if path.suffix == ".xlsx"))
-    assert "连续 3 次" in wb["任务结果"]["B15"].value
+    assert any(
+        row[0].value == "异常原因" and "连续 3 次" in row[1].value
+        for row in wb["任务结果"].iter_rows()
+    )
     assert wb["结论"]["B2"].value == "结果"
     assert wb["结果表1"]["A2"].value == "系统"
     assert wb["执行证据"]["A1"].font.name == "微软雅黑"
@@ -810,6 +815,51 @@ def test_failed_report_exposes_logged_reason_and_formats_chinese_markdown(isolat
     ]
 
 
+def test_prompt_based_verdict_is_consistent_in_history_batch_and_reports(
+    client, isolated, monkeypatch
+):
+    from artemis.runtime import case_result
+
+    monkeypatch.setattr(case_result, "TRACES_PATH", reports.TRACES_PATH)
+    batch = service.create_batch(payload())
+    job = isolated.jobs(batch["id"])[0]
+    isolated.save_job(
+        {**job, "status": "completed", "started_at": time.time() - 50, "ended_at": time.time()}
+    )
+    result = {
+        "status": "failed",
+        "reason": "预期出现成功提示，实际显示错误",
+        "evidence": ["步骤 3 点击后截图显示错误提示"],
+    }
+    folder = reports.TRACES_PATH / job["session_id"]
+    folder.mkdir(parents=True)
+    (folder / "run_outcome.json").write_text(
+        json.dumps(
+            {"task_status": "completed", "tests": {}, "test_result": result}, ensure_ascii=False
+        ),
+        encoding="utf-8",
+    )
+    history = client.get("/api/automation/history").json()["items"][0]
+    batch_job = client.get(f"/api/automation/batches/{batch['id']}").json()["jobs"][0]
+    report = client.get(f"/api/automation/reports/{job['session_id']}").json()
+    for item in (history, batch_job, report):
+        assert item["status"] == "completed"
+        assert item["test_result"] == result
+    assert report["verdict"] == "测试不通过"
+    paths = reports.export_session(job["session_id"])
+    doc = Document(next(path for path in paths if path.suffix == ".docx"))
+    doc_text = "\n".join(
+        cell.text for table in doc.tables for row in table.rows for cell in row.cells
+    )
+    doc_text += "\n" + "\n".join(paragraph.text for paragraph in doc.paragraphs)
+    assert result["reason"] in doc_text
+    assert result["evidence"][0] in doc_text
+    wb = load_workbook(next(path for path in paths if path.suffix == ".xlsx"))
+    values = [cell.value for row in wb["任务结果"].iter_rows() for cell in row]
+    assert "测试不通过" in values
+    assert result["reason"] in values
+
+
 def test_task_material_zip_contains_ordered_evidence_and_reports(
     client, isolated, tmp_path, monkeypatch
 ):

tests/unit/agents/test_flash_acceptance.py

@@ -0,0 +1,118 @@
+import asyncio
+import json
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from artemis.agents.checker import flash
+from artemis.agents.checker.checker import CheckReport
+from artemis.context import ArtemisContext, ExecutionSetup
+from artemis.runtime.case_result import CaseTestResult
+
+
+def context(tmp_path, **setup):
+    ctx = MagicMock(spec=ArtemisContext)
+    ctx.execution_setup = ExecutionSetup(**setup)
+    ctx.data_engine = MagicMock()
+    ctx.data_engine.base_dir = tmp_path
+    return ctx
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("verdict", ["passed", "failed", "inconclusive"])
+async def test_flash_reuses_checker_with_original_prompt_and_final_report(
+    tmp_path, monkeypatch, verdict
+):
+    result = CaseTestResult(
+        status=verdict, reason="提示词预期与实际对比", evidence=["步骤 3 截图显示实际状态"]
+    )
+    check = AsyncMock(return_value=CheckReport(test_result=result))
+    monkeypatch.setattr(flash, "run_final_check", check)
+    ctx = context(tmp_path)
+    execution = {"status": "completed", "explanation": "操作已执行"}
+    data = await flash.settle_flash_outcome(ctx, "打开设置，预期看到应用入口", execution)
+    assert check.call_count == 1
+    assert check.call_args.kwargs["goal"] == "打开设置，预期看到应用入口"
+    assert check.call_args.kwargs["execution_report"] == execution
+    assert check.call_args.kwargs["check_items"] == []
+    assert data["task_status"] == "completed"
+    assert data["test_result"]["status"] == verdict
+    assert (
+        json.loads((tmp_path / "run_outcome.json").read_text(encoding="utf-8"))["test_result"]
+        == data["test_result"]
+    )
+    assert ctx.run_outcome == data
+
+
+@pytest.mark.asyncio
+async def test_acceptance_timeout_finishes_with_unknown_verdict(tmp_path, monkeypatch):
+    async def hanging_check(*args, **kwargs):
+        await asyncio.Event().wait()
+
+    monkeypatch.setattr(flash, "run_final_check", hanging_check)
+    data = await flash.settle_flash_outcome(
+        context(tmp_path, checkpoint_timeout=0.01), "任务", {"status": "completed"}
+    )
+    assert data["task_status"] == "completed"
+    assert data["test_result"]["status"] == "inconclusive"
+    assert "0.01 秒" in data["test_result"]["reason"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("setup", [{"disable_checker": True}, {"disable_final_check": True}])
+async def test_disabled_acceptance_does_not_call_a_model_or_claim_pass(
+    tmp_path, monkeypatch, setup
+):
+    check = AsyncMock()
+    monkeypatch.setattr(flash, "run_final_check", check)
+    data = await flash.settle_flash_outcome(
+        context(tmp_path, **setup), "任务", {"status": "completed"}
+    )
+    check.assert_not_called()
+    assert data["test_result"]["status"] == "inconclusive"
+
+
+@pytest.mark.asyncio
+async def test_execution_failure_is_blocked_without_another_model_retry(tmp_path, monkeypatch):
+    check = AsyncMock()
+    monkeypatch.setattr(flash, "run_final_check", check)
+    data = await flash.settle_flash_outcome(
+        context(tmp_path), "任务", {"status": "failed", "explanation": "设备观测失败"}
+    )
+    check.assert_not_called()
+    assert data["test_result"]["status"] == "blocked"
+    assert data["test_result"]["reason"] == "设备观测失败"
+
+
+@pytest.mark.asyncio
+async def test_reported_expectation_mismatch_is_checked_even_when_executor_reports_failed(
+    tmp_path, monkeypatch
+):
+    check = AsyncMock(
+        return_value=CheckReport(
+            test_result=CaseTestResult(
+                status="failed",
+                reason="预期英文 Search settings，实际中文 搜索设置。",
+                evidence=["设置首页截图和 UI 元素 1 显示 搜索设置"],
+            )
+        )
+    )
+    monkeypatch.setattr(flash, "run_final_check", check)
+    data = await flash.settle_flash_outcome(
+        context(tmp_path),
+        "预期搜索框为英文 Search settings",
+        {"status": "failed", "explanation": "已检查，实际为中文"},
+        final_report_received=True,
+    )
+    check.assert_awaited_once()
+    assert data["test_result"]["status"] == "failed"
+    assert "Search settings" in data["test_result"]["reason"]
+
+
+@pytest.mark.asyncio
+async def test_cancellation_of_acceptance_propagates(tmp_path, monkeypatch):
+    check = AsyncMock(side_effect=asyncio.CancelledError)
+    monkeypatch.setattr(flash, "run_final_check", check)
+    with pytest.raises(asyncio.CancelledError):
+        await flash.settle_flash_outcome(context(tmp_path), "任务", {"status": "completed"})
+    assert not (tmp_path / "run_outcome.json").exists()

tests/unit/clients/test_ui_automator_startup.py

@@ -0,0 +1,250 @@
+"""UIAutomator startup recovery, using the real library with isolated device IO."""
+
+from dataclasses import dataclass, field
+import json
+import threading
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+from uiautomator2 import base, core
+
+from artemis.clients import ui_automator_client
+from artemis.clients.accessibility_client import HelperUnavailable
+from artemis.clients.screen_client_factory import FallbackScreenClient
+
+
+@dataclass
+class Clock:
+    now: float = 0.0
+
+    def time(self):
+        return self.now
+
+    def sleep(self, seconds):
+        self.now += seconds
+
+
+class DeviceSocket:
+    def __init__(self, mode):
+        self.mode = mode
+        self.closed = threading.Event()
+        self.started = threading.Event()
+
+    def recv(self, _size):
+        if not self.started.is_set():
+            self.started.set()
+            output = b"[server] INFO: [UiAutomator2Server] Starting Server\n"
+            if self.mode == "conflict":
+                output += (
+                    b"java.lang.IllegalStateException: UiAutomationService already registered!"
+                )
+            return output
+        self.closed.wait()
+        return b""
+
+
+class DeviceConnection:
+    def __init__(self, mode):
+        self.conn = DeviceSocket(mode)
+        self.close_calls = 0
+        self.close_error = None
+
+    def close(self):
+        self.close_calls += 1
+        if self.close_error:
+            raise self.close_error
+        self.conn.closed.set()
+
+
+@dataclass
+class DeviceServer:
+    modes: list[str] = field(default_factory=lambda: ["not_ready"])
+    connections: list[DeviceConnection] = field(default_factory=list)
+    previous_closed_at_launch: list[bool] = field(default_factory=list)
+    clock: Clock = field(default_factory=Clock)
+    cleanup_callbacks: MagicMock = field(default_factory=MagicMock)
+    close_error: Exception | None = None
+    probe_error: BaseException | None = None
+
+    def shell(self, command, stream=False):
+        assert stream is True
+        assert "com.wetest.uia2.Main" in command
+        self.previous_closed_at_launch.append(all(c.conn.closed.is_set() for c in self.connections))
+        index = min(len(self.connections), len(self.modes) - 1)
+        connection = DeviceConnection(self.modes[index])
+        connection.close_error = self.close_error
+        self.connections.append(connection)
+        return connection
+
+    def http_request(self, _device, _port, method, path, *args, **kwargs):
+        if self.connections:
+            latest = self.connections[-1]
+            assert latest.conn.started.wait(timeout=2)
+            if self.probe_error is not None:
+                raise self.probe_error
+            if latest.conn.mode == "ready" and not latest.conn.closed.is_set():
+                if method == "POST":
+                    return core.HTTPResponse(
+                        json.dumps({"result": {"displayWidth": 1080}}).encode()
+                    )
+                return core.HTTPResponse(b"pong")
+        raise core.HTTPError("simulated device server is not ready")
+
+
+@pytest.fixture
+def server(monkeypatch):
+    simulated = DeviceServer()
+    device = SimpleNamespace(serial="isolated-startup-device", shell=simulated.shell)
+    monkeypatch.setattr(base._BaseClient, "_wait_for_device", lambda _self: device)
+    monkeypatch.setattr(core.BasicUiautomatorServer, "_setup_jar", lambda _self: None)
+    monkeypatch.setattr(core, "_http_request", simulated.http_request)
+    monkeypatch.setattr(core, "time", simulated.clock)
+    monkeypatch.setattr(core, "atexit", SimpleNamespace(register=simulated.cleanup_callbacks))
+    monkeypatch.setattr(ui_automator_client, "time", simulated.clock)
+    monkeypatch.setattr(ui_automator_client, "_ensure_maestro_not_installed", lambda _s: None)
+    monkeypatch.setattr(ui_automator_client, "ensure_device_awake", lambda _s: "simulated")
+    yield simulated
+    # Close fake sockets even when a regression leaves a real library reader alive.
+    for connection in simulated.connections:
+        connection.conn.closed.set()
+
+
+def test_each_failed_launch_is_closed_before_retry_and_final_failure(server):
+    client = ui_automator_client.UIAutomatorClient("isolated-startup-device")
+
+    with pytest.raises(RuntimeError, match="server not ready") as caught:
+        client.connect()
+
+    assert len(server.connections) == 3
+    assert server.previous_closed_at_launch == [True, True, True]
+    assert [c.close_calls for c in server.connections] == [1, 1, 1]
+    assert server.clock.now == 93.0
+    assert server.cleanup_callbacks.call_count == 0
+    assert client._device is None
+    assert client._awake_strategy is None
+    assert isinstance(caught.value.__cause__, core.LaunchUiAutomationError)
+    assert "isolated-startup-device" in str(caught.value)
+    assert "3" in str(caught.value)
+
+
+def test_retry_can_connect_after_failed_launch_was_stopped(server):
+    server.modes = ["not_ready", "ready"]
+    client = ui_automator_client.UIAutomatorClient("isolated-startup-device")
+
+    client.connect()
+    client.connect()  # Cached device info also uses the real library RPC path.
+
+    assert len(server.connections) == 2
+    assert server.previous_closed_at_launch == [True, True]
+    assert server.connections[0].close_calls == 1
+    assert server.connections[1].close_calls == 0
+    assert server.cleanup_callbacks.call_count == 1
+    client.disconnect(stop_server=True)
+    assert server.connections[1].close_calls == 1
+
+
+def test_later_connection_can_succeed_after_all_previous_attempts_failed(server):
+    server.modes = ["not_ready", "not_ready", "not_ready", "ready"]
+    client = ui_automator_client.UIAutomatorClient("isolated-startup-device")
+
+    with pytest.raises(RuntimeError, match="server not ready"):
+        client.connect()
+    client.connect()
+
+    assert len(server.connections) == 4
+    assert server.previous_closed_at_launch == [True, True, True, True]
+    assert [c.close_calls for c in server.connections] == [1, 1, 1, 0]
+    client.disconnect(stop_server=True)
+    assert server.connections[-1].close_calls == 1
+
+
+def test_auto_fallback_cleans_failed_launches_before_next_attempt(server):
+    helper = MagicMock()
+    helper.connect.side_effect = HelperUnavailable("simulated helper unavailable")
+    client = FallbackScreenClient(
+        "isolated-startup-device",
+        helper=helper,
+        uiautomator_factory=lambda: ui_automator_client.UIAutomatorClient(
+            "isolated-startup-device"
+        ),
+        state_probe=lambda _serial: "device",
+    )
+
+    with pytest.raises(RuntimeError, match="server not ready"):
+        client.connect()
+    client.disconnect()
+
+    assert len(server.connections) == 3
+    assert server.previous_closed_at_launch == [True, True, True]
+    assert [c.close_calls for c in server.connections] == [1, 1, 1]
+    helper.disconnect.assert_called_once()
+
+
+def test_service_conflict_closes_own_streams_and_preserves_cause(server):
+    server.modes = ["conflict"]
+    client = ui_automator_client.UIAutomatorClient("isolated-startup-device")
+
+    with pytest.raises(RuntimeError, match="already registered") as caught:
+        client.connect()
+
+    assert isinstance(caught.value.__cause__, core.AccessibilityServiceAlreadyRegisteredError)
+    assert len(server.connections) == 3
+    assert [c.close_calls for c in server.connections] == [1, 1, 1]
+
+
+def test_cleanup_error_does_not_replace_original_startup_failure(server):
+    server.close_error = OSError("simulated cleanup connection error")
+    client = ui_automator_client.UIAutomatorClient("isolated-startup-device")
+
+    with pytest.raises(RuntimeError, match="server not ready") as caught:
+        client.connect()
+
+    assert isinstance(caught.value.__cause__, core.LaunchUiAutomationError)
+    assert [c.close_calls for c in server.connections] == [1, 1, 1]
+
+
+def test_interrupt_during_startup_also_closes_own_stream(server):
+    server.probe_error = KeyboardInterrupt()
+    client = ui_automator_client.UIAutomatorClient("isolated-startup-device")
+
+    with pytest.raises(KeyboardInterrupt):
+        client.connect()
+
+    assert len(server.connections) == 1
+    assert server.connections[0].close_calls == 1
+
+
+def test_failure_before_launch_does_not_stop_another_device_service(server, monkeypatch):
+    failure = OSError("simulated jar transfer failure")
+
+    def fail_before_launch(_self):
+        raise failure
+
+    monkeypatch.setattr(core.BasicUiautomatorServer, "_setup_jar", fail_before_launch)
+    stop = MagicMock()
+    monkeypatch.setattr(core.BasicUiautomatorServer, "stop_uiautomator", stop)
+    client = ui_automator_client.UIAutomatorClient("isolated-startup-device")
+
+    with pytest.raises(RuntimeError, match="jar transfer failure") as caught:
+        client.connect()
+
+    assert caught.value.__cause__ is failure
+    assert server.connections == []
+    stop.assert_not_called()
+
+
+def test_default_helper_success_does_not_start_uiautomator(server):
+    factory = MagicMock()
+    client = FallbackScreenClient(
+        "isolated-startup-device",
+        helper=MagicMock(),
+        uiautomator_factory=factory,
+        state_probe=lambda _serial: "device",
+    )
+
+    client.connect()
+    client.disconnect()
+
+    factory.assert_not_called()
+    assert server.connections == []

tests/unit/test_case_result.py

@@ -0,0 +1,132 @@
+import json
+
+import pytest
+
+from artemis.runtime.case_result import (
+    CaseTestResult,
+    normalize_case_result,
+    read_case_result,
+    resolve_case_result,
+)
+from artemis.sdk.agent import attach_test_summary, resolve_trace_suffix
+
+
+def outcome(status="passed", **extra):
+    return {
+        "task_status": "completed",
+        "test_result": {
+            "status": status,
+            "reason": "预期与实际对比",
+            "evidence": ["步骤 2 截图显示实际结果"],
+        },
+        **extra,
+    }
+
+
+@pytest.mark.parametrize("status", ["passed", "failed"])
+def test_a_claim_without_evidence_is_never_a_verdict(status):
+    result = normalize_case_result(
+        CaseTestResult(status=status, reason="执行者说完成了", evidence=[" "])
+    )
+    assert result.status == "inconclusive"
+    assert "证据" in result.reason
+
+
+def test_completed_execution_and_failed_test_are_independent():
+    result = resolve_case_result(outcome("failed"), "completed")
+    assert result.status == "failed"
+    returned = attach_test_summary(
+        {"status": "completed", "explanation": "已完成", "test_result": {"status": "passed"}},
+        outcome("failed"),
+    )
+    assert returned["status"] == "completed"
+    assert returned["test_result"]["status"] == "failed"
+    assert resolve_trace_suffix("completed", outcome("failed")) == "_TESTFAIL"
+
+
+@pytest.mark.parametrize("status", ["failed", "cancelled", "interrupted", "skipped"])
+def test_incomplete_execution_does_not_inherit_a_pass(status):
+    result = resolve_case_result(outcome(), status, "环境初始化失败")
+    assert result.status == "blocked"
+    assert result.reason == "环境初始化失败"
+
+
+def test_a_previously_confirmed_assertion_failure_survives_interruption_and_later_pass():
+    data = outcome(
+        tests={
+            "failed": 1,
+            "failed_items": [
+                {
+                    "kind": "assert",
+                    "item_text": "预期显示成功提示",
+                    "evidence": "步骤 4 正确点击后显示错误提示",
+                }
+            ],
+        }
+    )
+    result = resolve_case_result(data, "interrupted", "设备掉线")
+    assert result.status == "failed"
+    assert result.evidence == ["步骤 4 正确点击后显示错误提示"]
+
+
+def test_navigation_failure_is_not_reported_as_a_product_defect():
+    result = resolve_case_result(
+        {
+            "tests": {
+                "failed": 1,
+                "failed_items": [
+                    {"kind": "verify", "item_text": "打开设置", "evidence": "页面未打开"}
+                ],
+            }
+        },
+        "failed",
+    )
+    assert result.status == "blocked"
+
+
+def test_legacy_completed_flash_without_verification_is_inconclusive():
+    assert resolve_case_result({}, "completed").status == "inconclusive"
+    assert resolve_case_result(None, "running") is None
+    assert resolve_case_result({"tests": {"passed": 2}}, "completed").status == "passed"
+
+
+def test_missing_checks_prevent_an_overall_pass():
+    assert (
+        resolve_case_result(outcome(tests={"passed": 1, "unchecked": 1}), "completed").status
+        == "inconclusive"
+    )
+
+
+def test_session_reader_handles_missing_and_corrupt_outcomes_without_claiming_pass(tmp_path):
+    assert read_case_result("case", "completed", root=tmp_path)["status"] == "inconclusive"
+    directory = tmp_path / "case"
+    directory.mkdir()
+    (directory / "run_outcome.json").write_text("{", encoding="utf-8")
+    assert read_case_result("case", "completed", root=tmp_path)["status"] == "inconclusive"
+    (directory / "run_outcome.json").write_text(json.dumps(outcome("failed")), encoding="utf-8")
+    assert read_case_result("case", "completed", root=tmp_path)["status"] == "failed"
+    with pytest.raises(ValueError):
+        read_case_result("../case", "completed", root=tmp_path)
+
+
+def test_failure_persistence_keeps_confirmed_test_failure_and_checkpoint_evidence(
+    tmp_path, monkeypatch
+):
+    from artemis.runtime import task_result
+
+    monkeypatch.setattr(task_result, "TRACES_PATH", tmp_path)
+    directory = tmp_path / "case"
+    directory.mkdir()
+    (directory / "result.json").write_text(
+        json.dumps(outcome("failed", tests={"checkpoint": {"passed": False}})),
+        encoding="utf-8",
+    )
+    (directory / "run_outcome.json").write_text(
+        json.dumps(outcome("passed", tests={"passed": 1})), encoding="utf-8"
+    )
+    task_result.save_failure_result("case", "任务超时")
+    for name in ("result.json", "run_outcome.json"):
+        saved = json.loads((directory / name).read_text(encoding="utf-8"))
+        assert saved["test_result"]["status"] == "failed"
+        assert saved["tests"]["checkpoint"]["passed"] is False
+        assert saved["test_result"]["evidence"] == ["步骤 2 截图显示实际结果"]

tests/unit/test_ui_automator_client_awake.py

@@ -6,7 +6,7 @@ from artemis.clients.ui_automator_client import UIAutomatorClient
 from artemis.drivers.android.adb_driver import AndroidAdbDriver
 
 
-@patch("artemis.clients.ui_automator_client.u2.connect")
+@patch("artemis.clients.ui_automator_client._ManagedUIAutomatorDevice")
 @patch(
     "artemis.clients.ui_automator_client.ensure_device_awake",
     return_value="host_heartbeat",
@@ -25,7 +25,10 @@ def test_new_ui_connection_enrolls_device_in_shared_awake_strategy(
     mock_connect.assert_called_once_with("device-123")
 
 
-@patch("artemis.clients.ui_automator_client.u2.connect", side_effect=RuntimeError("offline"))
+@patch(
+    "artemis.clients.ui_automator_client._ManagedUIAutomatorDevice",
+    side_effect=RuntimeError("offline"),
+)
 @patch(
     "artemis.clients.ui_automator_client.ensure_device_awake",
     return_value="host_heartbeat",
@@ -43,7 +46,7 @@ def test_failed_ui_connection_does_not_stop_process_awake_service(
     assert client._awake_strategy is None
 
 
-@patch("artemis.clients.ui_automator_client.u2.connect")
+@patch("artemis.clients.ui_automator_client._ManagedUIAutomatorDevice")
 @patch(
     "artemis.clients.ui_automator_client.ensure_device_awake",
     return_value="host_heartbeat",

uv.lock

@@ -206,7 +206,7 @@ wheels = [
 
 [[package]]
 name = "artemis"
-version = "1.4.0"
+version = "1.5.0"
 source = { editable = "." }
 dependencies = [
     { name = "adbutils" },