From 32441e79c4583cc4e2dac2d177f503b34c71a8e6 Mon Sep 17 00:00:00 2001 From: chenmingxuan Date: Sun, 18 Oct 2026 20:51:19 +0800 Subject: [PATCH] fix: give final acceptance a shared retry budget and diagnostics --- .env.example | 4 +- .../services/request_diagnostics.py | 30 +++ .../llm-request-status.component.html | 33 ++- .../llm-request-status.component.spec.ts | 42 ++- .../llm-request-status.component.ts | 34 ++- artemis/agents/checker/checker.py | 5 + artemis/agents/checker/flash.py | 15 +- artemis/data_engine/storage.py | 2 +- artemis/graph/graph.py | 16 +- artemis/llm/diagnostics.py | 31 +++ artemis/llm/final_review.py | 171 ++++++++++++ artemis/services/llm.py | 41 ++- docs/automation-workspace.md | 4 + skills/artemis-use/references/diagnostics.md | 1 + tests/unit/agents/test_flash_acceptance.py | 22 +- tests/unit/graph/test_exit_settlement.py | 66 +++++ tests/unit/test_final_review_budget.py | 245 ++++++++++++++++++ tests/unit/test_request_diagnostics.py | 57 +++- 18 files changed, 779 insertions(+), 40 deletions(-) create mode 100644 artemis/llm/final_review.py create mode 100644 tests/unit/test_final_review_budget.py diff --git a/.env.example b/.env.example index 4349986..4559328 100644 --- a/.env.example +++ b/.env.example @@ -51,14 +51,14 @@ ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS=30 ARTEMIS_VIDEO_CHUNK_SECONDS=60 ARTEMIS_VIDEO_TIMEOUT_SECONDS=120 ARTEMIS_VIDEO_MAX_RETRIES=1 -# 检查点、退出汇总与凭据验证超时（秒） +# 中间检查点、退出汇总与凭据验证超时（秒）；最终验收按模型请求配置自动分配独立预算。 ARTEMIS_CHECKPOINT_TIMEOUT_SECONDS=180 ARTEMIS_SETTLEMENT_TIMEOUT_SECONDS=120 ARTEMIS_CREDENTIAL_TIMEOUT_SECONDS=12 ARTEMIS_MODEL_CACHE_TTL_SECONDS=10 # 守护服务轻量探测超时；探测超时且端口仍占用时，不重复启动服务。 ARTEMIS_DAEMON_PROBE_TIMEOUT_SECONDS=2 -# 模型重试耗尽后的暂停等待上限（秒）；0 表示无限等待 +# 执行模型重试耗尽后的暂停等待上限（秒）；0 表示无限等待。最终验收耗尽重试直接记为未验证。 LLM_PAUSE_TIMEOUT_SECONDS=900 # 可选 Google Cloud Vision OCR，独立于模型调用，不配置也可运行。 diff --git a/apps/admin_console/services/request_diagnostics.py b/apps/admin_console/services/request_diagnostics.py index f88137a..2f6f230 100644 --- a/apps/admin_console/services/request_diagnostics.py +++ b/apps/admin_console/services/request_diagnostics.py @@ -27,6 +27,12 @@ def request_diagnostics( sql += " LIMIT ? OFFSET ?" params += (limit, offset) rows = conn.execute(sql, params).fetchall() + review_rows = conn.execute( + "SELECT payload FROM traces WHERE session_id = ? " + "AND type = 'final_review_diagnostic' ORDER BY timestamp DESC" + + (" LIMIT ?" if limit is not None else ""), + (session_id, limit) if limit is not None else (session_id,), + ).fetchall() requests = [] seen = set() running = bool(session and session["status"] in {"running", "resuming"}) @@ -65,12 +71,36 @@ def request_diagnostics( if isinstance(value.get(field), (int, float)) } requests.append(value) + final_reviews = [] + for row in review_rows: + try: + value = json.loads(row["payload"]) + except (ValueError, TypeError): + continue + if not isinstance(value, dict): + continue + value["active"] = ( + running and offset == 0 and not final_reviews and value.get("status") == "running" + ) + if value["active"] and isinstance(value.get("started_at"), (int, float)): + value["elapsed_seconds"] = round(max(0.0, time.time() - value["started_at"]), 3) + value["remaining_seconds"] = round( + max(0.0, value.get("total_seconds", 0.0) - value["elapsed_seconds"]), 3 + ) + value["budget_observed_at"] = time.time() + value["timestamps_utc"] = { + field: datetime.fromtimestamp(value[field], UTC).isoformat() + for field in ("started_at", "deadline_at", "ended_at") + if isinstance(value.get(field), (int, float)) + } + final_reviews.append(value) return { "schema_version": 1, "session_id": session_id, "session_status": session["status"] if session else "unknown", "collected_at": time.time(), "requests": requests, + "final_reviews": final_reviews, "total": total, "offset": offset, "next_offset": offset + limit if limit is not None and offset + limit < total else None, diff --git a/apps/showcase_ui/src/app/components/llm-request-status/llm-request-status.component.html b/apps/showcase_ui/src/app/components/llm-request-status/llm-request-status.component.html index b1de9f4..fe698a3 100644 --- a/apps/showcase_ui/src/app/components/llm-request-status/llm-request-status.component.html +++ b/apps/showcase_ui/src/app/components/llm-request-status/llm-request-status.component.html @@ -1,5 +1,18 @@ @if (sessionId()) {
+ @for (review of activeReviews(); track review.review_id) { +
+
+ + {{ unavailable() ? '验收状态暂时无法更新' : reviewStage(review) }} + 验收已用 {{ reviewElapsed(review) }} 秒 +
+ @if (!unavailable()) { +

完整结论确认前，测试仍为待判定。剩余验收预算 {{ reviewRemaining(review) }} 秒 / 总计 {{ review.total_seconds }} 秒。

+

已发起 {{ review.request_count }} 次模型请求，已重试 {{ review.retry_count }} 次；重试复用已有证据。

+ } +
+ } @for (request of active(); track request.attempt_id) {
@@ -21,6 +34,23 @@ @if (loading()) {

正在加载请求记录…

} @if (!loading() && !unavailable() && !requests().length) {

暂无请求诊断。旧任务没有记录的时间和用量无法补推。

}
+ @for (review of reviews(); track review.review_id) { +
+ {{ review.started_at * 1000 | date:'HH:mm:ss' }} · 最终验收 · {{ reviewStage(review) }} +
+
验收编号
{{ review.review_id }} · {{ review.attempt_id }}
+
验收预算
总计 {{ review.total_seconds }} 秒 · 已用 {{ reviewElapsed(review) }} 秒
+
模型请求
{{ review.request_count }} 次，实际重试 {{ review.retry_count }} 次；每个请求最多重试 {{ review.max_retries }} 次，总预算内执行
+
开始时间
{{ review.started_at * 1000 | date:'yyyy-MM-dd HH:mm:ss.SSS Z' }}
+
预算截止
{{ review.deadline_at * 1000 | date:'yyyy-MM-dd HH:mm:ss.SSS Z' }}
+
结束时间
{{ review.ended_at ? (review.ended_at * 1000 | date:'HH:mm:ss.SSS') : '未记录' }}
+ @if (review.last_failure_category) {
最近错误类别
{{ review.last_failure_category }}
} + @if (review.status === 'budget_exhausted') {
结束原因
取证、模型请求与重试已耗尽验收总预算；未确认通过。
} + @if (review.status === 'request_retries_exhausted') {
结束原因
验收模型请求重试已用尽；未确认通过，没有重跑设备操作。
} + @if (review.status === 'cancelled') {
结束原因
验收被停止或指导中断，不作为请求超时重试。
} +
+
+ } @for (request of requests(); track request.attempt_id) {
{{ request.started_at * 1000 | date:'HH:mm:ss' }} {{ request.model || '模型' }} · {{ role(request) }} · {{ phase(request) }} @if (!request.active && !request.ended_at) { （未记录结束，当前未在执行） } @if (request.is_fallback) { 兜底 } @@ -29,6 +59,7 @@
本次尝试编号
{{ request.attempt_id }}
服务端请求编号
{{ request.server_request_id || '服务未提供' }}
模型配置
{{ request.model || '未记录' }} / {{ request.reasoning_effort || '默认' }} · 第 {{ request.attempt }} 次请求，最多重试 {{ request.max_retries }} 次
+ @if (request.final_review; as review) {
所属最终验收
{{ review.review_id }} · {{ reviewStage(review) }} · 请求记录时剩余 {{ review.remaining_seconds }} 秒
}
输出 Token 上限
{{ request.max_output_tokens || '未设置（由模型服务决定）' }}
有效输出片段
{{ request.output_chunks ?? '未记录' }}（客户端计数，与服务端 SSE 事件数口径不同）
工具参数
{{ request.tool_argument_chars ?? '未记录' }} 字符 @for (tool of request.tools || []; track $index) { {{ tool.name || '工具名未提供' }} · {{ tool.argument_chars }} 字符 }
@@ -39,7 +70,7 @@
结束时间
{{ request.ended_at ? (request.ended_at * 1000 | date:'HH:mm:ss.SSS') : '尚未记录' }}
输入图片
{{ request.images.length }} 张 @for (image of request.images; track $index) { {{ image.width && image.height ? image.width + ' × ' + image.height : '尺寸未知' }} }
Token 用量
{{ request.usage ? '输入 ' + request.usage.prompt_tokens + ' / 输出 ' + request.usage.completion_tokens + ' / 合计 ' + request.usage.total_tokens : '未知（服务未报告）' }}
- @if (request.error) {
失败原因
{{ request.error_type }} · {{ request.error }} · 阶段 {{ phase(request.failure_phase || '') }}
} + @if (request.error) {
失败原因
{{ request.error_type }} · {{ request.error }} · 阶段 {{ phase(request.failure_phase || '') }} @if (request.category) { · {{ request.category }} }
}
} diff --git a/apps/showcase_ui/src/app/components/llm-request-status/llm-request-status.component.spec.ts b/apps/showcase_ui/src/app/components/llm-request-status/llm-request-status.component.spec.ts index d12e794..fe579c6 100644 --- a/apps/showcase_ui/src/app/components/llm-request-status/llm-request-status.component.spec.ts +++ b/apps/showcase_ui/src/app/components/llm-request-status/llm-request-status.component.spec.ts @@ -1,7 +1,7 @@ import { fakeAsync, TestBed, tick } from '@angular/core/testing'; import { provideHttpClient } from '@angular/common/http'; import { HttpTestingController, provideHttpClientTesting } from '@angular/common/http/testing'; -import { LlmRequestStatusComponent, ModelRequest, requestHint } from './llm-request-status.component'; +import { FinalReview, LlmRequestStatusComponent, ModelRequest, requestHint, reviewStage } from './llm-request-status.component'; const waiting: ModelRequest = { request_id: 'call-1', attempt_id: 'attempt-1', model: 'gpt-6-luna', role: 'FlashRunner', @@ -78,4 +78,44 @@ describe('模型请求诊断', () => { expect(text).toContain('本次工具尚未执行'); fixture.destroy(); }); + + it('无模型请求时仍展示验收阶段与总预算，耗尽后保留明确原因', fakeAsync(() => { + const fixture = TestBed.createComponent(LlmRequestStatusComponent); + const http = TestBed.inject(HttpTestingController); + const now = Date.now() / 1000; + const review: FinalReview = { + review_id: 'review-1', attempt_id: 'final#1', status: 'running', stage: 'reading_evidence', active: true, + started_at: now - 20, deadline_at: now + 80, total_seconds: 100, elapsed_seconds: 20, remaining_seconds: 80, + request_count: 2, retry_count: 1, max_retries: 5 + }; + fixture.componentRef.setInput('sessionId', 'review-task'); + fixture.componentRef.setInput('running', true); + fixture.detectChanges(); + http.expectOne('/api/sessions/review-task/request-diagnostics').flush({ requests: [], final_reviews: [review] }); + fixture.detectChanges(); + const text = fixture.nativeElement.querySelector('.review-live').textContent; + expect(text).toContain('正在读取验收证据'); + expect(text).toContain('测试仍为待判定'); + expect(text).toContain('剩余验收预算 80 秒'); + expect(text).toContain('已重试 1 次'); + tick(1000); fixture.detectChanges(); + expect(fixture.nativeElement.querySelector('.review-live').textContent).toContain('剩余验收预算 79 秒'); + tick(2000); + http.expectOne('/api/sessions/review-task/request-diagnostics').flush({ requests: [], final_reviews: [{ + ...review, active: false, status: 'budget_exhausted', ended_at: now + 3, elapsed_seconds: 100 + }] }); + fixture.detectChanges(); + expect(fixture.nativeElement.querySelector('.review-live')).toBeNull(); + expect(fixture.nativeElement.textContent).toContain('验收总预算已用尽'); + expect(fixture.nativeElement.textContent).toContain('未确认通过'); + fixture.destroy(); tick(3000); + http.expectNone('/api/sessions/review-task/request-diagnostics'); + })); + + it('区分重试耗尽、验收中断与完成，不把完成当成通过', () => { + const review = { status: 'request_retries_exhausted' } as FinalReview; + expect(reviewStage(review)).toBe('验收请求重试已用尽'); + expect(reviewStage({ ...review, status: 'cancelled' })).toBe('验收已中断'); + expect(reviewStage({ ...review, status: 'completed' })).toBe('验收完成'); + }); }); diff --git a/apps/showcase_ui/src/app/components/llm-request-status/llm-request-status.component.ts b/apps/showcase_ui/src/app/components/llm-request-status/llm-request-status.component.ts index 356dfd2..cb1f241 100644 --- a/apps/showcase_ui/src/app/components/llm-request-status/llm-request-status.component.ts +++ b/apps/showcase_ui/src/app/components/llm-request-status/llm-request-status.component.ts @@ -14,6 +14,23 @@ export interface ModelRequest { tools?: { name: string | null; argument_chars: number }[]; usage: { prompt_tokens: number; completion_tokens: number; total_tokens: number } | null; images: { width: number | null; height: number | null }[]; + final_review?: FinalReview; + category?: string; +} + +export interface FinalReview { + review_id: string; attempt_id: string; status: string; stage: string; active: boolean; + started_at: number; deadline_at: number; ended_at?: number; + total_seconds: number; elapsed_seconds: number; remaining_seconds: number; + request_count: number; retry_count: number; request_attempt?: number; max_retries: number; + request_phase?: string; last_failure_category?: string; +} + +export function reviewStage(review: FinalReview): string { + if (review.status !== 'running') { + return ({ completed: '验收完成', budget_exhausted: '验收总预算已用尽', request_retries_exhausted: '验收请求重试已用尽', cancelled: '验收已中断', failed: '验收未完成' } as Record)[review.status] || '验收状态待核实'; + } + return ({ preparing_evidence: '正在准备验收证据', reviewing_evidence: '正在核对证据与结论', reading_evidence: '正在读取验收证据', structuring_verdict: '正在整理验收结论', validating_verdict: '正在校验完整结论' } as Record)[review.stage] || '正在进行最终验收'; } export function requestPhase(request: ModelRequest | string): string { @@ -46,20 +63,24 @@ export class LlmRequestStatusComponent { running = input(false); activeChange = output(); requests = signal([]); + reviews = signal([]); now = signal(Date.now() / 1000); unavailable = signal(false); loading = signal(false); active = computed(() => this.running() ? this.requests().filter(r => r.active) : []); + activeReviews = computed(() => this.running() ? this.reviews().filter(r => r.active) : []); downloadUrl = computed(() => `/api/sessions/${encodeURIComponent(this.sessionId() || '')}/request-diagnostics?download=true`); phase = requestPhase; hint = requestHint; + reviewStage = reviewStage; constructor() { - effect(() => this.activeChange.emit(this.active().length > 0)); + effect(() => this.activeChange.emit(this.active().length > 0 || this.activeReviews().length > 0)); effect(onCleanup => { const id = this.sessionId(); const running = this.running(); this.requests.set([]); + this.reviews.set([]); this.unavailable.set(false); this.loading.set(!!id); if (!id) return; @@ -68,9 +89,10 @@ export class LlmRequestStatusComponent { let subscription: Subscription | undefined; const schedule = () => { if (running && !stopped) timer = setTimeout(load, 3000); }; const load = () => { - subscription = this.http.get<{ requests: ModelRequest[] }>(`/api/sessions/${encodeURIComponent(id)}/request-diagnostics`).subscribe({ + subscription = this.http.get<{ requests: ModelRequest[]; final_reviews?: FinalReview[] }>(`/api/sessions/${encodeURIComponent(id)}/request-diagnostics`).subscribe({ next: data => { this.requests.set(data.requests || []); + this.reviews.set(data.final_reviews || []); this.unavailable.set(false); this.loading.set(false); schedule(); @@ -88,6 +110,14 @@ export class LlmRequestStatusComponent { return Math.max(0, Math.floor(request.ended_at ? request.elapsed_seconds : this.now() - request.started_at)); } + reviewElapsed(review: FinalReview): number { + return Math.max(0, Math.floor(review.ended_at ? review.elapsed_seconds : this.now() - review.started_at)); + } + + reviewRemaining(review: FinalReview): number { + return Math.max(0, Math.ceil(review.deadline_at - this.now())); + } + role(request: ModelRequest): string { return ({ FlashRunner: '任务执行', operator: '任务执行', planner: '规划', checker: '结果检查', outputter: '报告整理', step_summarizer: '步骤整理' } as Record)[request.role] || request.role; } diff --git a/artemis/agents/checker/checker.py b/artemis/agents/checker/checker.py index c0d57bc..7d4811c 100644 --- a/artemis/agents/checker/checker.py +++ b/artemis/agents/checker/checker.py @@ -51,6 +51,7 @@ 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 from artemis.llm.parser import extract_thinking_tags +from artemis.llm.final_review import final_review_stage from artemis.llm.structured import content_to_text from artemis.services.llm import acomplete, get_llm, invoke_llm_with_timeout_message from artemis.tools.history import get_history_tools @@ -499,7 +500,9 @@ def _completed_text_report(response, check_items) -> CheckReport | None: async def _structured_report(llm, messages, check_items=()) -> CheckReport: if messages and (report := _completed_text_report(messages[-1], check_items)) is not None: + final_review_stage("validating_verdict") return report + final_review_stage("structuring_verdict") structured_llm = llm.with_structured_output(CheckReport) # A conversation must end with a user turn: Gemini rejects requests whose # last message is the model's own answer ("Requests ending with a model @@ -575,6 +578,7 @@ async def _run_check_loop( report: CheckReport | None = None for i in range(max_iterations): + final_review_stage("reviewing_evidence", iteration=i + 1) if i == max_iterations - 1: messages.append( HumanMessage( @@ -594,6 +598,7 @@ async def _run_check_loop( break messages.append(response) + final_review_stage("reading_evidence") for tc in response.tool_calls: tool_name = tc["name"].split(":")[-1] if ":" in tc["name"] else tc["name"] args = dict(tc["args"]) diff --git a/artemis/agents/checker/flash.py b/artemis/agents/checker/flash.py index 89df9a0..ae87ffd 100644 --- a/artemis/agents/checker/flash.py +++ b/artemis/agents/checker/flash.py @@ -1,10 +1,10 @@ """Flash's exit-only acceptance check uses the same read-only Checker as Pro.""" -import asyncio 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, @@ -14,6 +14,7 @@ from artemis.graph.checkpoints import ( 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 @@ -50,7 +51,6 @@ async def settle_flash_outcome( logger.warning(f"Flash 最终答复保存失败：{exc}") delivery = {"ok": False, "verified": False, "error": str(exc)} if final_check_enabled(ctx): - timeout = float(getattr(ctx.execution_setup, "checkpoint_timeout", 180.0)) publish_checker_event( ctx, { @@ -63,8 +63,8 @@ async def settle_flash_outcome( }, ) try: - report = await asyncio.wait_for( - run_final_check( + report = await final_review_budget(ctx, attempt_id="final#1").run( + lambda: run_final_check( ctx, goal=goal, plan_text="", @@ -74,7 +74,6 @@ async def settle_flash_outcome( execution_report=execution_report, report_delivery=delivery, ), - timeout=timeout, ) result = report.test_result if report.report_corrections and directory: @@ -101,11 +100,7 @@ async def settle_flash_outcome( }, ) except Exception as exc: - detail = ( - f"超过 {timeout:g} 秒" - if isinstance(exc, TimeoutError) - else f"{type(exc).__name__}: {exc}" - ) + detail = final_review_error_text(exc) logger.warning(f"Flash 测试验收未完成：{detail}") result = CaseTestResult( status="inconclusive", reason=f"操作已结束，但测试验收未完成：{detail}。" diff --git a/artemis/data_engine/storage.py b/artemis/data_engine/storage.py index b234a2b..2d69da9 100644 --- a/artemis/data_engine/storage.py +++ b/artemis/data_engine/storage.py @@ -993,7 +993,7 @@ class StorageManager: WHEN excluded.status = 'running' AND traces.status IN ('success', 'failed') THEN traces.payload ELSE excluded.payload END - WHERE traces.type != 'llm_diagnostic' + WHERE traces.type NOT IN ('llm_diagnostic', 'final_review_diagnostic') OR COALESCE(json_extract(excluded.payload, '$.sequence'), 0) > COALESCE(json_extract(traces.payload, '$.sequence'), -1) """, diff --git a/artemis/graph/graph.py b/artemis/graph/graph.py index 6b70554..f724659 100644 --- a/artemis/graph/graph.py +++ b/artemis/graph/graph.py @@ -62,6 +62,8 @@ from artemis.graph.checkpoints import ( from artemis.graph.perception import perception_node from artemis.graph.state import State from artemis.graph.visibility import strict_state +from artemis.llm.diagnostics import final_review_error_text +from artemis.services.llm import final_review_budget from artemis.tools.command_tool import ( HANG_GUIDANCE, manage_task_wrapper, @@ -325,7 +327,6 @@ async def _exit_settlement_node(state: State, ctx: ArtemisContext): attempt_no = ctx.final_check_attempts + 1 ctx.final_check_attempts = attempt_no final_attempt_id = f"final#{attempt_no}" - checkpoint_timeout = float(getattr(setup, "checkpoint_timeout", 180.0) or 180.0) ledger = read_ledger(base_dir) def _final_event(status: str, verdicts: list[dict], **extra) -> None: @@ -345,8 +346,8 @@ async def _exit_settlement_node(state: State, ctx: ArtemisContext): ) try: - review = asyncio.wait_for( - run_final_check( + review = final_review_budget(ctx, attempt_id=final_attempt_id).run( + lambda: run_final_check( ctx, goal=getattr(state, "initial_goal", ""), plan_text=plan_text, @@ -356,7 +357,6 @@ async def _exit_settlement_node(state: State, ctx: ArtemisContext): execution_report=execution_report, report_delivery=report_delivery, ), - timeout=checkpoint_timeout, ) control = getattr(ctx, "task_control", None) if getattr(ctx, "exploration_id", None) and control: @@ -379,11 +379,7 @@ async def _exit_settlement_node(state: State, ctx: ArtemisContext): 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}" - ) + detail = final_review_error_text(e) case_result = CaseTestResult( status="inconclusive", reason=f"最终验收未完成：{detail}。" ) @@ -413,7 +409,7 @@ async def _exit_settlement_node(state: State, ctx: ArtemisContext): "suggestion": "", } ) - _final_event("error", error_verdicts, error=str(e), route="end") + _final_event("error", error_verdicts, error=detail, route="end") report = None if report is not None: diff --git a/artemis/llm/diagnostics.py b/artemis/llm/diagnostics.py index 26b5d3d..64067e6 100644 --- a/artemis/llm/diagnostics.py +++ b/artemis/llm/diagnostics.py @@ -13,6 +13,8 @@ from uuid import uuid4 from PIL import Image from artemis.data_engine.context_vars import CURRENT_NODE_NAME, CURRENT_TRACE_ID +from artemis.llm.final_review import ACTIVE_FINAL_REVIEW, FinalReviewBudgetExceeded +from artemis.llm.reliability import LLMExhaustedError from artemis.llm.usage import usage_from_message from artemis.llm.stream_guard import ToolArgumentOverflow @@ -21,6 +23,8 @@ ACTIVE_ATTEMPT: ContextVar["RequestAttempt | None"] = ContextVar("llm_attempt", def error_text(error: BaseException, timeout: float | None = None) -> str: + if isinstance(error, FinalReviewBudgetExceeded): + return str(error) if isinstance(error, TimeoutError): return f"模型请求在 {timeout:g} 秒内未完成" if timeout else "模型请求超时" text = str(error).strip() or type(error).__name__ @@ -34,6 +38,14 @@ def error_text(error: BaseException, timeout: float | None = None) -> str: return text[:1000] +def final_review_error_text(error: Exception) -> str: + if isinstance(error, FinalReviewBudgetExceeded): + return str(error) + if isinstance(error, LLMExhaustedError): + return f"验收模型请求重试已用尽（{error.failure.category.value}）：{error_text(error)}" + return f"{type(error).__name__}: {error_text(error)}" + + def input_images(args: tuple, kwargs: dict) -> list[dict]: """Inspect only inline image headers; never fetch URLs or retain image data.""" images = [] @@ -76,6 +88,9 @@ class RequestAttempt: self.clock = time.monotonic() self.last_publish = self.clock self.tools: dict[int | str, dict] = {} + self.review = ACTIVE_FINAL_REVIEW.get() + if self.review: + self.review.request_started(request_id, attempt) self.data = { "images": [], **metadata, @@ -102,6 +117,8 @@ class RequestAttempt: self.data["updated_at"] = time.time() self.data["elapsed_seconds"] = round(time.monotonic() - self.clock, 3) self.last_publish = time.monotonic() + if self.review: + self.data["final_review"] = self.review.snapshot() if not self.engine or not getattr(self.engine, "current_session_id", None): return try: @@ -203,14 +220,28 @@ class RequestAttempt: if (usage := usage_from_message(response)) is not None: self.data["usage"] = usage self.data.update(phase=phase, ended_at=time.time()) + if self.review: + self.review.request_finished(phase, category) self.publish() def retry(self, delay): self.data.update( phase="retry_wait", retry_at=time.time() + delay, retry_delay_seconds=delay ) + if self.review: + self.review.request_finished("retry_wait") self.publish() + def cancel(self, error: BaseException | None = None): + if self.review and self.review.expired and not isinstance(error, KeyboardInterrupt): + self.finish( + "failed", + error=FinalReviewBudgetExceeded(self.review.total_seconds), + category="final_review_budget_exhausted", + ) + else: + self.finish("cancelled", error=error, category="cancelled") + async def observe_request_headers(request): """Correlate a wire request and record only hashes of routing headers.""" diff --git a/artemis/llm/final_review.py b/artemis/llm/final_review.py new file mode 100644 index 0000000..63a0a26 --- /dev/null +++ b/artemis/llm/final_review.py @@ -0,0 +1,171 @@ +"""One bounded, observable budget for a read-only final acceptance attempt.""" + +import asyncio +from collections.abc import Awaitable, Callable +from contextvars import ContextVar +import logging +import sys +import time +from uuid import uuid4 + +from artemis.data_engine.context_vars import CURRENT_TRACE_ID +from artemis.llm.reliability import LLMCallError, LLMExhaustedError + +logger = logging.getLogger(__name__) + +# Allow evidence selection, a verdict, and (when needed) structured formatting, +# plus one shared retry allowance. Do not multiply this by the checker's loop +# limit: every request, tool read, backoff and breaker wait spends the same pool. +FINAL_REVIEW_REQUEST_SLOTS = 3 +FINAL_REVIEW_TOOL_ALLOWANCE_SECONDS = 60.0 +MAX_FINAL_REVIEW_SECONDS = 3600.0 + +ACTIVE_FINAL_REVIEW: ContextVar["FinalReviewBudget | None"] = ContextVar( + "final_review_budget", default=None +) + + +class FinalReviewBudgetExceeded(TimeoutError): + def __init__(self, seconds: float): + super().__init__(f"最终验收总预算已用尽（{seconds:g} 秒），尚未取得完整结论") + + +class FinalReviewBudget: + def __init__( + self, + engine, + *, + attempt_id: str, + request_timeout: float, + max_retries: int, + retry_delay: float, + ): + self.engine = engine + self.trace_id = uuid4() + self.parent_id = CURRENT_TRACE_ID.get() + self.total_seconds = min( + MAX_FINAL_REVIEW_SECONDS, + request_timeout * (FINAL_REVIEW_REQUEST_SLOTS + max_retries) + + retry_delay * max_retries + + FINAL_REVIEW_TOOL_ALLOWANCE_SECONDS, + ) + self.clock = time.monotonic() + self.timer: asyncio.Timeout | None = None + self.data = { + "review_id": str(self.trace_id), + "attempt_id": attempt_id, + "status": "running", + "stage": "preparing_evidence", + "iteration": 0, + "request_count": 0, + "retry_count": 0, + "sequence": 0, + "total_seconds": self.total_seconds, + "request_timeout_seconds": request_timeout, + "max_retries": max_retries, + "retry_allowance_seconds": retry_delay * max_retries, + } + + @property + def expired(self) -> bool: + """Only our timer, never a user/guidance cancellation, exhausts the budget.""" + return self.timer is not None and self.timer.expired() + + def snapshot(self) -> dict: + elapsed = self.data.get("elapsed_seconds", time.monotonic() - self.clock) + if self.data["status"] == "running": + elapsed = time.monotonic() - self.clock + return { + **self.data, + "elapsed_seconds": round(elapsed, 3), + "remaining_seconds": round(max(0.0, self.total_seconds - elapsed), 3), + } + + def publish(self) -> None: + self.data["sequence"] += 1 + self.data["updated_at"] = time.time() + if not self.engine or not getattr(self.engine, "current_session_id", None): + return + try: + self.engine.record_trace( + type="final_review_diagnostic", + name="final_review", + trace_id=self.trace_id, + step_id=getattr(self.engine, "current_step_id", None), + parent_trace_id=self.parent_id, + payload=self.snapshot(), + status="running" + if self.data["status"] == "running" + else "success" + if self.data["status"] == "completed" + else "failed", + duration=self.snapshot()["elapsed_seconds"], + ) + except (OSError, RuntimeError, ValueError, TypeError, AttributeError): + logger.warning("Could not persist final acceptance diagnostics", exc_info=True) + + def stage(self, stage: str, *, iteration: int | None = None) -> None: + self.data["stage"] = stage + if iteration is not None: + self.data["iteration"] = iteration + self.publish() + + def request_started(self, request_id: str, attempt: int) -> None: + self.data["request_count" if attempt == 1 else "retry_count"] += 1 + self.data.update( + request_id=request_id, request_attempt=attempt, request_phase="waiting_response" + ) + self.publish() + + def request_finished(self, phase: str, category: str | None = None) -> None: + self.data["request_phase"] = phase + if category: + self.data["last_failure_category"] = category + self.publish() + + def finish(self, status: str, *, error: BaseException | None = None) -> None: + self.data.update( + status=status, + ended_at=time.time(), + elapsed_seconds=time.monotonic() - self.clock, + ) + if error is not None: + self.data["error_type"] = type(error).__name__ + if isinstance(error, LLMCallError): + self.data["last_failure_category"] = error.failure.category.value + self.publish() + + async def run[T](self, operation: Callable[[], Awaitable[T]]) -> T: + self.clock = time.monotonic() + self.data["started_at"] = time.time() + self.data["deadline_at"] = self.data["started_at"] + self.total_seconds + token = ACTIVE_FINAL_REVIEW.set(self) + self.timer = asyncio.timeout(self.total_seconds) + self.publish() + try: + async with self.timer: + result = await operation() + except (KeyboardInterrupt, asyncio.CancelledError) as exc: + self.finish("cancelled", error=exc) + raise + except TimeoutError as exc: + if self.expired: + error = FinalReviewBudgetExceeded(self.total_seconds) + self.finish("budget_exhausted", error=error) + raise error from exc + raise + except LLMExhaustedError as exc: + self.finish("request_retries_exhausted", error=exc) + raise + else: + self.finish("completed") + return result + finally: + if self.data["status"] == "running": + self.finish("failed", error=sys.exception()) + ACTIVE_FINAL_REVIEW.reset(token) + + +def final_review_stage(stage: str, *, iteration: int | None = None) -> None: + if review := ACTIVE_FINAL_REVIEW.get(): + review.stage(stage, iteration=iteration) diff --git a/artemis/services/llm.py b/artemis/services/llm.py index 7904e34..c920463 100644 --- a/artemis/services/llm.py +++ b/artemis/services/llm.py @@ -47,6 +47,10 @@ from artemis.config import ( from artemis.context import ArtemisContext from artemis.data_engine.trace import CURRENT_TRACE_ID, DataEngineCallbackHandler from artemis.llm.diagnostics import ACTIVE_ATTEMPT, RequestAttempt, error_text, input_images +from artemis.llm.final_review import ( + ACTIVE_FINAL_REVIEW, + FinalReviewBudget, +) from artemis.llm.cache_affinity import cache_affinity_options from artemis.llm.reliability import ( CircuitBreaker, @@ -353,8 +357,9 @@ async def _run_with_recovery[T]( immediately: retrying cannot help and pausing would hang the task. - When retryable attempts are exhausted: if a fallback model is waiting (with_fallback), raise LLMExhaustedError so it takes over immediately; - otherwise pause the task (bounded by settings.LLM_PAUSE_TIMEOUT_SECONDS) - and retry from scratch on resume. + final acceptance also raises instead of pausing an already-finished task. + Other calls pause (bounded by settings.LLM_PAUSE_TIMEOUT_SECONDS) and + retry the same request on resume. """ while True: request_token = _begin_llm_request(provider) @@ -394,7 +399,7 @@ async def _run_with_recovery[T]( async with asyncio.timeout(request_timeout or settings.OPENAI_TIMEOUT_SECONDS): result = await call_once() except (KeyboardInterrupt, asyncio.CancelledError) as e: - observation.finish("cancelled", error=e, category="cancelled") + observation.cancel(e) raise except LLMCallError as e: observation.finish("failed", error=e, category=e.failure.category.value) @@ -458,7 +463,7 @@ async def _run_with_recovery[T]( except asyncio.CancelledError: # Guidance may cancel a retry wait while the task continues. # Close this attempt so it cannot look active next to the new request. - observation.finish("cancelled") + observation.cancel() raise else: observation.finish("completed", response=result) @@ -469,7 +474,7 @@ async def _run_with_recovery[T]( ACTIVE_ATTEMPT.reset(observation_token) # Retryable attempts exhausted. - if _FALLBACK_AVAILABLE.get(): + if _FALLBACK_AVAILABLE.get() or ACTIVE_FINAL_REVIEW.get() is not None: if failed_endpoints is not None and endpoint_key: failed_endpoints[endpoint_key] = last_error _record_llm_event( @@ -478,7 +483,7 @@ async def _run_with_recovery[T]( "error": str(last_error)[:1000], "category": last_failure.category.value, "retryable": True, - "handover": "fallback", + "handover": "fallback" if _FALLBACK_AVAILABLE.get() else "final_review_end", }, status="failed", ) @@ -799,6 +804,16 @@ class RobustChatModelWrapper: if emit_deltas: self._emit_stream_delta(stream_exec_id, chunk) except (KeyboardInterrupt, asyncio.CancelledError): + if full_response is not None: + _record_llm_event( + "llm_stream_reset", + { + "stream_exec_id": str(stream_exec_id), + "action": "discard", + "reason": "incomplete_response", + "message": "本次输出未完成，已丢弃半截内容。", + }, + ) raise except Exception as stream_error: if full_response is not None or isinstance(stream_error, ToolArgumentOverflow): @@ -1085,6 +1100,20 @@ def _resolve_endpoint( ) +def final_review_budget(ctx: ArtemisContext, *, attempt_id: str) -> FinalReviewBudget: + """Use the same effective Checker endpoint as get_llm, without creating a model.""" + endpoint = _resolve_endpoint(ctx, "checker") + return FinalReviewBudget( + ctx.data_engine, + attempt_id=attempt_id, + request_timeout=endpoint.timeout_seconds, + max_retries=endpoint.max_retries, + retry_delay=max( + settings.OPENAI_RETRY_INTERVAL_SECONDS, settings.OPENAI_RATE_LIMIT_RETRY_SECONDS + ), + ) + + @overload def get_llm( ctx: ArtemisContext, diff --git a/docs/automation-workspace.md b/docs/automation-workspace.md index 9fd49fe..0d5ec31 100644 --- a/docs/automation-workspace.md +++ b/docs/automation-workspace.md @@ -32,6 +32,10 @@ 诊断可单独下载，也会收录到单任务“全部资料”和批次 ZIP 各任务目录的 `模型请求诊断.json`。请求状态复用任务 trace 数据库，每次尝试一条记录；同一阶段的流式更新至多每 5 秒保存一次，阶段变化立即记录，不额外调用模型、抓取屏幕或记录认证头。强制退出导致缺少结束事件时保留最后阶段，不推断为成功。 +Flash、Pro 和探索的最终验收使用独立总预算，不再复用中间检查点的 180 秒时限。预算按实际 Checker 的单次请求超时 `T`、重试次数 `R`，以及普通/限流重试间隔的较大值 `D` 自动计算：`min(3600, T × (3 + R) + D × R + 60)` 秒。它为取证、结论及必要的结构化整理预留三次正常请求时间，再共用一份重试时间余量和 60 秒工具读取余量；不会按 20 轮循环逐轮放大。每个请求仍最多重试 `R` 次，但全部请求和重试必须在总预算内完成。默认配置的上限为 2760 秒，提前完成就立即结束；用户停止、探索指导及任务本身的执行时限仍生效。 + +页面在验收期间单独显示证据准备/核对/读取、结论整理、完整性校验阶段，以及已用时间、剩余总预算、请求与实际重试次数。诊断 JSON 的 `final_reviews` 记录每次验收的预算、截止时间和结束原因，相关模型请求通过 `final_review.review_id` 关联。请求超时仅重试同一模型请求，保留已取得的工具证据；耗尽请求重试或验收总预算时明确记为未验证，不再进入模型故障后的人工恢复等待。用户停止或探索指导引发的取消单独记录为中断，不重试。超时或中断后的半截输出会从流式展示中移除，只有完整响应经解析、校验后才可能确认通过。 + 探索模式将固定规则放在上下文前部，用户指导变更追加到历史末尾；AI 进度消息计数不再改写前缀。同一次探索恢复后沿用按探索、角色与模型生成的 `prompt_cache_key`，该哈希值也写入请求诊断，便于核对路由。缓存仍取决于服务端的支持和实际命中，不能视为已节省的 token。输入总量包含缓存输入，不能重复相加；结构化调用也按原始响应的 usage 计量一次，旧任务中漏计的用量不会自动回填。300k 是单次上下文限制，并非整个任务累计预算。 Flash 跨端任务由模型通过内部 `update_task_progress` 记录短阶段清单（`notes/task_progress.md`），仅在阶段边界或重新规划时更新，可与下一动作一起提交。执行器允许其他资源的只读预检，但会暂缓不属于当前阶段的操作，避免预检顺势变成整段跨端执行。阶段顺序来自任务要求；确实需要交替操作时，同一阶段可以包含多个资源。完成检查与产品通过仍是两回事。 diff --git a/skills/artemis-use/references/diagnostics.md b/skills/artemis-use/references/diagnostics.md index f2a6288..4b10d6a 100644 --- a/skills/artemis-use/references/diagnostics.md +++ b/skills/artemis-use/references/diagnostics.md @@ -27,6 +27,7 @@ - 按角色/模型分组判断消耗来源，不能把 Checker、视频分析和执行模型混为一个请求。旧数据未记录模型时注明 unknown。 - 请求阶段和重试用于解释等待，时间戳 Unix 秒及 UTC 转成用户当地时区；注明时区。无终止事件或 unknown 不能当成功。 - 区分模型生成耗时、设备/插件操作耗时、排队及用户等待。先找实际重复观测、工具参数膨胀、无进展循环，再建议优化。 +- 最终验收读取 `view_requests` 返回的 `final_reviews`：核对 stage、total/elapsed/remaining_seconds、request_count、retry_count 和 status；用 review_id 关联请求的 final_review 字段。最终验收与中间检查点独立计时。budget_exhausted 是验收总预算耗尽，request_retries_exhausted 是模型请求重试耗尽，cancelled 是停止或指导中断，不能混称为 API 超时。验收耗尽不自动重跑设备操作；流式半截 passed 和验收 completed 都不能代替正式 test_result。 ## 环境与插件 diff --git a/tests/unit/agents/test_flash_acceptance.py b/tests/unit/agents/test_flash_acceptance.py index f747554..c7f07b5 100644 --- a/tests/unit/agents/test_flash_acceptance.py +++ b/tests/unit/agents/test_flash_acceptance.py @@ -111,16 +111,36 @@ async def test_flash_reuses_checker_with_original_prompt_and_final_report( @pytest.mark.asyncio async def test_acceptance_timeout_finishes_with_unknown_verdict(tmp_path, monkeypatch): + from artemis.llm import final_review + async def hanging_check(*args, **kwargs): await asyncio.Event().wait() monkeypatch.setattr(flash, "run_final_check", hanging_check) + monkeypatch.setattr(final_review, "MAX_FINAL_REVIEW_SECONDS", 0.01) 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"] + assert "总预算已用尽（0.01 秒）" in data["test_result"]["reason"] + + +@pytest.mark.asyncio +async def test_flash_final_review_can_outlast_midway_checkpoint_timeout(tmp_path, monkeypatch): + async def slower_review(*args, **kwargs): + await asyncio.sleep(0.02) + return CheckReport( + test_result=CaseTestResult( + status="passed", reason="完整核对实际证据", evidence=["步骤 1 截图"] + ) + ) + + monkeypatch.setattr(flash, "run_final_check", slower_review) + data = await flash.settle_flash_outcome( + context(tmp_path, checkpoint_timeout=0.001), "任务", {"status": "completed"} + ) + assert data["test_result"]["status"] == "passed" @pytest.mark.asyncio diff --git a/tests/unit/graph/test_exit_settlement.py b/tests/unit/graph/test_exit_settlement.py index b29a937..abcec41 100644 --- a/tests/unit/graph/test_exit_settlement.py +++ b/tests/unit/graph/test_exit_settlement.py @@ -125,6 +125,46 @@ async def test_review_stops_capture_before_waiting_and_seals_exploration(explora ctx.on_execution_resume.assert_not_called() +@pytest.mark.asyncio +async def test_exploration_guidance_can_interrupt_shared_final_review_budget(exploration_exit): + from types import SimpleNamespace + + ctx, _, _, _ = exploration_exit + started = asyncio.Event() + closed = asyncio.Event() + + async def review(*args, **kwargs): + started.set() + try: + await asyncio.Event().wait() + finally: + closed.set() + + async def interrupt(operation, **kwargs): + task = asyncio.create_task(operation) + await started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + return True, None + + ctx.task_control = SimpleNamespace( + run_interruptibly=interrupt, directory=ctx.data_engine.base_dir / "control" + ) + with patch("artemis.graph.graph.run_final_check", new=review): + result = await exit_settlement_node(_make_state(execution_report="调查报告"), ctx) + assert result == {"exit_settlement_route": "continue"} + assert closed.is_set() + assert ctx.final_check_attempts == 0 + assert not (ctx.data_engine.base_dir / "run_outcome.json").exists() + diagnostics = [ + call.kwargs["payload"] + for call in ctx.data_engine.record_trace.call_args_list + if call.kwargs.get("type") == "final_review_diagnostic" + ] + assert diagnostics[-1]["status"] == "cancelled" + + @pytest.mark.asyncio async def test_completed_exploration_waits_for_report_before_review(exploration_exit): ctx, repo, session, run = exploration_exit @@ -237,6 +277,32 @@ def _make_state(**overrides): return state +@pytest.mark.asyncio +async def test_pro_final_review_has_its_own_budget_and_reports_exhaustion(tmp_path, monkeypatch): + from artemis.llm import final_review + from artemis.runtime.case_result import CaseTestResult + + _write_plan(tmp_path, DONE_PLAN_NO_CHECKS) + ctx = _make_ctx(tmp_path, checkpoint_timeout=0.001) + + async def slow_review(*args, **kwargs): + await asyncio.sleep(0.02) + return CheckReport( + test_result=CaseTestResult(status="passed", reason="证据齐全", evidence=["步骤 1 截图"]) + ) + + with patch("artemis.graph.graph.run_final_check", new=slow_review): + result = await exit_settlement_node(_make_state(), ctx) + assert result["run_outcome"]["test_result"]["status"] == "passed" + + monkeypatch.setattr(final_review, "MAX_FINAL_REVIEW_SECONDS", 0.001) + with patch("artemis.graph.graph.run_final_check", new=slow_review): + result = await exit_settlement_node(_make_state(), ctx) + assert result["exit_settlement_route"] == "end" + assert result["run_outcome"]["test_result"]["status"] == "inconclusive" + assert "总预算已用尽" in result["run_outcome"]["test_result"]["reason"] + + def _write_plan(tmp_path, content): notes = tmp_path / "notes" notes.mkdir(parents=True, exist_ok=True) diff --git a/tests/unit/test_final_review_budget.py b/tests/unit/test_final_review_budget.py new file mode 100644 index 0000000..1f17b07 --- /dev/null +++ b/tests/unit/test_final_review_budget.py @@ -0,0 +1,245 @@ +import asyncio +from copy import deepcopy +import json +from types import SimpleNamespace + +from langchain_core.messages import AIMessageChunk, HumanMessage, ToolMessage +from langchain_core.tools import tool +import pytest + +from artemis.agents.checker import checker +from artemis.config.llm import LLM +from artemis.llm import final_review +from artemis.llm.diagnostics import ACTIVE_ATTEMPT +from artemis.llm.final_review import ( + ACTIVE_FINAL_REVIEW, + FinalReviewBudget, + FinalReviewBudgetExceeded, +) +from artemis.llm.reliability import LLMExhaustedError +from artemis.llm.router import ModelEndpoint +from artemis.services import llm + + +class RecordingEngine: + current_session_id = "review-test" + current_step_id = None + + def __init__(self): + self.records = [] + + def record_trace(self, **record): + self.records.append(deepcopy(record)) + + def _publish(self, *_): + pass + + def latest(self, kind): + return next(r["payload"] for r in reversed(self.records) if r["type"] == kind) + + +@pytest.fixture +def engine(monkeypatch, tmp_path): + engine = RecordingEngine() + monkeypatch.setattr(llm, "_CURRENT_DATA_ENGINE", engine) + monkeypatch.setattr(llm.settings, "TRACES_PATH", str(tmp_path)) + monkeypatch.setattr(llm, "PAUSE_FILE", tmp_path / "pause") + monkeypatch.setattr( + llm, "retry_policy_for", lambda _: SimpleNamespace(delay_for=lambda _: 0, max_attempts=2) + ) + llm._ENDPOINT_BREAKER._states.clear() + yield engine + llm._ENDPOINT_BREAKER._states.clear() + assert ACTIVE_FINAL_REVIEW.get() is None + assert ACTIVE_ATTEMPT.get() is None + + +def budget(engine, *, timeout=1, retries=1): + return FinalReviewBudget( + engine, attempt_id="final#1", request_timeout=timeout, max_retries=retries, retry_delay=0 + ) + + +def test_budget_uses_effective_checker_config_and_respects_overall_cap(monkeypatch): + monkeypatch.setattr(llm.settings, "OPENAI_TIMEOUT_SECONDS", 300) + monkeypatch.setattr(llm.settings, "OPENAI_MAX_RETRIES", 5) + monkeypatch.setattr(llm.settings, "OPENAI_RETRY_INTERVAL_SECONDS", 10) + monkeypatch.setattr(llm.settings, "OPENAI_RATE_LIMIT_RETRY_SECONDS", 60) + cfg = LLM(model="test", timeout=20, max_retries=2) + ctx = SimpleNamespace(llm_config=SimpleNamespace(get_agent=lambda name: cfg), data_engine=None) + review = llm.final_review_budget(ctx, attempt_id="final#1") + assert review.total_seconds == 280 # three normal calls + two retries + waits/tool reads + assert review.data["request_timeout_seconds"] == 20 + assert review.data["max_retries"] == 2 + + cfg.timeout = 300 + cfg.max_retries = 100 + assert llm.final_review_budget(ctx, attempt_id="final#2").total_seconds == 3600 + + +@pytest.mark.asyncio +async def test_checker_retries_only_failed_request_with_existing_tool_evidence(engine, monkeypatch): + calls = [] + evidence_reads = 0 + stream_closed = asyncio.Event() + + @tool + async def read_evidence() -> str: + """Read the already-recorded page evidence.""" + nonlocal evidence_reads + evidence_reads += 1 + return "实际为中文搜索，预期为英文搜索" + + class ScriptedModel: + def bind_tools(self, *args, **_): + return self + + async def astream(self, messages, **_): + calls.append(deepcopy(messages)) + if len(calls) == 1: + yield AIMessageChunk( + content="", + tool_calls=[{"name": "read_evidence", "args": {}, "id": "read-1"}], + ) + elif len(calls) == 2: + try: + yield AIMessageChunk(content='{"test_result":{"status":"passed","reason":"') + await asyncio.Event().wait() # the *request* timer triggers a retry + finally: + stream_closed.set() + else: + yield AIMessageChunk( + content=json.dumps( + { + "test_result": { + "status": "failed", + "reason": "中文搜索不满足英文预期", + "evidence": ["read_evidence：实际为中文搜索"], + } + } + ) + ) + + wrapper = llm.RobustChatModelWrapper( + ScriptedModel(), + endpoint=ModelEndpoint(model_name="test", timeout_seconds=0.02, max_retries=1), + ) + monkeypatch.setattr(checker, "get_llm", lambda **_: wrapper) + monkeypatch.setattr(checker, "trace_langchain_tool", lambda tool, ctx: tool) + events = [] + monkeypatch.setattr(llm, "_record_llm_event", lambda *args, **kwargs: events.append(args)) + ctx = SimpleNamespace(execution_setup=SimpleNamespace(checker_max_iterations=4)) + review = budget(engine) + report = await review.run( + lambda: checker._run_check_loop( + ctx, [HumanMessage(content="核实英文搜索")], [read_evidence], [] + ) + ) + + assert report.test_result.status == "failed" # partial `passed` was never a verdict + assert evidence_reads == 1 + assert len(calls) == 3 + assert calls[1] == calls[2] + assert any(isinstance(m, ToolMessage) and "中文搜索" in m.content for m in calls[2]) + assert stream_closed.is_set() + assert any(e[0] == "llm_stream_reset" and e[1]["action"] == "discard" for e in events) + stored = engine.latest("final_review_diagnostic") + assert stored["status"] == "completed" + assert stored["request_count"] == 2 + assert stored["retry_count"] == 1 + assert stored["last_failure_category"] == "timeout" + assert engine.latest("llm_diagnostic")["final_review"]["review_id"] == stored["review_id"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("phase", ["request", "retry_wait", "evidence"]) +async def test_total_budget_expiry_is_not_user_cancellation_or_a_new_retry( + engine, monkeypatch, tmp_path, phase +): + monkeypatch.setattr(final_review, "MAX_FINAL_REVIEW_SECONDS", 0.02) + monkeypatch.setattr( + llm, "retry_policy_for", lambda _: SimpleNamespace(delay_for=lambda _: 60, max_attempts=6) + ) + cancelled = asyncio.Event() + calls = 0 + + async def request(): + nonlocal calls + calls += 1 + if phase == "retry_wait": + raise ConnectionError("connection reset") + try: + await asyncio.Event().wait() + finally: + cancelled.set() + + review = budget(engine, retries=5) + + async def operation(): + if phase == "evidence": + return await request() + return await llm._run_with_recovery(request, request_timeout=1, max_retries=5) + + with pytest.raises(FinalReviewBudgetExceeded, match="总预算已用尽"): + await review.run(operation) + assert calls == 1 + assert not (tmp_path / "pause").exists() + assert engine.latest("final_review_diagnostic")["status"] == "budget_exhausted" + if phase != "retry_wait": + assert cancelled.is_set() + if phase != "evidence": + stored = engine.latest("llm_diagnostic") + assert stored["phase"] == "failed" + assert stored["category"] == "final_review_budget_exhausted" + + +@pytest.mark.asyncio +async def test_exhausted_final_review_request_ends_without_hidden_pause( + engine, monkeypatch, tmp_path +): + calls = 0 + + async def fails(): + nonlocal calls + calls += 1 + raise TimeoutError() + + pause = [] + monkeypatch.setattr(llm, "_handle_llm_pause_and_resume", lambda *_: pause.append(True)) + with pytest.raises(LLMExhaustedError): + await budget(engine).run( + lambda: llm._run_with_recovery(fails, request_timeout=1, max_retries=1) + ) + assert calls == 2 + assert not pause + assert not (tmp_path / "pause").exists() + assert engine.latest("final_review_diagnostic")["status"] == "request_retries_exhausted" + + +@pytest.mark.asyncio +async def test_user_cancellation_closes_review_and_child_without_retry(engine): + started = asyncio.Event() + cancelled = asyncio.Event() + calls = 0 + + async def request(): + nonlocal calls + calls += 1 + started.set() + try: + await asyncio.Event().wait() + finally: + cancelled.set() + + task = asyncio.create_task( + budget(engine).run( + lambda: llm._run_with_recovery(request, request_timeout=1, max_retries=5) + ) + ) + await started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert calls == 1 and cancelled.is_set() + assert engine.latest("final_review_diagnostic")["status"] == "cancelled" + assert engine.latest("llm_diagnostic")["category"] == "cancelled" diff --git a/tests/unit/test_request_diagnostics.py b/tests/unit/test_request_diagnostics.py index bf7fae4..8c88289 100644 --- a/tests/unit/test_request_diagnostics.py +++ b/tests/unit/test_request_diagnostics.py @@ -251,7 +251,10 @@ def test_error_redaction_and_image_headers(): assert error_text(RuntimeError()) == "RuntimeError" -def test_snapshots_survive_out_of_order_writes_and_export(tmp_path): +@pytest.mark.parametrize( + "kind, key", [("llm_diagnostic", "requests"), ("final_review_diagnostic", "final_reviews")] +) +def test_snapshots_survive_out_of_order_writes_and_export(tmp_path, kind, key): from artemis.data_engine.models import TraceRecord from artemis.data_engine.storage import StorageManager from apps.admin_console.database.connection import db_session @@ -270,13 +273,14 @@ def test_snapshots_survive_out_of_order_writes_and_export(tmp_path): TraceRecord( trace_id=trace_id, session_id=session_id, - type="llm_diagnostic", + type=kind, name="llm_request", timestamp=10 + sequence, status="success" if sequence == 3 else "running", payload={ "sequence": sequence, "phase": phase, + "status": "completed" if sequence == 3 else "running", "request_id": "call-one", "started_at": 10, "ended_at": 13 if sequence == 3 else None, @@ -284,10 +288,51 @@ def test_snapshots_survive_out_of_order_writes_and_export(tmp_path): ) ) result = request_diagnostics(str(session_id), db_path=path) - assert len(result["requests"]) == 1 - assert result["requests"][0]["phase"] == "completed" - assert result["requests"][0]["active"] is False - assert result["requests"][0]["timestamps_utc"]["started_at"].endswith("+00:00") + assert len(result[key]) == 1 + assert result[key][0]["phase"] == "completed" + assert result[key][0]["active"] is False + assert result[key][0]["timestamps_utc"]["started_at"].endswith("+00:00") + assert result["total"] == (1 if kind == "llm_diagnostic" else 0) + + +def test_review_diagnostics_show_evidence_stage_without_a_live_model_request(tmp_path, monkeypatch): + from artemis.data_engine.storage import StorageManager + from apps.admin_console.database.connection import db_session + from apps.admin_console.services import request_diagnostics as service + + path = tmp_path / "data.db" + StorageManager(db_path=path, base_trace_dir=tmp_path) + monkeypatch.setattr(service.time, "time", lambda: 130) + with db_session(path) as conn: + conn.execute("INSERT INTO sessions(session_id,status) VALUES ('task', 'running')") + value = { + "review_id": "final-review", + "status": "running", + "stage": "reading_evidence", + "started_at": 100, + "deadline_at": 200, + "total_seconds": 100, + "updated_at": 110, + } + conn.execute( + "INSERT INTO traces(trace_id,session_id,type,timestamp,payload) " + "VALUES ('review', 'task', 'final_review_diagnostic', 100, ?)", + (json.dumps(value),), + ) + conn.commit() + result = service.request_diagnostics("task", db_path=path, limit=100) + assert result["requests"] == [] + review = result["final_reviews"][0] + assert review["active"] is True + assert review["stage"] == "reading_evidence" + assert review["elapsed_seconds"] == 30 + assert review["remaining_seconds"] == 70 + assert review["updated_at"] == 110 # preserve the actual last persisted event time + assert review["timestamps_utc"]["deadline_at"].endswith("+00:00") + assert ( + service.request_diagnostics("task", db_path=path, offset=1)["final_reviews"][0]["active"] + is False + ) @pytest.mark.asyncio