from __future__ import annotations import hashlib import json import math import platform import re import time from copy import deepcopy from contextvars import ContextVar from datetime import datetime, timedelta, timezone from pathlib import Path from urllib.parse import parse_qsl, urlsplit from fastapi import HTTPException from pydantic import ValidationError from .db import Store, digest, encoded, identifier, now from .learning_checks import evidence_graph_errors, output_assertion_checks, step_evidence_references from .learning_contracts import Action, valid_selector, execution_metadata, execution_metadata_matches from .reading import audit_note_binding, detail_contract from .worker import WorkerError, safe_request_observations, safe_diagnostics, safe_login_detection, safe_popup_observations, safe_batch_evidence, safe_error_codes, safe_failure_diagnostics, safe_verify_timings SENSITIVE_KEY = re.compile(r"(^|[_-])(password|passwd|phone|mobile|otp|sms|api_key|access_token|refresh_token)($|[_-])|密码|验证码|手机号", re.I) NAME = re.compile(r"[A-Za-z_][A-Za-z0-9_]{0,63}") MAX_STEPS = 30 PLAIN_ROOT_FORBIDDEN = {"input", "textarea", "select", "script", "style", "noscript", "iframe", "button", "form", "body", "html"} SEEK_STOP_REASONS = {"target_reached", "boundary", "scroll_budget", "time_budget", "ambiguous", "obscured", "target_changed", "source_changed", "unsafe_target", "binding_mismatch", "session_unavailable", "blocked_request"} def wait_scope_evidence_valid(observed: dict, selector: str) -> bool: """Only the current ordinary container is evidence; rows may not exist yet.""" if not isinstance(selector, str) or not valid_selector(selector, relative=True): return False roots = [item for item in observed.get("elements", []) if isinstance(item, dict) and item.get("selector") == selector] if len(roots) != 1: return False root = roots[0] return bool(root.get("kind") == "read" and root.get("unique") is True and root.get("sensitive") is False and root.get("in_viewport") is True and isinstance(root.get("tag"), str) and re.fullmatch(r"[a-z][a-z0-9-]{0,63}", root["tag"]) and root["tag"] not in PLAIN_ROOT_FORBIDDEN) def seek_evidence_valid(observed: dict, root: dict | None, target: dict, scroll_scope: str | None = None) -> bool: if (not isinstance(target, dict) or not root or root.get("selector") in {"body", "html"} or root.get("sensitive") or root.get("unique") is not True or not isinstance(root.get("tag"), str) or not re.fullmatch(r"[a-z][a-z0-9-]{0,63}", root["tag"]) or root["tag"] in PLAIN_ROOT_FORBIDDEN): return False if scroll_scope == "viewport": viewport = observed.get("viewport") if not isinstance(viewport, dict) or any(type(viewport.get(key)) not in {int, float} or not 0 <= viewport[key] <= 2 ** 53 or not math.isfinite(viewport[key]) for key in ("width", "height", "scroll_x", "scroll_y")) or not viewport["width"] or not viewport["height"]: return False elif scroll_scope is not None or type((root.get("scroll_y") or {}).get("can_scroll")) is not bool: return False selector = target.get("selector") if any(hint.get("selector") == selector and hint.get("tag") not in PLAIN_ROOT_FORBIDDEN for hint in root.get("read_hints", []) if isinstance(hint, dict)): return True return any(item.get("collection") is True and not item.get("sensitive") and item.get("tag") not in PLAIN_ROOT_FORBIDDEN and (item.get("selector") == selector or (isinstance(item.get("source_selectors"), list) and selector in item["source_selectors"])) for item in observed.get("elements", []) if isinstance(item, dict)) def safe_seek_result(value, step: dict, position: int) -> dict | None: if not isinstance(value, dict) or set(value) != {"step_position", "scrolls", "movements", "target_reached", "stop_reason"}: return None count, movements, reached, reason = (value.get(key) for key in ("scrolls", "movements", "target_reached", "stop_reason")) if (type(value.get("step_position")) is not int or value["step_position"] != position or type(count) is not int or not 0 <= count <= step.get("max_scrolls", 1) or not isinstance(movements, list) or len(movements) != count or len(movements) > 10 or type(reached) is not bool or not isinstance(reason, str) or reason not in SEEK_STOP_REASONS or reached != (reason == "target_reached")): return None for movement in movements: if not isinstance(movement, dict) or set(movement) != {"before", "after", "scroll_height", "client_height"}: return None if any(type(item) not in {int, float} or not math.isfinite(item) or not 0 <= item <= 2 ** 53 for item in movement.values()): return None return deepcopy(value) def normalize_worker_unicode(value): """Repair invalid UTF-16 fragments at the browser boundary, not JSON/digests. JavaScript string limits can leave a lone surrogate. Valid Unicode (including combining sequences and supplementary characters) is returned unchanged. """ if isinstance(value, str): if re.search(r"[\ud800-\udfff]", value): return value.encode("utf-16-le", "surrogatepass").decode("utf-16-le", "replace") return value if isinstance(value, list): return [normalize_worker_unicode(item) for item in value] if isinstance(value, dict): return {normalize_worker_unicode(key): normalize_worker_unicode(item) for key, item in value.items()} return value def check_entry_url(value: str) -> str: try: parsed = urlsplit(value) if parsed.scheme not in ("http", "https") or not parsed.hostname or parsed.username or parsed.password: raise ValueError if any(SENSITIVE_KEY.search(key) for key, _ in parse_qsl(parsed.query)): raise ValueError parsed.port except ValueError: raise HTTPException(422, "入口需为不含凭据的 http 或 https 页面地址") from None return value def check_parameters(value: dict) -> dict: if len(value) > 20: raise HTTPException(422, "首版每条链路最多 20 个参数") for key, item in value.items(): if not NAME.fullmatch(key) or SENSITIVE_KEY.search(key): raise HTTPException(422, "参数名需为普通业务字段，不能用于登录凭据") if not isinstance(item, str) or len(item) > 2000: raise HTTPException(422, "首版参数值需为不超过 2000 字符的文本") return value def os_name(value: str) -> str: return {"win32": "windows", "windows": "windows", "darwin": "macos", "macos": "macos", "linux": "linux"}.get(value.lower(), value.lower()) def compare_outputs(actual: dict, expected: dict, contracts: dict | None = None) -> list[str]: return [item["message"] for item in output_assertion_checks(actual, expected, contracts) if not item["passed"]] class LearningService: def __init__(self, store: Store, worker): self.store, self.worker = store, worker self.sessions: dict[str, dict] = {} self.route_tasks: set[str] = set() self.route_task: ContextVar[str | None] = ContextVar("learning_route_task", default=None) # Browser contexts are process-owned. Persisted learning evidence survives, # but a new server cannot inherit a predecessor's ready confirmation. self.store.execute("UPDATE learning_tasks SET status='paused',updated_at=$1 WHERE status IN ('awaiting_user','observing')", (now(),)) def browser_readiness(self, task_id: str) -> dict: session = self.sessions.get(task_id, {}) status = session.get("status", "closed") detection = session.get("login_detection") extra = {"login_detection": deepcopy(detection)} if detection else {} if session.get("execution_profile") and execution_metadata_matches(session, session["execution_profile"]): extra.update(execution_metadata(session["execution_profile"])) block_reason = session.get("block_reason") if block_reason in {"popup_blocked", "scope_blocked"}: message = "出现未确认的弹窗，当前会话已暂停；请关闭学习浏览器后重新打开并确认页面" if block_reason == "popup_blocked" else "页面尝试离开允许的站点范围，当前会话已暂停；请关闭学习浏览器后重新打开并确认页面" return {"ready": False, "code": block_reason, "block_reason": block_reason, "message": message, "next_action": "close_and_reopen_browser", **extra} if status == "ready": return {"ready": True, "code": "ready", "message": "本次浏览器会话已由维护者确认，可以观察或试跑", "next_action": "none", **extra} if status == "awaiting_user": state = (detection or {}).get("state") messages = {"logged_in": "已检测到登录，无需重新扫码；请确认当前账号后继续", "logged_out": "当前页面显示未登录，请在官方页面自行完成登录后再确认", "verification_required": "当前页面需要人工验证，请先在官方页面处理后再确认"} code = "login_required" if state == "logged_out" else "verification_required" if state == "verification_required" else "confirmation_required" return {"ready": False, "code": code, "message": messages.get(state, "请核对官方页面登录状态并人工确认当前账号；确认前不采集页面"), "next_action": "confirm_ready", **extra} return {"ready": False, "code": "session_closed", "message": "请打开学习浏览器并人工确认登录状态；历史证据和验证记录已保留", "next_action": "open_browser"} def require_route_access(self, task_id: str) -> None: if task_id in self.route_tasks and self.route_task.get() != task_id: raise HTTPException(409, {"code": "ROUTE_BUSY", "message": "当前任务正在验证完整链路，请等待结束后再操作学习浏览器"}) def pause_active_task(self, task_id: str) -> None: self.store.execute("UPDATE learning_tasks SET status='paused',updated_at=$1 WHERE id=$2 AND status IN ('awaiting_user','observing')", (now(), task_id)) async def refresh_browser(self, task_id: str, owner_id: str) -> dict: self.task(task_id, owner_id) if task_id in self.sessions: try: state = await self.call(task_id, "status", {}) except HTTPException: if self.sessions.get(task_id, {}).get("block_reason") not in {"popup_blocked", "scope_blocked"}: self.sessions.pop(task_id, None) self.pause_active_task(task_id) else: if state.get("status") == "closed": self.sessions.pop(task_id, None) self.pause_active_task(task_id) else: detected_state = (state.get("login_detection") or {}).get("state") invalidated = self.sessions[task_id].get("confirmation_invalidated", False) or detected_state in {"logged_out", "verification_required"} or state.get("block_reason") in {"popup_blocked", "scope_blocked"} confirmed = state.get("manual_confirmation") and not invalidated self.sessions[task_id] = {**state, "confirmation_invalidated": invalidated, "status": "ready" if confirmed else "awaiting_user"} return self.browser_readiness(task_id) async def require_ready(self, task_id: str, owner_id: str) -> None: readiness = await self.refresh_browser(task_id, owner_id) if not readiness["ready"]: code = readiness["code"] if readiness["code"] in {"popup_blocked", "scope_blocked"} else "browser_confirmation_required" if readiness["code"] == "confirmation_required" else "browser_session_required" raise HTTPException(409, {"code": code, "message": readiness["message"], "next_action": readiness["next_action"]}) def task(self, task_id: str, owner_id: str) -> dict: row = self.store.one("SELECT * FROM learning_tasks WHERE id=$1 AND owner_id=$2", (task_id, owner_id)) if not row: raise HTTPException(404, "学习任务不存在") row["parameters"] = json.loads(row.pop("parameters_json")) row.pop("owner_id", None) row.pop("ordinal", None) if row.get("execution_profile") is None: row.pop("execution_profile", None) row["browser_status"] = self.sessions.get(task_id, {}).get("status", "closed") row["browser_readiness"] = self.browser_readiness(task_id) return row def candidate(self, candidate_id: str, owner_id: str) -> dict: row = self.store.one("SELECT c.* FROM candidates c JOIN learning_tasks t ON t.id=c.task_id WHERE c.id=$1 AND t.owner_id=$2", (candidate_id, owner_id)) if not row: raise HTTPException(404, "候选能力不存在") row["manifest"] = json.loads(row.pop("manifest_json")) row.pop("ordinal", None) validation = self.store.one(f"SELECT kind,status FROM validations WHERE candidate_id=$1 AND kind IN ('static','browser') ORDER BY {self.store.order_column()} DESC LIMIT 1", (candidate_id,)) row["status"] = "validated" if validation and validation["kind"] == "browser" and validation["status"] == "passed" else "checked" if validation and validation["kind"] == "static" and validation["status"] == "passed" else "candidate" row["quality"] = self.candidate_quality(candidate_id) return row def tasks(self, owner_id: str) -> list[dict]: return [self.task(row["id"], owner_id) for row in self.store.rows(f"SELECT id FROM learning_tasks WHERE owner_id=$1 ORDER BY {self.store.order_column()} DESC", (owner_id,))] def create_task(self, owner_id: str, values: dict) -> dict: check_entry_url(values["entry_url"]) try: execution_metadata(values.get("execution_profile")) except ValueError: raise HTTPException(422, "执行模式不受支持") from None parameters = check_parameters(values.get("parameters") or {}) if not values["name"].strip(): raise HTTPException(422, "学习任务名称不能为空") timestamp, task_id = now(), identifier("learn") account_key = values["account_key"].strip() if not re.fullmatch(r"[A-Za-z0-9_.-]{2,80}", account_key) or re.fullmatch(r"1\d{10}", account_key): raise HTTPException(422, "本地账号标识请使用 2 至 80 位字母数字编号，不使用平台手机号") self.store.execute("INSERT INTO learning_tasks(id,owner_id,name,platform,entry_url,goal,account_key,parameters_json,status,created_at,updated_at,execution_profile) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)", (task_id, owner_id, values["name"].strip(), values["platform"], values["entry_url"], values["goal"].strip(), account_key, encoded(parameters), "draft", timestamp, timestamp, values.get("execution_profile"))) return self.task(task_id, owner_id) def set_status(self, task_id: str, status: str) -> None: self.store.execute("UPDATE learning_tasks SET status=$1,updated_at=$2 WHERE id=$3", (status, now(), task_id)) def overview(self, owner_id: str) -> dict: task_ids = "SELECT id FROM learning_tasks WHERE owner_id=$1" def count(table, where="", args=()): scope = "owner_id=$1" if table == "learning_tasks" else f"task_id IN ({task_ids})" return self.store.one(f"SELECT count(*) AS n FROM {table} WHERE {scope}{where}", (owner_id, *args))["n"] return {"tasks": count("learning_tasks"), "candidates": count("candidates"), "validations": count("validations"), "passed": count("validations", " AND status='passed'"), "failed": count("validations", " AND status='failed'"), "published": 0} def observations(self, task_id: str) -> list[dict]: return [{"id": row["id"], "task_id": row["task_id"], "created_at": row["created_at"], **json.loads(row["data_json"])} for row in self.store.rows(f"SELECT * FROM observations WHERE task_id=$1 ORDER BY {self.store.order_column()} ASC", (task_id,))] def steps(self, task_id: str) -> list[dict]: return [{"id": row["id"], "position": row["position"], "created_at": row["created_at"], **json.loads(row["data_json"])} for row in self.store.rows("SELECT * FROM learning_steps WHERE task_id=$1 ORDER BY position", (task_id,))] def validations(self, owner_id: str, candidate_id: str | None = None, task_id: str | None = None) -> list[dict]: query = "SELECT v.* FROM validations v JOIN learning_tasks t ON t.id=v.task_id WHERE t.owner_id=$1" args = [owner_id] if candidate_id: args.append(candidate_id) query += f" AND v.candidate_id=${len(args)}" if task_id: args.append(task_id) query += f" AND v.task_id=${len(args)}" return [self.validation_dto(row) for row in self.store.rows(query + f" ORDER BY {self.store.order_column('v')} DESC", tuple(args))] def quality_observations(self, validation_id: str) -> list[dict]: return [{"id": row["id"], "validation_id": row["validation_id"], "candidate_id": row["candidate_id"], "task_id": row["task_id"], "created_at": row["created_at"], **json.loads(row["data_json"])} for row in self.store.rows(f"SELECT * FROM quality_observations WHERE validation_id=$1 ORDER BY {self.store.order_column()} ASC", (validation_id,))] def validation_dto(self, row: dict) -> dict: record = {key: row[key] for key in ("id", "candidate_id", "task_id", "kind", "status", "created_at")} record.update(json.loads(row["data_json"])) record["quality_observations"] = self.quality_observations(row["id"]) record["quality"] = self.validation_quality(record) return record @staticmethod def quality_result(state: str, **data) -> dict: return {"state": state, "basis": "manual_scope_review", "scope": "local_maintainer", "published": False, "automatic_platform_detection": False, **data} def validation_quality(self, record: dict) -> dict: if record.get("kind") != "browser" or record.get("mode", "exploration") != "runtime": return self.quality_result("experimental") observations = record.get("quality_observations") or [] latest = observations[-1] if observations else None if record.get("status") != "passed" or any(item["state"] in {"alert", "uncertain"} or item["process_review"] == "failed" for item in observations): state = "review_required" elif latest and latest["state"] == "clear" and latest["process_review"] == "passed": state = "verified_in_scope" else: state = "awaiting_observation" return self.quality_result(state, validation_id=record["id"], observation_due_at=record.get("observation_due_at"), latest_observation_id=latest["id"] if latest else None) def candidate_quality(self, candidate_id: str) -> dict: records = [self.validation_dto(row) for row in self.store.rows(f"SELECT * FROM validations WHERE candidate_id=$1 AND kind='browser' ORDER BY {self.store.order_column()} DESC", (candidate_id,))] runtime = next((row for row in records if row.get("mode") == "runtime"), None) if not runtime: return self.quality_result("experimental") blockers = [item["created_at"] for record in records for item in record["quality_observations"] if item["state"] in {"alert", "uncertain"} or item["process_review"] == "failed"] # Receipt time, not an operator-supplied event time, fences recovery from # late feedback. An old successful run cannot clear a later alert. barrier = max(blockers, default="") if barrier and runtime.get("started_at", "") <= barrier: return self.quality_result("review_required", validation_id=runtime["id"], review_since=barrier) return runtime["quality"] def append_quality_observation(self, validation_id: str, owner_id: str, values: dict) -> dict: row = self.store.one("SELECT v.* FROM validations v JOIN learning_tasks t ON t.id=v.task_id WHERE v.id=$1 AND t.owner_id=$2", (validation_id, owner_id)) if not row: raise HTTPException(404, "验证记录不存在") record = self.validation_dto(row) if record["kind"] != "browser": raise HTTPException(422, "只有实际浏览器试跑可登记后续质量观察") if not values["note"].strip(): raise HTTPException(422, "请记录实际检查内容及关联依据") timestamp = now() received = datetime.fromisoformat(timestamp) occurred = values.get("occurred_at") if occurred: try: parsed = datetime.fromisoformat(occurred.replace("Z", "+00:00")) if parsed.tzinfo is None or parsed > received or parsed < datetime.fromisoformat(record.get("started_at") or record["created_at"]): raise ValueError occurred = parsed.astimezone(timezone.utc).isoformat(timespec="milliseconds") except (ValueError, TypeError): raise HTTPException(422, "反馈发生时间需含时区，且位于本次试跑开始与登记时间之间") from None if values["state"] == "clear": due = record.get("observation_due_at") if record.get("mode") != "runtime" or record["status"] != "passed" or values["process_review"] != "passed": raise HTTPException(409, "无异常结论需要正式模式成功试跑及通过人工过程复核") if not due or received < datetime.fromisoformat(due): raise HTTPException(409, "后续观察窗口尚未到期，请先登记等待或实际异常") if occurred and datetime.fromisoformat(occurred) < datetime.fromisoformat(due): raise HTTPException(409, "无异常结论需基于观察窗口到期后的实际检查，不能用到期前的旧检查补登记") observation_id = identifier("quality") content = {**values, "note": values["note"].strip(), "occurred_at": occurred, "verification": "operator_reported"} # This append preserves the original validation and all previous reports. self.store.execute("INSERT INTO quality_observations(id,validation_id,candidate_id,task_id,data_json,created_at) VALUES($1,$2,$3,$4,$5,$6)", (observation_id, validation_id, row["candidate_id"], row["task_id"], encoded(content), timestamp)) return {"id": observation_id, "validation_id": validation_id, "candidate_id": row["candidate_id"], "task_id": row["task_id"], "created_at": timestamp, **content} def detail(self, task_id: str, owner_id: str) -> dict: task = self.task(task_id, owner_id) task.update(observations=self.observations(task_id), steps=self.steps(task_id), candidates=[self.candidate(row["id"], owner_id) for row in self.store.rows("SELECT id FROM candidates WHERE task_id=$1 ORDER BY version DESC", (task_id,))], validations=self.validations(owner_id, task_id=task_id)) return task def candidates(self, owner_id: str) -> list[dict]: return [self.candidate(row["id"], owner_id) for row in self.store.rows(f"SELECT c.id FROM candidates c JOIN learning_tasks t ON t.id=c.task_id WHERE t.owner_id=$1 ORDER BY {self.store.order_column('c')} DESC", (owner_id,))] async def call(self, task_id: str, method: str, params: dict, *, preserve_failure: bool = False) -> dict: # Recheck at the actual RPC boundary as another request may have begun # a route while this request awaited a read-only readiness/status call. if method in {"open", "ready", "capture", "action", "close", "verify"}: self.require_route_access(task_id) try: result = normalize_worker_unicode(await self.worker.call(method, {"task_id": task_id, **params})) if method in {"open", "ready", "status", "capture", "action", "verify"} and not (method == "status" and result.get("status") == "closed"): self.require_execution_metadata(task_id, result) captures = ([result] if method == "capture" else [result.get("before"), result.get("after"), *[item.get("capture") for item in result.get("batch_evidence", []) if isinstance(item, dict)]] if method == "action" else result.get("evidence", []) if method == "verify" else []) for capture in captures: if capture is not None: self.require_execution_metadata(task_id, capture) if "login_detection" in result: result["login_detection"] = safe_login_detection(result["login_detection"]) if "block_reason" in result and (not isinstance(result["block_reason"], str) or result["block_reason"] not in {"popup_blocked", "scope_blocked"}): result["block_reason"] = None if "popup_observations" in result: result["popup_observations"] = safe_popup_observations(result["popup_observations"]) if "batch_evidence" in result: result["batch_evidence"] = safe_batch_evidence(result["batch_evidence"]) if "error_codes" in result: result["error_codes"] = safe_error_codes(result["error_codes"]) if method == "verify" and result.get("session_closed") is True: # A normal failure result may still contain real partial outputs # and evidence; clear readiness without discarding that result. self.sessions.pop(task_id, None) self.set_status(task_id, "paused") return result except WorkerError as error: if error.session_closed or error.code.lower() in ("worker_closed", "worker_timeout", "browser_closed", "session_not_found", "session_closed"): self.sessions.pop(task_id, None) self.pause_active_task(task_id) elif error.code.lower() in {"login_required", "verification_required"}: session = self.sessions.get(task_id) if session: session.update(status="awaiting_user", confirmation_invalidated=True) session.pop("login_detection", None) elif error.code.lower() in {"popup_blocked", "scope_blocked"}: session = self.sessions.get(task_id) if session: session.update(status="awaiting_user", confirmation_invalidated=True, block_reason=error.code.lower()) if preserve_failure: raise if error.code.lower() in {"popup_blocked", "scope_blocked"} and task_id in self.sessions: readiness = self.browser_readiness(task_id) raise HTTPException(409, {"code": readiness["code"], "message": readiness["message"], "next_action": readiness["next_action"]}) from None raise HTTPException(409, {"message": str(error), "code": error.code}) from None def task_execution_profile(self, task_id: str): row = self.store.one("SELECT execution_profile FROM learning_tasks WHERE id=$1", (task_id,)) if not row: raise HTTPException(404, "学习任务不存在") try: execution_metadata(row["execution_profile"]) except ValueError: raise HTTPException(409, {"code": "EXECUTION_PROFILE_MISMATCH", "message": "任务执行模式不受支持，不能降级执行"}) from None return row["execution_profile"] def require_execution_metadata(self, task_id: str, value) -> None: if not execution_metadata_matches(value, self.task_execution_profile(task_id)): # An unproven session cannot keep a previous ready confirmation. self.sessions.pop(task_id, None) self.pause_active_task(task_id) raise HTTPException(409, {"code": "EXECUTION_PROFILE_MISMATCH", "message": "任务、执行器或页面证据的执行模式不一致，未认可本次执行"}) async def open(self, task_id: str, owner_id: str) -> dict: task = self.task(task_id, owner_id) self.require_route_access(task_id) scoped_account = hashlib.sha256(f"{owner_id}|{task['platform']}|{task['account_key']}".encode()).hexdigest() profile = self.task_execution_profile(task_id) result = await self.call(task_id, "open", {"account_key": scoped_account, "entry_url": task["entry_url"], **({"execution_profile": profile} if profile else {})}) self.sessions[task_id] = {**result, "status": "awaiting_user"} self.set_status(task_id, "awaiting_user") return result async def ready(self, task_id: str, owner_id: str, identity_label: str) -> dict: self.task(task_id, owner_id) self.require_route_access(task_id) if task_id not in self.sessions: raise HTTPException(409, {"code": "browser_session_required", "message": self.browser_readiness(task_id)["message"], "next_action": "open_browser"}) if re.search(r"(? str: target = Path(raw_path).resolve() if not target.is_relative_to(self.store.home) or target.suffix.lower() != ".png" or not target.is_file(): raise HTTPException(409, "执行器证据文件无效，未登记到验证记录") content = target.read_bytes() if not content.startswith(b"\x89PNG\r\n\x1a\n"): raise HTTPException(409, "执行器证据不是有效 PNG") artifact_id = identifier("evidence") self.store.execute("INSERT INTO artifacts(id,task_id,relative_path,sha256,bytes,created_at) VALUES($1,$2,$3,$4,$5,$6)", (artifact_id, task_id, target.relative_to(self.store.home).as_posix(), hashlib.sha256(content).hexdigest(), len(content), now())) return artifact_id def save_observation(self, task_id: str, capture: dict, note: str = "", source: str = "learning") -> dict: # Failure captures can reach this method without a successful call(). self.require_execution_metadata(task_id, capture) data = normalize_worker_unicode(deepcopy(capture)) raw_path = data.pop("screenshot_path", None) data["source"], data["note"] = source, note if raw_path: artifact_id = self.register_artifact(task_id, raw_path) data["artifact_id"] = artifact_id data["screenshot_url"] = "/api/learning/artifacts/" + artifact_id data["text"] = str(data.get("text", ""))[:30000] data["elements"] = list(data.get("elements") or [])[:300] if "request_observations" in data: data["request_observations"] = safe_request_observations(data["request_observations"]) if "popup_observations" in data: data["popup_observations"] = safe_popup_observations(data["popup_observations"]) observation_id, timestamp = identifier("observation"), now() self.store.execute("INSERT INTO observations(id,task_id,data_json,created_at) VALUES($1,$2,$3,$4)", (observation_id, task_id, encoded(data), timestamp)) return {"id": observation_id, "task_id": task_id, "created_at": timestamp, **data} def action_failure(self, task_id: str, kind: str, error: WorkerError) -> HTTPException: """Append actual failure captures; evidence trouble never hides the action error.""" diagnostics = safe_diagnostics(error.diagnostics, code=error.code, execution_profile=self.task_execution_profile(task_id)) metadata = {"attempt_id": identifier("attempt"), "code": error.code, "kind": kind, "phase": diagnostics["phase"], "diagnostics": diagnostics} observations, references, rejected = [], {}, False for moment, capture in (("before", error.before),): if not capture: continue try: observation = self.save_observation(task_id, {**capture, "action_failure": {**metadata, "moment": moment}}, "失败动作前" if moment == "before" else "失败动作后的实际页面状态", "action_failure") observations.append(observation["id"]) references[moment + "_observation_id"] = observation["id"] except Exception: # An invalid/missing screenshot or storage error cannot turn an # already failed action into another error, success, or retry. rejected = True batch_ids = [] for item in error.batch_evidence: try: observation = self.save_observation(task_id, {**item["capture"], "batch": {key: item[key] for key in ("batch_index", "added_count", "duplicate_count")}, "action_failure": metadata}, "失败动作的实际批次页面", "action_failure_batch") observations.append(observation["id"]) batch_ids.append(observation["id"]) except Exception: rejected = True if error.after: try: observation = self.save_observation(task_id, {**error.after, "action_failure": {**metadata, "moment": "after"}}, "失败动作后的实际页面状态", "action_failure") observations.append(observation["id"]) references["after_observation_id"] = observation["id"] except Exception: rejected = True return HTTPException(409, {"code": error.code, "message": "动作未完成，请先核对失败阶段及当前页面；未记为成功步骤，也不会自动重放", "action_failure": metadata, "failure_observation_ids": observations, "batch_evidence_ids": batch_ids, "error_codes": safe_error_codes([error.code, *error.error_codes]), **references, "evidence_status": "partial" if rejected and observations else "recorded" if observations else "unavailable"}) async def capture(self, task_id: str, owner_id: str, note: str = "") -> dict: self.task(task_id, owner_id) self.require_route_access(task_id) await self.require_ready(task_id, owner_id) capture = await self.call(task_id, "capture", {"note": note}) return self.save_observation(task_id, capture, note) async def action(self, task_id: str, owner_id: str, values: dict) -> dict: task = self.task(task_id, owner_id) self.require_route_access(task_id) try: values = Action.model_validate(values).model_dump(exclude_none=True) except ValidationError: raise HTTPException(422, "动作字段、CSS 定位、提取字段或预算不符合只读契约") from None kind = values["kind"] if kind == "scroll": values = {"direction": "down", "pixels": 800, "max_scrolls": 1, **values} elif kind == "wait": values = {"state": "visible", "timeout_ms": 10000, **values} elif kind == "extract": values = {"limit": 50, **values} await self.require_ready(task_id, owner_id) observations = self.observations(task_id) if not observations: raise HTTPException(409, "请先采集当前页面，再记录链路动作") if len(self.steps(task_id)) >= MAX_STEPS: raise HTTPException(409, "本次学习已达到 30 步上限，请拆分能力") last = observations[-1] root_selector = values.get("wait_scope") or values["selector"] element = next((item for item in last.get("elements", []) if item.get("selector") == root_selector), None) if values.get("wait_scope") and not wait_scope_evidence_valid(last, values["wait_scope"]): raise HTTPException(422, "集合等待必须绑定最近观察中唯一、可见且非敏感的普通读取容器") selector_evidence = None if kind == "wait" and values.get("state") == "hidden" and not element: selector_evidence = next((observation for observation in reversed(observations) if any(item.get("selector") == values["selector"] and not item.get("sensitive") for item in observation.get("elements", []))), None) body_exception = not values.get("wait_scope") and values["selector"] == "body" and kind in {"scroll", "wait"} if not body_exception and not selector_evidence and (not element or element.get("sensitive")): raise HTTPException(422, "请选择最近页面证据中的非敏感元素") if kind == "press" and (not element or element.get("kind") != "fill" or element.get("tag") not in {"input", "textarea"} or element.get("search_input") is not True or not element.get("unique")): raise HTTPException(422, "Enter 仅支持已观察且唯一的普通搜索输入框") if values.get("until_target") and not seek_evidence_valid(last, element, values["until_target"], values.get("scroll_scope")): raise HTTPException(422, "动态滚动需最近观察中的唯一普通滚动根和实际集合或根内定位依据") if values.get("row_selector") and (not element or not element.get("unique") or element.get("tag") in {"input", "textarea", "select", "script", "style", "noscript", "iframe", "button", "form"}): raise HTTPException(422, "容器内行提取必须绑定最近观察中唯一普通容器") if values.get("batch"): scroll = next((item for item in last.get("elements", []) if item.get("selector") == values["batch"]["scroll_selector"]), None) if not scroll or scroll.get("sensitive") or not scroll.get("unique") or type((scroll.get("scroll_y") or {}).get("can_scroll")) is not bool or scroll.get("tag") in {"input", "textarea", "select", "script", "style", "noscript", "iframe", "button", "form"}: raise HTTPException(422, "批次滚动必须绑定最近观察中唯一普通可滚动容器") parameter, value = values.get("parameter"), values.get("value", "") if parameter: if parameter not in task["parameters"]: raise HTTPException(422, "参数未在学习任务中声明") value = task["parameters"][parameter] target_parameter = values.get("target", {}).get("parameter") if target_parameter and target_parameter not in task["parameters"]: raise HTTPException(422, "动态对象参数未在学习任务中声明") seek_parameter = values.get("until_target", {}).get("parameter") if seek_parameter and seek_parameter not in task["parameters"]: raise HTTPException(422, "滚动目标参数未在学习任务中声明") if kind in {"read", "extract"} and not NAME.fullmatch(values.get("output_name") or ""): raise HTTPException(422, "读取步骤需要输出字段名，例如 results") wire = deepcopy(values) wire.pop("parameter", None) if kind == "fill": wire["value"] = value if target_parameter: wire["target"].pop("parameter", None) wire["target"]["value"] = task["parameters"][target_parameter] if seek_parameter: wire["until_target"].pop("parameter", None) wire["until_target"]["value"] = task["parameters"][seek_parameter] try: result = await self.call(task_id, "action", wire, preserve_failure=True) except WorkerError as error: raise self.action_failure(task_id, kind, error) from None if values.get("wait_scope") and not wait_scope_evidence_valid(result.get("before") or last, values["wait_scope"]): raise self.action_failure(task_id, kind, WorkerError("集合等待根证据无效", "WAIT_SCOPE_CHANGED", before=result.get("before"), after=result.get("after"))) if values.get("until_target"): action_result = result.get("result") if isinstance(result.get("result"), dict) else {} summary = {"step_position": 1, **{key: action_result.get(key) for key in ("scrolls", "movements", "target_reached", "stop_reason")}} if not safe_seek_result(summary, values, 1) or summary["target_reached"] is not True: raise self.action_failure(task_id, kind, WorkerError("滚动定位结果无效", "SCROLL_TARGET_RESULT_INVALID", before=result.get("before"), after=result.get("after"))) if kind == "extract": checks = output_assertion_checks({values["output_name"]: result.get("result")}, {values["output_name"]: []}, {values["output_name"]: values}) if not checks[0]["passed"]: raise self.action_failure(task_id, kind, WorkerError("输出契约不符", "OUTPUT_CONTRACT_MISMATCH", before=result.get("before"), after=result.get("after"), batch_evidence=result.get("batch_evidence"))) batches = result.get("batch_evidence") or [] if values.get("batch"): summary = result["result"]["batch_summary"] if len(batches) != summary["batches_read"] or [item["batch_index"] for item in batches] != list(range(1, len(batches) + 1)) or sum(item["added_count"] for item in batches) != result["result"]["count"] or sum(item["duplicate_count"] for item in batches) != summary["duplicates_removed"]: raise self.action_failure(task_id, kind, WorkerError("批次证据不符", "BATCH_EVIDENCE_INVALID", before=result.get("before"), after=result.get("after"), batch_evidence=batches)) before = self.save_observation(task_id, result["before"], "动作前", "action") if result.get("before") else last batch_observations = [] if values.get("batch"): try: for item in batches: batch_observations.append(self.save_observation(task_id, {**item["capture"], "batch": {key: item[key] for key in ("batch_index", "added_count", "duplicate_count")}}, "真实批次页面", "action_batch")) except Exception: raise self.action_failure(task_id, kind, WorkerError("批次证据无法登记", "BATCH_EVIDENCE_INVALID", before=result.get("before"), after=result.get("after"), batch_evidence=batches)) from None after = self.save_observation(task_id, result["after"], "动作后", "action") if result.get("after") else None step = {**values, "evidence_id": before["id"], "after_evidence_id": after["id"] if after else None} if values.get("batch"): step["batch_evidence_ids"] = [item["id"] for item in batch_observations] if kind == "wait" and values.get("state") == "hidden" and values["selector"] != "body": # A disappeared element cannot occur in the fresh action snapshot. # Freeze the prior same-task evidence that established its selector. source = selector_evidence or next((observation for observation in reversed(observations) if any(item.get("selector") == values["selector"] and not item.get("sensitive") for item in observation.get("elements", []))), None) if source: step["selector_evidence_id"] = source["id"] with self.store.connect(immediate=True) as conn: position = conn.execute("SELECT COALESCE(MAX(position),0)+1 AS position FROM learning_steps WHERE task_id=$1", (task_id,)).fetchone()["position"] step_id, timestamp = identifier("step"), now() conn.execute("INSERT INTO learning_steps(id,task_id,position,data_json,created_at) VALUES($1,$2,$3,$4,$5)", (step_id, task_id, position, encoded(step), timestamp)) return {"step": {"id": step_id, "position": position, "created_at": timestamp, **step}, "result": result.get("result"), "before": before, "after": after, **({"batch_observations": batch_observations} if values.get("batch") else {})} async def close(self, task_id: str, owner_id: str) -> dict: self.task(task_id, owner_id) self.require_route_access(task_id) result = await self.call(task_id, "close", {}) self.sessions.pop(task_id, None) self.pause_active_task(task_id) return result def create_candidate(self, task_id: str, owner_id: str, name: str | None = None, step_ids: list[str] | None = None) -> dict: task = self.task(task_id, owner_id) profile = self.task_execution_profile(task_id) recorded = self.steps(task_id) if step_ids is not None: if (not isinstance(step_ids, list) or not 1 <= len(step_ids) <= MAX_STEPS or any(not isinstance(key, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,120}", key) for key in step_ids) or len(set(step_ids)) != len(step_ids)): raise HTTPException(422, "候选选步须为1至30个唯一的本任务真实步骤编号") lookup = {step["id"]: step for step in recorded} if any(key not in lookup for key in step_ids): raise HTTPException(422, "候选选步包含不存在或不属于本任务的步骤") recorded = [lookup[key] for key in step_ids] steps = [{k: v for k, v in step.items() if k not in ("id", "created_at", "position")} for step in recorded] references = sorted(step_evidence_references(steps)) bindings = [] for reference in references: evidence = self.evidence_snapshot(task_id, reference) if profile and (not evidence or not execution_metadata_matches(evidence["data"], profile)): raise HTTPException(409, {"code": "EXECUTION_PROFILE_MISMATCH", "message": "所选步骤缺少相同执行模式的实际页面证据，请重新实录"}) bindings.append({"observation_id": reference, "content_digest": digest(evidence) if evidence else None}) manifest = {"schema_version": 2, "integrity_version": 1, "entry_url": task["entry_url"], "parameters": deepcopy(task["parameters"]), "steps": steps, "evidence_bindings": bindings, "platform": task["platform"], "identity_requirement": "operator_confirmed_session", "scope": "readonly_learning"} manifest.update(execution_metadata(profile)) if step_ids is not None: manifest["source_step_ids"] = list(step_ids) candidate_id = identifier("candidate") with self.store.connect(immediate=True) as conn: version = conn.execute("SELECT COALESCE(MAX(version),0)+1 AS version FROM candidates WHERE task_id=$1", (task_id,)).fetchone()["version"] conn.execute("INSERT INTO candidates(id,task_id,name,version,digest,manifest_json,created_at) VALUES($1,$2,$3,$4,$5,$6,$7)", (candidate_id, task_id, name or task["name"], version, digest(manifest), encoded(manifest), now())) self.set_status(task_id, "candidate") return self.candidate(candidate_id, owner_id) def evidence_snapshot(self, task_id: str, observation_id: str) -> dict | None: row = self.store.one("SELECT * FROM observations WHERE id=$1 AND task_id=$2", (observation_id, task_id)) if not row: return None data = json.loads(row["data_json"]) artifact = self.store.one("SELECT id,sha256,bytes FROM artifacts WHERE id=$1 AND task_id=$2", (data.get("artifact_id", ""), task_id)) return {"observation_id": row["id"], "task_id": task_id, "created_at": row["created_at"], "data": data, "artifact": artifact} def source_integrity_errors(self, candidate: dict) -> list[str]: manifest, task_id = candidate["manifest"], candidate["task_id"] if manifest.get("integrity_version") != 1: return ["候选尚未冻结动作前后证据，请基于原任务生成新版本"] bindings = manifest.get("evidence_bindings") or [] references = {item.get("observation_id") for item in bindings} current = {reference: snapshot for reference in references if (snapshot := self.evidence_snapshot(task_id, reference))} errors = evidence_graph_errors(manifest.get("steps") or [], bindings, current) profile = self.task_execution_profile(task_id) if not execution_metadata_matches(manifest, profile): errors.append("候选与任务执行模式不一致") if any(not execution_metadata_matches(snapshot["data"], profile) for snapshot in current.values()): errors.append("冻结来源证据与候选执行模式不一致") if "source_step_ids" in manifest: selected = manifest["source_step_ids"] lookup = {step["id"]: {key: value for key, value in step.items() if key not in {"id", "created_at", "position"}} for step in self.steps(task_id)} if (not isinstance(selected, list) or not 1 <= len(selected) <= MAX_STEPS or any(not isinstance(key, str) or key not in lookup for key in selected) or len(set(selected)) != len(selected) or [lookup[key] for key in selected] != manifest.get("steps")): errors.append("候选选步编号、顺序或步骤内容与本任务真实记录不一致") for snapshot in current.values(): artifact = snapshot["artifact"] if not artifact: errors.append("冻结来源证据缺少关联截图") continue stored = self.store.one("SELECT relative_path FROM artifacts WHERE id=$1 AND task_id=$2", (artifact["id"], task_id)) target = (self.store.home / stored["relative_path"]).resolve() if not target.is_relative_to(self.store.home) or not target.is_file(): errors.append("冻结来源截图不存在或越出证据目录") elif hashlib.sha256(target.read_bytes()).hexdigest() != artifact["sha256"]: errors.append("冻结来源截图摘要不匹配") return list(dict.fromkeys(errors)) def add_validation(self, candidate: dict, kind: str, errors: list[str], **data) -> dict: record_id, timestamp = identifier("validation"), now() content = {"candidate_digest": candidate["digest"], "errors": errors, "parameters": {}, "expected_outputs": {}, "actual_outputs": {}, "environment": {}, "evidence": [], "sample_name": "", "duration_ms": 0, "model_calls": 0, "model_usage_status": "not_used", **data} status = "failed" if errors else "passed" self.store.execute("INSERT INTO validations(id,candidate_id,task_id,kind,status,data_json,created_at) VALUES($1,$2,$3,$4,$5,$6,$7)", (record_id, candidate["id"], candidate["task_id"], kind, status, encoded(content), timestamp)) return {"id": record_id, "candidate_id": candidate["id"], "task_id": candidate["task_id"], "kind": kind, "status": status, "created_at": timestamp, **content} def static_errors(self, candidate: dict) -> list[str]: manifest = candidate["manifest"] errors = [] if digest(manifest) != candidate["digest"]: errors.append("能力清单校验值不匹配") if manifest.get("schema_version") not in {1, 2}: errors.append("不支持的能力契约版本") errors.extend(self.source_integrity_errors(candidate)) steps = manifest.get("steps") or [] if not steps: errors.append("候选没有实际记录的步骤") if len(steps) > MAX_STEPS: errors.append("候选超过步骤预算") output_names = set() for position, step in enumerate(steps, 1): prefix = f"第 {position} 步" kinds = ("fill", "click", "press", "read", "scroll", "wait", "extract") if manifest.get("schema_version") == 2 else ("fill", "click", "read") if step.get("kind") not in kinds: errors.append(prefix + "动作不在首版只读范围内") if manifest.get("schema_version") == 2: try: Action.model_validate({key: value for key, value in step.items() if key not in {"evidence_id", "after_evidence_id", "selector_evidence_id", "batch_evidence_ids"}}) except ValidationError: errors.append(prefix + "动作字段或预算不符合固定契约") fixed_fields = {"press": {"key"}, "scroll": {"direction", "pixels", "max_scrolls"}, "wait": {"state", "timeout_ms"}, "extract": {"limit", "fields"}}.get(step.get("kind"), set()) if not fixed_fields.issubset(step): errors.append(prefix + "缺少已固化的动作参数或预算") if not isinstance(step.get("selector"), str) or not step["selector"].strip(): errors.append(prefix + "缺少定位规则") selector_reference = step.get("selector_evidence_id") if step.get("kind") == "wait" and step.get("state") == "hidden" else step.get("evidence_id") observation = self.store.one("SELECT data_json FROM observations WHERE id=$1 AND task_id=$2", (selector_reference or "", candidate["task_id"])) if not observation: errors.append(prefix + "没有同任务来源证据") else: observed = json.loads(observation["data_json"]) root_selector = step.get("wait_scope") or step.get("selector") element = next((item for item in observed.get("elements", []) if item.get("selector") == root_selector), None) if step.get("wait_scope") and not wait_scope_evidence_valid(observed, step["wait_scope"]): errors.append(prefix + "集合等待缺少同页唯一可见普通读取根依据") if step.get("until_target") and not seek_evidence_valid(observed, element, step["until_target"], step.get("scroll_scope")): errors.append(prefix + "动态滚动缺少实际滚动根与集合定位依据") body_exception = not step.get("wait_scope") and step.get("selector") == "body" and step.get("kind") in {"scroll", "wait"} if not body_exception and (not element or element.get("sensitive")): errors.append(prefix + "定位规则缺少非敏感页面依据") elif not body_exception: kind = step.get("kind") if (kind in {"fill", "click", "press", "scroll"} or (kind == "wait" and step.get("state") != "hidden" and not step.get("min_count"))) and not element.get("unique") and not step.get("target"): errors.append(prefix + "操作目标不唯一") if kind == "fill" and element.get("kind") != "fill": errors.append(prefix + "填写动作没有普通输入框依据") if kind == "press" and (step.get("key") != "Enter" or element.get("kind") != "fill" or element.get("tag") not in {"input", "textarea"} or element.get("search_input") is not True): errors.append(prefix + "搜索 Enter 没有明确的普通搜索输入框依据") if kind == "click" and not step.get("target") and element.get("kind") != "click": errors.append(prefix + "点击动作没有链接或按钮依据") if kind in {"wait", "scroll", "extract"} and element.get("tag") in {"script", "style", "noscript", "iframe", "input", "textarea", "select"}: errors.append(prefix + "动作不能以脚本或表单控件作为读取根") if step.get("row_selector") and (not element.get("unique") or element.get("tag") in {"button", "form"}): errors.append(prefix + "行提取缺少唯一普通容器依据") if step.get("batch"): scroll = next((item for item in observed.get("elements", []) if item.get("selector") == step["batch"].get("scroll_selector")), None) if not scroll or scroll.get("sensitive") or not scroll.get("unique") or type((scroll.get("scroll_y") or {}).get("can_scroll")) is not bool or scroll.get("tag") in {"script", "style", "noscript", "iframe", "input", "textarea", "select", "button", "form"}: errors.append(prefix + "批次缺少同页唯一可滚动容器依据") if step.get("kind") == "fill" and step.get("parameter") and step["parameter"] not in manifest.get("parameters", {}): errors.append(prefix + "参数未声明") if step.get("kind") == "fill" and not step.get("parameter") and not isinstance(step.get("value"), str): errors.append(prefix + "缺少明确输入值或参数绑定") if step.get("target", {}).get("parameter") and step["target"]["parameter"] not in manifest.get("parameters", {}): errors.append(prefix + "动态对象参数未声明") if isinstance(step.get("until_target"), dict) and step["until_target"].get("parameter") and step["until_target"]["parameter"] not in manifest.get("parameters", {}): errors.append(prefix + "滚动目标参数未声明") if step.get("kind") in {"read", "extract"}: name = step.get("output_name", "") if not NAME.fullmatch(name) or name in output_names: errors.append(prefix + "输出名缺失、非法或重复") output_names.add(name) if not output_names: errors.append("首版候选至少需要一个可核对的读取输出") return errors def check(self, candidate_id: str, owner_id: str) -> dict: candidate = self.candidate(candidate_id, owner_id) return self.add_validation(candidate, "static", self.static_errors(candidate), environment={"validator": "learning-v1", "os": os_name(platform.system())}) async def verify(self, candidate_id: str, owner_id: str, values: dict) -> dict: candidate = self.candidate(candidate_id, owner_id) self.require_route_access(candidate["task_id"]) manifest = candidate["manifest"] static = self.store.one(f"SELECT status FROM validations WHERE candidate_id=$1 AND kind='static' ORDER BY {self.store.order_column()} DESC LIMIT 1", (candidate_id,)) if not static or static["status"] != "passed" or self.static_errors(candidate): raise HTTPException(409, "请先通过候选固定校验，再进行浏览器试跑") expected = values.get("expected_outputs") or {} contracts = {step["output_name"]: step for step in manifest["steps"] if step["kind"] in {"read", "extract"}} output_names = set(contracts) if not expected or not set(expected).issubset(output_names): raise HTTPException(422, "请为候选的读取输出填写至少一项预期文本") for output_name, expectation in expected.items(): if isinstance(expectation, dict): if expectation == {"state": "nonempty"} and contracts[output_name].get("kind") == "extract": continue if expectation != {"state": "empty"} or contracts[output_name].get("allow_empty") is not True: raise HTTPException(422, "空结果预期只允许已固化明确空态分支") continue items = [expectation] if isinstance(expectation, str) else expectation if not isinstance(items, list) or not items or any(not isinstance(item, str) or not item.strip() or len(item) > 2000 for item in items): raise HTTPException(422, "预期结果需为非空文本或文本列表") overrides = check_parameters(values.get("parameters") or {}) if not set(overrides).issubset(manifest["parameters"]): raise HTTPException(422, "试跑包含候选未声明的参数") parameters = {**manifest["parameters"], **overrides} binding = values.get("note_binding") if binding is not None: if not isinstance(binding, dict) or set(binding) != {"note_id"} or not isinstance(binding["note_id"], str) or not re.fullmatch(r"[a-fA-F0-9]{24}", binding["note_id"]) or not detail_contract(manifest): raise HTTPException(422, "笔记来源绑定只支持已固化的单篇正文与评论读取合同") binding = {"note_id": binding["note_id"].lower()} mode = values.get("mode", "exploration") baseline = values.get("baseline_note", "").strip() window = values.get("observation_window_minutes", 60) if mode not in {"exploration", "runtime"} or (mode == "runtime" and not baseline): raise HTTPException(422, "正式运行模式须先记录本次官方通知与账号功能基线") if type(window) is not int or not 1 <= window <= 10080: raise HTTPException(422, "观察窗口需为 1 至 10080 分钟") await self.require_ready(candidate["task_id"], owner_id) self.require_execution_metadata(candidate["task_id"], manifest) self.require_execution_metadata(candidate["task_id"], self.sessions.get(candidate["task_id"])) started_at = now() started = time.monotonic() environment = {"os": os_name(platform.system()), "python_version": platform.python_version(), "runner": "playwright-learning-v1", "identity_verification": "operator_confirmed"} evidence, errors, outputs, error_codes = [], [], {}, [] binding_result, binding_passed = None, False session_closed = False seek_steps = {position: step for position, step in enumerate(manifest["steps"], 1) if step.get("until_target")} seek_data = {"seek_results": []} if seek_steps else {} timing_data, failure_data = {}, {} try: result = await self.call(candidate["task_id"], "verify", {"schema_version": manifest["schema_version"], "entry_url": manifest["entry_url"], "steps": manifest["steps"], "parameters": parameters, "expected_outputs": expected, **execution_metadata(manifest.get("execution_profile")), **({"note_binding": binding} if binding else {})}) environment.update(execution_metadata(result.get("execution_profile"))) session_closed = result.get("session_closed") is True if session_closed: environment["browser_session_closed"] = True outputs = result.get("outputs") or {} error_codes = safe_error_codes(result.get("error_codes")) errors = [str(error) for error in result.get("errors", [])] failure = safe_failure_diagnostics(result.get("failure_diagnostics"), manifest["steps"], error_codes, passed=result.get("passed"), execution_profile=manifest.get("execution_profile")) if failure: failure_data["failure_diagnostics"] = failure if "step_timings" in result or "verify_timing" in result: timing_data = safe_verify_timings(result.get("step_timings"), result.get("verify_timing"), manifest["steps"], passed=result.get("passed") is True) if timing_data is None: timing_data = {} errors.append("执行耗时记录不符合当前步骤及有限阶段契约") error_codes = safe_error_codes([*error_codes, "TIMING_TRACE_INVALID"]) environment["browser_version"] = str(result.get("browser_version") or "unknown") expected_budget = 90000 + (manifest["steps"][3]["timeout_ms"] if binding and len(manifest["steps"]) == 8 else 0) budget = result.get("verify_budget_ms") if type(budget) is int and budget == expected_budget: environment["verify_budget_ms"] = budget elif budget is not None or (binding and len(manifest["steps"]) == 8): errors.append("执行器实际试跑时间预算与当前链路不一致") error_codes = safe_error_codes([*error_codes, "VERIFY_BUDGET_INVALID"]) if seek_steps: raw_results = result.get("seek_results") cleaned, positions = [], [] valid_trace = isinstance(raw_results, list) and len(raw_results) <= len(seek_steps) for item in raw_results if valid_trace else []: position = item.get("step_position") if isinstance(item, dict) else None step = seek_steps.get(position) if type(position) is int else None summary = safe_seek_result(item, step, position) if step else None if not summary or position in positions: valid_trace = False break cleaned.append(summary) positions.append(position) if result.get("passed") and (set(positions) != set(seek_steps) or any(not item["target_reached"] for item in cleaned)): valid_trace = False if valid_trace: seek_data["seek_results"] = cleaned else: errors.append("滚动定位执行摘要缺失或不符合既定预算") error_codes = safe_error_codes([*error_codes, "SEEK_RESULT_INVALID"]) if environment["browser_version"] == "unknown": errors.append("执行器没有提供实际浏览器版本") if result.get("os") and os_name(result["os"]) != environment["os"]: errors.append("执行器返回的系统与本机实测系统不一致") for capture in result.get("evidence", []): if isinstance(capture, dict): observation = self.save_observation(candidate["task_id"], capture, values.get("sample_name", "变参试跑"), "verification") evidence.append({"observation_id": observation["id"], "artifact_id": observation.get("artifact_id"), "screenshot_url": observation.get("screenshot_url"), "url": observation.get("url"), "title": observation.get("title")}) errors.extend(compare_outputs(outputs, expected, contracts)) if binding: binding_result, binding_passed = audit_note_binding(manifest, binding, result.get("note_binding_result"), outputs, parameters) if not binding_passed: errors.append("实际笔记来源或标题未通过绑定核对，不能认定为所选笔记") error_codes = safe_error_codes([*error_codes, "NOTE_BINDING_MISMATCH"]) if not result.get("passed") and not errors: errors.append("执行器未确认试跑成功") if not evidence: errors.append("试跑缺少页面证据") except HTTPException as error: detail = error.detail # Missing process-owned session is an operational precondition, not a # failed business assertion. Do not fabricate a new validation result. if isinstance(detail, dict) and detail.get("code", "").lower() in ("session_closed", "session_not_found", "worker_closed", "worker_timeout", "awaiting_user", "route_busy"): raise errors.append(detail.get("message", "试跑未完成") if isinstance(detail, dict) else str(detail)) if isinstance(detail, dict): error_codes = safe_error_codes([*error_codes, detail.get("code")]) finished_at = now() binding_data = {**timing_data, **seek_data, **(failure_data if errors else {}), **({"note_binding": binding, "note_binding_result": binding_result} if binding else {})} checks = output_assertion_checks(outputs, expected, contracts) if binding: checks.append({"name": "note_source_binding", "passed": binding_passed, "message": "实际笔记编号、标题和来源已核对" if binding_passed else "实际笔记来源未通过绑定核对"}) due = (datetime.fromisoformat(finished_at) + timedelta(minutes=window)).isoformat(timespec="milliseconds") if mode == "runtime" else None record = self.add_validation(candidate, "browser", list(dict.fromkeys(errors)), parameters=parameters, expected_outputs=expected, actual_outputs=outputs, error_codes=error_codes, environment=environment, evidence=evidence, sample_name=values.get("sample_name", "变参试跑"), duration_ms=round((time.monotonic() - started) * 1000), mode=mode, baseline_note=baseline, observation_window_minutes=window, started_at=started_at, finished_at=finished_at, observation_due_at=due, assertion_checks=checks, **binding_data, verification_basis="synthetic_browser_fixture" if manifest.get("platform") == "demo" else "browser_execution_with_fixed_assertions", independent_readback={"status": "not_configured", "independent_identity": False, "message": "当前保存浏览器执行输出并由后台固定比较；尚未接入独立身份与来源读回，不能等同 A2A 的独立业务核验"}) record["quality_observations"] = [] record["quality"] = self.validation_quality(record) self.set_status(candidate["task_id"], "paused" if session_closed else "validated" if record["status"] == "passed" else "validation_failed") return record def publish_check(self, candidate_id: str, owner_id: str) -> dict: candidate = self.candidate(candidate_id, owner_id) rows = self.validations(owner_id, candidate_id=candidate_id) static = next((row for row in rows if row["kind"] == "static"), None) passed = [row for row in rows if row["kind"] == "browser" and row["status"] == "passed" and row.get("candidate_digest") == candidate["digest"]] samples = {encoded(row["parameters"]) for row in passed} systems = {row.get("environment", {}).get("os") for row in passed} evidence_ok = bool(passed) for row in passed: if not row.get("evidence"): evidence_ok = False for item in row.get("evidence", []): if not item.get("artifact_id"): evidence_ok = False continue try: self.artifact(item["artifact_id"], owner_id) except HTTPException: evidence_ok = False checks = [ {"name": "候选结构与来源校验", "passed": bool(static and static["status"] == "passed" and not self.static_errors(candidate)), "detail": "引用同版本固定校验与原始页面证据"}, {"name": "至少两组不同参数试跑", "passed": len(samples) >= 2, "detail": f"已通过 {len(samples)} 组不同参数"}, {"name": "Windows 实际试跑", "passed": "windows" in systems, "detail": "只认可当前执行主机记录的真实系统"}, {"name": "macOS 实际试跑", "passed": "macos" in systems, "detail": "未在 macOS 执行不能计为通过"}, {"name": "试跑证据完整", "passed": evidence_ok, "detail": "各通过记录需具备关联截图证据"}, {"name": "独立来源读回", "passed": False, "detail": "本轮仅完成浏览器执行与固定断言，尚未接入独立身份的来源业务读回"}, {"name": "正式模式及任务后质量观察", "passed": candidate["quality"]["state"] == "verified_in_scope", "detail": "只认可正式模式成功且观察窗口到期后的人工过程与反馈检查；不构成商家发布"}, ] return {"ready": all(item["passed"] for item in checks), "checks": checks, "candidate_id": candidate_id, "digest": candidate["digest"], "published": False, "message": "当前阶段提供发布预检，尚未接入商家能力分发"} def artifact(self, artifact_id: str, owner_id: str) -> Path: row = self.store.one("SELECT a.* FROM artifacts a JOIN learning_tasks t ON t.id=a.task_id WHERE a.id=$1 AND t.owner_id=$2", (artifact_id, owner_id)) if not row: raise HTTPException(404, "证据不存在") target = (self.store.home / row["relative_path"]).resolve() if not target.is_relative_to(self.store.home) or not target.is_file(): raise HTTPException(404, "证据文件不存在") if hashlib.sha256(target.read_bytes()).hexdigest() != row["sha256"]: raise HTTPException(409, "证据内容已变化，不能当作原验证依据") return target async def demo(self, owner_id: str, base_url: str) -> dict: task = self.create_task(owner_id, {"name": "本机搜索详情评论示例", "platform": "demo", "entry_url": base_url.rstrip("/") + "/api/learning/demo-page", "goal": "记录搜索、打开详情与读取评论，通过两组参数核对当前对象与输出", "account_key": "local-demo", "parameters": {"keyword": "建材"}}) task_id = task["id"] try: await self.open(task_id, owner_id) await self.ready(task_id, owner_id, "本机演示页面 无平台账号") await self.capture(task_id, owner_id, "示例初始页面") await self.action(task_id, owner_id, {"kind": "fill", "selector": '[data-testid="search-input"]', "parameter": "keyword"}) await self.action(task_id, owner_id, {"kind": "click", "selector": '[data-testid="search-button"]'}) await self.action(task_id, owner_id, {"kind": "read", "selector": '[data-testid="result"]', "output_name": "results"}) await self.action(task_id, owner_id, {"kind": "click", "selector": '[data-testid="detail-link"]'}) await self.action(task_id, owner_id, {"kind": "read", "selector": '[data-testid="comment"]', "output_name": "comments"}) candidate = self.create_candidate(task_id, owner_id) self.check(candidate["id"], owner_id) for keyword in ("建材", "装修"): await self.verify(candidate["id"], owner_id, {"parameters": {"keyword": keyword}, "expected_outputs": {"results": keyword, "comments": keyword}, "sample_name": f"关键词 {keyword}"}) return {"task_id": task_id, "candidate_id": candidate["id"]} finally: if task_id in self.sessions: try: await self.close(task_id, owner_id) except HTTPException: pass