419 lines
24 KiB
Python
419 lines
24 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import uuid
|
||
from copy import deepcopy
|
||
from pathlib import Path
|
||
from urllib.parse import unquote, urlsplit
|
||
|
||
from .learning_contracts import NAVIGATION_PROFILE, execution_metadata_matches
|
||
|
||
|
||
FAILURE_PHASES = {"validate", "inspect_current", "capture_before", "resolve_target", "execute", "settle", "request_guard", "capture_after"}
|
||
PRIVATE_PATH = re.compile(r"password|passwd|credential|token|secret|session|cookie|authorization|captcha|verification|phone|mobile|验证码|密码|手机号", re.I)
|
||
LOGIN_EVIDENCE = {"visible_current_account_navigation", "visible_login_dialog", "visible_login_entry", "visible_verification_widget", "no_conclusive_visible_evidence", "unsupported_platform", "inspection_unavailable"}
|
||
TIMING_PHASES = {"validate", "inspect_current", "capture_before", "resolve_target", "execute", "settle", "request_guard", "capture_after", "failure_capture", "batch_capture"}
|
||
REQUEST_POLICY_RULES = {
|
||
"xhs-search-experimental-v1": {"collect", "search_history", "search_notes", "search_onebox", "search_dqa"},
|
||
"xhs-search-experimental-v2": {"collect", "search_history", "search_notes", "search_onebox", "search_dqa", "apm_data"},
|
||
"xhs-search-experimental-v3": {"collect", "search_history", "search_notes", "search_onebox", "search_dqa", "apm_data", "apm_data_xhr"},
|
||
"xhs-note-open-experimental-v1": {"collect", "note_share_code", "note_widgets", "note_feed", "note_seo", "note_metrics", "apm_data"},
|
||
"xhs-note-open-experimental-v2": {"collect", "note_share_code", "note_widgets", "note_feed", "note_seo", "note_metrics", "apm_data", "note_history_report"},
|
||
"xhs-readonly-ui-experimental-v1": {"collect", "apm_data"},
|
||
"xhs-search-scroll-experimental-v1": {"collect", "apm_data", "search_notes"},
|
||
"xhs-page-native-v1": {"native_page_traffic"},
|
||
"xhs-page-navigation-v1": {"native_page_traffic"},
|
||
}
|
||
REQUEST_RESOURCE_TYPES = {"document", "fetch", "xhr", "other", "script", "stylesheet", "image", "font", "media", "manifest", "websocket", "eventsource", "texttrack"}
|
||
ACTIVE_REQUEST_POLICIES = {"xhs-search-experimental-v3", "xhs-note-open-experimental-v2", "xhs-readonly-ui-experimental-v1", "xhs-search-scroll-experimental-v1", "xhs-page-native-v1", "xhs-page-navigation-v1"}
|
||
REQUEST_POLICY_REASONS = {"missing_policy", "inactive_policy", "invalid_time", "policy_expired", "automation_inactive", "session_out_of_scope", "page_out_of_scope", "not_main_frame", "method_not_allowed", "page_changed", "policy_not_started", "invalid_url", "request_out_of_scope", "resource_type_not_allowed", "navigation_out_of_scope"}
|
||
|
||
|
||
def safe_login_detection(value) -> dict | None:
|
||
if not isinstance(value, dict):
|
||
return None
|
||
state = value.get("state")
|
||
if not isinstance(state, str) or state not in {"logged_in", "logged_out", "verification_required", "unknown"} or value.get("basis") != "visible_page":
|
||
return None
|
||
timestamp = value.get("observed_at")
|
||
if not isinstance(timestamp, str) or not re.fullmatch(r"\d{4}-\d\d-\d\dT[0-9:.+Z-]{8,25}", timestamp):
|
||
return None
|
||
evidence = value.get("evidence")
|
||
return {"state": state, "basis": "visible_page", "observed_at": timestamp,
|
||
"evidence": [item for item in evidence[:7] if isinstance(item, str) and item in LOGIN_EVIDENCE] if isinstance(evidence, list) else []}
|
||
|
||
|
||
def safe_worker_code(value) -> str:
|
||
return value if isinstance(value, str) and re.fullmatch(r"[A-Za-z][A-Za-z0-9_]{0,63}", value) else "worker_error"
|
||
|
||
|
||
def safe_request_summary(value, *, blocked: bool = False) -> dict | None:
|
||
"""A second boundary: never accept arbitrary worker error/request payloads."""
|
||
if not isinstance(value, dict) or not isinstance(value.get("method"), str) or value["method"] not in {"GET", "HEAD", "OPTIONS", "POST", "PUT", "PATCH", "DELETE", "CONNECT", "TRACE"}:
|
||
return None
|
||
if not isinstance(value.get("resource_type"), str) or value["resource_type"] not in REQUEST_RESOURCE_TYPES:
|
||
return None
|
||
location = safe_location(value.get("origin"), value.get("pathname"))
|
||
if not location:
|
||
return None
|
||
summary = {"method": value["method"], **location, "resource_type": value["resource_type"]}
|
||
if blocked:
|
||
reason, version = value.get("policy_reason"), value.get("active_policy_version")
|
||
if isinstance(reason, str) and reason in REQUEST_POLICY_REASONS:
|
||
summary["policy_reason"] = reason
|
||
if isinstance(version, str) and version in ACTIVE_REQUEST_POLICIES:
|
||
summary["active_policy_version"] = version
|
||
expected = value.get("expected_resource_types")
|
||
if reason == "resource_type_not_allowed" and isinstance(expected, list):
|
||
types = list(dict.fromkeys(item for item in expected if isinstance(item, str) and item in REQUEST_RESOURCE_TYPES))[:4]
|
||
if types:
|
||
summary["expected_resource_types"] = types
|
||
return summary
|
||
|
||
|
||
def safe_location(origin_value, raw_path) -> dict | None:
|
||
try:
|
||
parsed = urlsplit(origin_value) if isinstance(origin_value, str) else None
|
||
if not parsed:
|
||
return None
|
||
if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}:
|
||
return None
|
||
parsed.port
|
||
origin = f"{parsed.scheme}://{parsed.netloc}"
|
||
except (ValueError, TypeError):
|
||
return None
|
||
if not isinstance(raw_path, str) or not raw_path.startswith("/"):
|
||
return None
|
||
segments = raw_path.split("?", 1)[0].split("#", 1)[0].split("/")
|
||
path = []
|
||
for segment in segments:
|
||
decoded = unquote(segment)
|
||
if PRIVATE_PATH.search(decoded):
|
||
path.append(":redacted")
|
||
elif len(decoded) > 64 or "@" in decoded or re.search(r"1[3-9]\d{9}", decoded) or re.fullmatch(r"[a-f0-9-]{16,}|[A-Za-z0-9_=-]{32,}", decoded) or re.search(r"[\x00-\x1f]", decoded):
|
||
path.append(":id")
|
||
else:
|
||
path.append(segment)
|
||
return {"origin": origin, "pathname": "/".join(path)[:300]}
|
||
|
||
|
||
def safe_popup_observations(value) -> list[dict]:
|
||
records = []
|
||
for item in value[-10:] if isinstance(value, list) else []:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
allowed = {"phase": {"manual", "automation", "confirmed"}, "decision": {"adopted", "blocked"},
|
||
"reason": {"same_origin_manual", "same_origin_click", "unconfirmed_origin", "outside_scope", "unrequested_popup"}}
|
||
if any(not isinstance(item.get(key), str) or item[key] not in options for key, options in allowed.items()):
|
||
continue
|
||
sequence, timestamp = item.get("sequence"), item.get("recorded_at")
|
||
if type(sequence) is not int or not 0 <= sequence <= 2 ** 53 or not isinstance(timestamp, str) or not re.fullmatch(r"\d{4}-\d\d-\d\dT[0-9:.+Z-]{8,25}", timestamp):
|
||
continue
|
||
record = {"sequence": sequence, "recorded_at": timestamp, **{key: item[key] for key in allowed}}
|
||
location = safe_location(item.get("origin"), item.get("pathname"))
|
||
if location:
|
||
record.update(location)
|
||
records.append(record)
|
||
return records
|
||
|
||
|
||
def safe_request_observations(value) -> list[dict]:
|
||
records = []
|
||
for item in value[-30:] if isinstance(value, list) else []:
|
||
summary = safe_request_summary(item, blocked=isinstance(item, dict) and item.get("blocked") is True)
|
||
if not summary:
|
||
continue
|
||
if type(item.get("sequence")) is int and 0 <= item["sequence"] <= 2 ** 53:
|
||
summary["sequence"] = item["sequence"]
|
||
timestamp = item.get("recorded_at")
|
||
if isinstance(timestamp, str) and re.fullmatch(r"\d{4}-\d\d-\d\dT[0-9:.+Z-]{8,25}", timestamp):
|
||
summary["recorded_at"] = timestamp
|
||
for key in ("automating", "blocked"):
|
||
if type(item.get(key)) is bool:
|
||
summary[key] = item[key]
|
||
rule, version = item.get("policy_rule"), item.get("policy_version")
|
||
allowed_rules = REQUEST_POLICY_RULES.get(version) if isinstance(version, str) else None
|
||
if allowed_rules and isinstance(rule, str) and rule in allowed_rules and item.get("automating") is True and item.get("blocked") is False:
|
||
summary.update(policy_version=version, policy_rule=rule)
|
||
records.append(summary)
|
||
return records
|
||
|
||
|
||
def safe_navigation_comparison(value) -> dict | None:
|
||
"""Only fixed relationship facts; never retain either navigation address."""
|
||
if not isinstance(value, dict):
|
||
return None
|
||
kinds = {"note", "profile", "notification", "explore", "search", "invalid"}
|
||
booleans = ("same_origin", "same_pathname", "same_object_id", "same_query", "same_query_pairs", "same_hash", "other_query_equal")
|
||
if any(not isinstance(value.get(key), str) or value[key] not in kinds for key in ("expected_kind", "actual_kind")):
|
||
return None
|
||
if any(type(value.get(key)) is not bool for key in booleans):
|
||
return None
|
||
result = {key: value[key] for key in ("expected_kind", "actual_kind", *booleans)}
|
||
for key in ("xsec_token", "xsec_source"):
|
||
item = value.get(key)
|
||
if (not isinstance(item, dict) or type(item.get("values_equal")) is not bool
|
||
or any(type(item.get(count)) is not int or not 0 <= item[count] <= 1000 for count in ("expected_count", "actual_count"))):
|
||
return None
|
||
result[key] = {field: item[field] for field in ("expected_count", "actual_count", "values_equal")}
|
||
return result
|
||
|
||
|
||
def safe_diagnostics(value, *, code=None, execution_profile=None) -> dict:
|
||
value = value if isinstance(value, dict) else {}
|
||
requests = [summary for item in value.get("blocked_requests", [])[:12] if (summary := safe_request_summary(item, blocked=True))] if isinstance(value.get("blocked_requests"), list) else []
|
||
count = value.get("blocked_request_count")
|
||
phase = value.get("phase")
|
||
result = {"phase": phase if isinstance(phase, str) and phase in FAILURE_PHASES else "unknown", "blocked_requests": requests,
|
||
"blocked_request_count": min(count, 1_000_000) if type(count) is int and count >= 0 else len(requests)}
|
||
if code == "NAVIGATION_BINDING_MISMATCH" and execution_profile == NAVIGATION_PROFILE:
|
||
if comparison := safe_navigation_comparison(value.get("navigation_comparison")):
|
||
result["navigation_comparison"] = comparison
|
||
return result
|
||
|
||
|
||
def safe_element_evidence(value: dict) -> dict:
|
||
fields = {"selector", "tag", "label", "kind", "unique", "count", "sensitive", "in_viewport", "search_input"}
|
||
result = {key: item for key, item in value.items() if key in fields and isinstance(item, (str, int, bool))}
|
||
stable_label = value.get("stable_label")
|
||
if isinstance(stable_label, str) and len(stable_label) <= 500:
|
||
result["stable_label"] = stable_label
|
||
basis = value.get("selector_basis")
|
||
if isinstance(basis, str) and basis in {"id", "semantic", "scroll_container", "id_family", "sibling_class"}:
|
||
result["selector_basis"] = basis
|
||
if type(value.get("collection")) is bool:
|
||
result["collection"] = value["collection"]
|
||
sources = value.get("source_selectors")
|
||
if isinstance(sources, list):
|
||
result["source_selectors"] = [item[:1000] for item in sources[:3] if isinstance(item, str) and item.strip()]
|
||
scroll = value.get("scroll_y")
|
||
if isinstance(scroll, dict):
|
||
cleaned = {key: item for key, item in scroll.items() if key in {"scroll_top", "scroll_height", "client_height"} and type(item) in {int, float} and 0 <= item <= 2 ** 53}
|
||
if type(scroll.get("can_scroll")) is bool:
|
||
cleaned["can_scroll"] = scroll["can_scroll"]
|
||
result["scroll_y"] = cleaned
|
||
hints = value.get("read_hints")
|
||
if isinstance(hints, list):
|
||
result["read_hints"] = []
|
||
for hint in hints[:12]:
|
||
if not isinstance(hint, dict):
|
||
continue
|
||
cleaned = {key: item[:limit] for key, limit in (("selector", 1000), ("tag", 32), ("label", 100)) if isinstance(item := hint.get(key), str)}
|
||
for key in ("unique", "in_viewport", "link"):
|
||
if type(hint.get(key)) is bool:
|
||
cleaned[key] = hint[key]
|
||
if type(hint.get("count")) is int and 0 <= hint["count"] <= 2 ** 53:
|
||
cleaned["count"] = hint["count"]
|
||
source = hint.get("source")
|
||
if isinstance(source, str) and source in {"text", "href", "attribute", "media"}:
|
||
cleaned["source"] = source
|
||
if source == "attribute" and hint.get("attribute") == "datetime":
|
||
cleaned["attribute"] = "datetime"
|
||
if source == "media" and isinstance(hint.get("media_kind"), str) and hint["media_kind"] in {"image", "video"} and type(hint.get("loaded")) is bool:
|
||
cleaned.update(media_kind=hint["media_kind"], loaded=hint["loaded"])
|
||
elif source == "media":
|
||
cleaned.pop("source", None)
|
||
if cleaned.get("selector"):
|
||
result["read_hints"].append(cleaned)
|
||
return result
|
||
|
||
|
||
def safe_failure_capture(value) -> dict | None:
|
||
if not isinstance(value, dict) or not isinstance(value.get("url"), str) or not isinstance(value.get("elements"), list) or not isinstance(value.get("screenshot_path"), str):
|
||
return None
|
||
# These are already redacted page captures, not diagnostics. Preserve their
|
||
# actual before/after state, while excluding arbitrary exception attributes.
|
||
allowed = {"capture_id", "url", "title", "text", "viewport_text", "text_scope", "viewport", "elements", "screenshot_path", "captured_at", "identity_verification", "request_observations", "popup_observations"}
|
||
capture = {key: deepcopy(item) for key, item in value.items() if key in allowed}
|
||
if any(key in value for key in ("execution_profile", "execution_contract")):
|
||
if value.get("execution_profile") is None or not execution_metadata_matches(value, value.get("execution_profile")):
|
||
return None
|
||
capture.update(execution_profile=value["execution_profile"], execution_contract=deepcopy(value["execution_contract"]))
|
||
for key, limit in (("url", 4000), ("title", 250), ("text", 30000), ("viewport_text", 30000), ("capture_id", 128), ("text_scope", 64), ("captured_at", 64), ("identity_verification", 64), ("screenshot_path", 4000)):
|
||
if key in capture:
|
||
if not isinstance(capture[key], str):
|
||
capture.pop(key)
|
||
else:
|
||
capture[key] = capture[key][:limit]
|
||
capture["elements"] = [safe_element_evidence(element) for element in value["elements"][:300] if isinstance(element, dict)]
|
||
capture["request_observations"] = safe_request_observations(capture.get("request_observations"))
|
||
capture["popup_observations"] = safe_popup_observations(capture.get("popup_observations"))
|
||
viewport = capture.get("viewport")
|
||
if isinstance(viewport, dict):
|
||
capture["viewport"] = {key: item for key, item in viewport.items() if key in {"width", "height", "scroll_x", "scroll_y"} and type(item) in {int, float}}
|
||
else:
|
||
capture.pop("viewport", None)
|
||
return capture
|
||
|
||
|
||
def safe_error_codes(value) -> list[str]:
|
||
return list(dict.fromkeys(item for item in value[:10] if isinstance(item, str) and re.fullmatch(r"[A-Za-z][A-Za-z0-9_]{0,63}", item))) if isinstance(value, list) else []
|
||
|
||
|
||
def safe_failure_diagnostics(value, steps: list[dict], error_codes, *, passed=None, execution_profile=None) -> dict | None:
|
||
"""Keep a failed worker's bounded cause, never a new outcome or raw payload."""
|
||
if passed is not False or not isinstance(value, dict) or set(value) != {"step_position", "kind", "code", "diagnostics"}:
|
||
return None
|
||
position, kind, code = (value[key] for key in ("step_position", "kind", "code"))
|
||
if (type(position) is not int or not 1 <= position <= len(steps) or not isinstance(kind, str)
|
||
or kind != steps[position - 1].get("kind") or not isinstance(code, str) or code not in safe_error_codes(error_codes)
|
||
or not isinstance(value["diagnostics"], dict)):
|
||
return None
|
||
return {"step_position": position, "kind": kind, "code": code,
|
||
"diagnostics": safe_diagnostics(value["diagnostics"], code=code, execution_profile=execution_profile)}
|
||
|
||
|
||
def safe_verify_timings(step_values, total_value, steps: list[dict], *, passed: bool = False) -> dict | None:
|
||
def milliseconds(value):
|
||
return type(value) is int and 0 <= value <= 120000
|
||
if not isinstance(total_value, dict) or set(total_value) != {"duration_ms", "entry_ms", "close_ms"}:
|
||
return None
|
||
if not milliseconds(total_value["duration_ms"]) or not milliseconds(total_value["entry_ms"]) or total_value["entry_ms"] > total_value["duration_ms"] + 1:
|
||
return None
|
||
if total_value["close_ms"] is not None and (not milliseconds(total_value["close_ms"]) or total_value["close_ms"] > total_value["duration_ms"] + 1):
|
||
return None
|
||
if not isinstance(step_values, list) or len(step_values) > min(30, len(steps)):
|
||
return None
|
||
if passed and (len(step_values) != len(steps) or any(not isinstance(item, dict) or item.get("status") != "passed" for item in step_values)):
|
||
return None
|
||
terminal = False
|
||
for position, item in enumerate(step_values, 1):
|
||
if terminal or not isinstance(item, dict) or set(item) != {"step_position", "kind", "status", "duration_ms", "phase_ms"}:
|
||
return None
|
||
if type(item["step_position"]) is not int or item["step_position"] != position or item["kind"] != steps[position - 1].get("kind") or not isinstance(item["status"], str) or item["status"] not in {"passed", "failed", "interrupted"}:
|
||
return None
|
||
terminal = item["status"] != "passed"
|
||
if not milliseconds(item["duration_ms"]) or item["duration_ms"] > total_value["duration_ms"] + 1 or not isinstance(item["phase_ms"], dict) or not set(item["phase_ms"]) <= TIMING_PHASES:
|
||
return None
|
||
if any(not milliseconds(value) for value in item["phase_ms"].values()) or sum(item["phase_ms"].values()) > item["duration_ms"] + len(TIMING_PHASES):
|
||
return None
|
||
# Entry, step execution, and explicit closing are exclusive measurements;
|
||
# allow only a small aggregate tolerance for independently rounded values.
|
||
recorded_ms = total_value["entry_ms"] + sum(item["duration_ms"] for item in step_values) + (total_value["close_ms"] or 0)
|
||
if recorded_ms > total_value["duration_ms"] + 30:
|
||
return None
|
||
return {"step_timings": deepcopy(step_values), "verify_timing": deepcopy(total_value)}
|
||
|
||
|
||
def safe_batch_evidence(value) -> list[dict]:
|
||
records = []
|
||
for item in value[:5] if isinstance(value, list) else []:
|
||
if not isinstance(item, dict) or type(item.get("batch_index")) is not int or not 1 <= item["batch_index"] <= 5:
|
||
continue
|
||
if any(type(item.get(key)) is not int or not 0 <= item[key] <= 500 for key in ("added_count", "duplicate_count")):
|
||
continue
|
||
capture = safe_failure_capture(item.get("capture"))
|
||
if capture:
|
||
records.append({"batch_index": item["batch_index"], "added_count": item["added_count"], "duplicate_count": item["duplicate_count"], "capture": capture})
|
||
return records
|
||
|
||
|
||
class WorkerError(Exception):
|
||
def __init__(self, message: str, code: str = "worker_error", *, before=None, after=None, diagnostics=None, batch_evidence=None, error_codes=None, session_closed=False):
|
||
super().__init__(message)
|
||
self.code = safe_worker_code(code)
|
||
self.before = safe_failure_capture(before)
|
||
self.after = safe_failure_capture(after)
|
||
captures = [capture for capture in (self.before, self.after) if capture is not None]
|
||
profile = NAVIGATION_PROFILE if captures and all(execution_metadata_matches(capture, NAVIGATION_PROFILE) for capture in captures) else None
|
||
self.diagnostics = safe_diagnostics(diagnostics, code=self.code, execution_profile=profile)
|
||
self.batch_evidence = safe_batch_evidence(batch_evidence)
|
||
self.error_codes = safe_error_codes(error_codes)
|
||
self.session_closed = session_closed is True
|
||
|
||
|
||
class BrowserWorker:
|
||
"""Serialized RPC to the single browser owner; never imports the old application."""
|
||
|
||
def __init__(self, root: Path, home: Path):
|
||
self.root, self.home = root, home
|
||
self.process: asyncio.subprocess.Process | None = None
|
||
self.lock = asyncio.Lock()
|
||
self.stderr_task: asyncio.Task | None = None
|
||
self.closing = False
|
||
self.close_task: asyncio.Task | None = None
|
||
|
||
async def _start(self) -> None:
|
||
if self.process and self.process.returncode is None:
|
||
return
|
||
node = os.environ.get("ASSISTANT_NODE") or shutil.which("node")
|
||
if not node:
|
||
raise WorkerError("找不到 Node.js，请通过项目启动脚本运行后台", "node_not_found")
|
||
script = self.root / "tools" / "learning" / "browser-worker.mjs"
|
||
if not script.is_file():
|
||
raise WorkerError("本机学习执行器尚未就绪", "worker_unavailable")
|
||
environment = {**os.environ, "ASSISTANT_LEARNING_HOME": str(self.home)}
|
||
self.process = await asyncio.create_subprocess_exec(node, str(script), cwd=self.root, env=environment, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, limit=8 * 1024 * 1024)
|
||
if self.closing:
|
||
await self._stop()
|
||
raise WorkerError("学习执行器正在正常退出", "worker_closed")
|
||
process = self.process
|
||
|
||
async def consume_stderr() -> None:
|
||
assert process.stderr
|
||
while await process.stderr.readline():
|
||
pass # Do not forward browser diagnostic text or URLs into business logs.
|
||
|
||
self.stderr_task = asyncio.create_task(consume_stderr())
|
||
|
||
async def call(self, method: str, params: dict) -> dict:
|
||
async with self.lock:
|
||
if self.closing:
|
||
raise WorkerError("学习执行器正在正常退出", "worker_closed")
|
||
await self._start()
|
||
process = self.process
|
||
assert process and process.stdin and process.stdout
|
||
request_id = uuid.uuid4().hex
|
||
try:
|
||
process.stdin.write((json.dumps({"id": request_id, "method": method, "params": params}, ensure_ascii=False) + "\n").encode())
|
||
await process.stdin.drain()
|
||
async with asyncio.timeout(120):
|
||
while True:
|
||
line = await process.stdout.readline()
|
||
if not line:
|
||
raise WorkerError("本机学习浏览器已退出，请重新打开会话", "worker_closed")
|
||
try:
|
||
response = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
if response.get("id") != request_id:
|
||
continue
|
||
if response.get("error"):
|
||
error = response["error"]
|
||
raise WorkerError(str(error.get("message", "学习执行失败")), error.get("code"), before=error.get("before"), after=error.get("after"), diagnostics=error.get("diagnostics"), batch_evidence=error.get("batch_evidence"), error_codes=error.get("error_codes"), session_closed=error.get("session_closed"))
|
||
return response.get("result") or {}
|
||
except (TimeoutError, BrokenPipeError, ConnectionResetError):
|
||
await self._stop()
|
||
raise WorkerError("学习执行器响应中断；请核对页面与步骤后重新打开，不会自动重放动作", "worker_timeout") from None
|
||
|
||
async def _stop(self) -> None:
|
||
process, self.process = self.process, None
|
||
if not process:
|
||
return
|
||
if process.stdin:
|
||
process.stdin.close()
|
||
try:
|
||
await asyncio.wait_for(process.wait(), 15)
|
||
except TimeoutError:
|
||
process.terminate()
|
||
try:
|
||
await asyncio.wait_for(process.wait(), 3)
|
||
except TimeoutError:
|
||
process.kill()
|
||
await process.wait()
|
||
if self.stderr_task:
|
||
self.stderr_task.cancel()
|
||
self.stderr_task = None
|
||
|
||
async def close(self) -> None:
|
||
# EOF must reach the child even while an RPC holds the request lock.
|
||
# Node closes its contexts on EOF, unblocking that exact active action.
|
||
self.closing = True
|
||
if self.close_task is None:
|
||
self.close_task = asyncio.create_task(self._stop())
|
||
await asyncio.shield(self.close_task)
|