"""OpenCode 图片语境生成运行器。 模块只负责准备固定运行目录、调用公开 CLI 并把不可信输出转为候选 JSON；canonical sidecar 的读取、指纹复核和写入始终由上层元数据服务完成。 """ from __future__ import annotations from loguru import logger import hashlib import io import json import os import re import shutil import signal import socket import stat import subprocess import tempfile import time import uuid from dataclasses import replace from io import BytesIO from pathlib import Path, PurePosixPath from queue import Empty, Queue from threading import Event, Lock, RLock, Semaphore, local from typing import Any, BinaryIO, Callable, Iterator, Mapping from urllib.error import URLError from urllib.request import urlopen from backend.agent_executor import AgentExecutorClient, AgentExecutorError, ExecutorTaskResponse from backend.agent_resume import classify_resume_error, normalize_identifier from backend.config import Settings, validate_agent_concurrency from backend.metadata import MemeContext from backend.model_capability_provider import ( ModelCapabilityProvider, ModelCapabilityProviderError, ModelCapabilityRequest, capability_for_model_provider, normalize_observed_cost, ) from backend.opencode_result_store import OpenCodeResultStore from backend.runtime_execution import AttemptFence, ExecutionBinding, ExecutionBindingError, stable_input_digest from backend.opencode_workspace import ( LocalWorkspaceProvider, ResolvedWorkspace, TrustedWorkspaceContext, WorkspaceCapabilityError, WorkspaceProvider, WorkspaceResolutionError, build_edit_permission_rules, build_external_directory_rules, capability_for_provider, prepare_candidates_for_provider, validate_directory_path, validate_file_path, ) from backend.public_dto import PublicDataError, secret_inventory_from_settings, validate_agent_result from backend.visual_snapshot import ( VisualMatchSnapshotError, validate_visual_match_snapshot_manifest, visual_match_snapshot_manifest, ) try: from jsonschema import Draft202012Validator, FormatChecker except ImportError: # pragma: no cover - 兼容尚未同步依赖的开发环境 Draft202012Validator = None FormatChecker = None # 此配置只定义兼容 OpenAI API 的通用 provider，不保存部署地址、长期密钥或模型选择。 # broker 地址和当前任务的短期 capability 由 executor 在子进程环境中装配。 RUNTIME_OPENCODE_CONFIG: dict[str, Any] = { "$schema": "https://opencode.ai/config.json", "experimental": { "continue_loop_on_deny": True, }, "provider": { "mememeow": { "npm": "@ai-sdk/openai", "name": "MemeMeow OpenAI Responses provider", "options": { "baseURL": "{env:MEMEMEOW_MODEL_BROKER_URL}", "apiKey": "{env:MEMEMEOW_MODEL_CAPABILITY}", }, "models": { "gpt-5.6-luna": { "id": "gpt-5.6-luna", "name": "GPT-5.6 Luna", "family": "gpt-luna", "release_date": "2026-07-09", "status": "active", "reasoning": True, "tool_call": True, "temperature": False, "attachment": True, "interleaved": False, "modalities": { "input": ["text", "image", "pdf"], "output": ["text"], }, "limit": { "context": 1050000, "input": 922000, "output": 128000, }, "cost": { "input": 0.25, "output": 2, "cache_read": 0.02, "cache_write": 0.25, "context_over_200k": { "input": 0.4, "output": 1.8, "cache_read": 0.04, "cache_write": 0.5, }, }, "variants": { "max": {"reasoningEffort": "max"}, }, } }, } }, } # 当前研究任务统一使用 Luna 的 max 变体，避免不同任务因默认推理强度产生质量漂移。 OPENCODE_REASONING_VARIANT = "max" # 诊断日志只保留固定前缀，避免完整 Agent transcript 长期占用 runtime 磁盘。 # 该值不参与 OpenCode 输出读取或业务结果解析，因此不会拒绝大输出。 DIAGNOSTIC_LOG_BYTES = 256 * 1024 STREAM_COPY_CHUNK_BYTES = 64 * 1024 RESULT_FILE_NAME = "result.json.tmp" RESULT_DRAFT_NAME = "result.json.draft" RESULT_DEFAULT_MAX_BYTES = 1024 * 1024 CANDIDATE_MANIFEST_MAX_BYTES = 4 * 1024 * 1024 MAX_CANDIDATE_FILE_BYTES = 64 * 1024 * 1024 MAX_CANDIDATE_ROOT_BYTES = 256 * 1024 * 1024 EXECUTOR_IMAGE_ROOT = Path("/images") _SECRET_PATTERNS = ( # 分开处理 JSON 字符串值和普通 header/日志格式，保留原有引号结构。 re.compile(r"(?i)(authorization\s*[\"']?\s*:\s*[\"']?bearer\s+)[^\s,;}\]\"']+"), re.compile(r"(?i)([\"']?(?:api[_-]?key|access[_-]?token|refresh[_-]?token|token|secret|password)[\"']?\s*:\s*)([\"'])(.*?)\2"), re.compile(r"(?i)([\"']?(?:api[_-]?key|access[_-]?token|refresh[_-]?token|token|secret|password)[\"']?\s*[:=]\s*)[^\s,;}\]\"']+"), ) _PATH_PATTERN = re.compile(r"(?:/runtime|/images|/skills|/app|[A-Za-z]:\\)[^\s,;]+") _GENERIC_PATH_PATTERN = re.compile(r"(? str: """从 host OpenCode 诊断中移除凭据、路径和多行 transcript。""" value = value.splitlines()[0] if value.splitlines() else value for secret in secrets: if secret: value = value.replace(secret, "[REDACTED]") for pattern in _SECRET_PATTERNS: if pattern.groups == 3: value = pattern.sub(r"\1\2[REDACTED]\2", value) else: value = pattern.sub(r"\1[REDACTED]", value) value = _PATH_PATTERN.sub("[PATH]", value) return _GENERIC_PATH_PATTERN.sub("[PATH]", value)[:500] def _stream_sample(stream: BinaryIO, limit: int) -> bytes: """读取临时输出的头尾样本，避免大 JSONL 将 provider 错误留在不可见尾部。""" stream.seek(0) head = stream.read(limit) stream.seek(0, os.SEEK_END) size = stream.tell() if size <= limit: return head stream.seek(max(0, size - limit)) return head + b"\n" + stream.read(limit) class OpenCodeError(RuntimeError): """携带稳定错误码的 OpenCode 运行失败。""" def __init__( self, code: str, message: str | None = None, *, session_id: str | None = None, executor_attempt_id: str | None = None, retryable: bool | None = None, http_status: int | None = None, reason_code: str | None = None, process_reaped: bool | None = None, observed_cost: str | None = None, usage_checked_at: str | None = None, reminder_sent: bool = False, termination_reason: str | None = None, termination_signal: str | None = None, check_stage: str | None = None, trigger_reason: str | None = None, ): """保存稳定错误及受信 Executor 返回的有限 attempt 诊断。""" super().__init__(message or code) self.code = code self.session_id = normalize_identifier(session_id, kind="session") self.executor_attempt_id = normalize_identifier(executor_attempt_id, kind="attempt") self.retryable = retryable self.http_status = http_status self.reason_code = reason_code if isinstance(reason_code, str) and re.fullmatch(r"[a-z0-9][a-z0-9_.-]{0,127}", reason_code) else None self.process_reaped = process_reaped if isinstance(process_reaped, bool) else None self.observed_cost = observed_cost if isinstance(observed_cost, str) else None self.usage_checked_at = usage_checked_at if isinstance(usage_checked_at, str) else None self.reminder_sent = bool(reminder_sent) self.termination_reason = termination_reason self.termination_signal = termination_signal self.check_stage = check_stage self.trigger_reason = trigger_reason def _workspace_opencode_config(workspace: ResolvedWorkspace) -> dict[str, Any]: """生成当前 workspace 的无密钥 OpenCode 配置和文件工具权限规则。""" config = json.loads(json.dumps(RUNTIME_OPENCODE_CONFIG, ensure_ascii=False)) config["permission"] = { "external_directory": {path: decision for path, decision in workspace.permission_rules}, "edit": { path: decision for path, decision in build_edit_permission_rules( task_scratch_root=workspace.task_scratch_root, config_file=workspace.config_file, config_dir=workspace.config_dir, draft_path=workspace.draft_path, result_path=workspace.result_path, candidate_root=workspace.candidate_root, ) }, } return config class OpenCodeRunner: """在固定 runtime 中按 slot 受控并行执行 OpenCode，并返回已校验研究结果。""" def __init__( self, settings: Settings, project_root: Path | None = None, *, workspace_provider: WorkspaceProvider | None = None, model_capability_provider: ModelCapabilityProvider | None = None, ): """初始化 runner；应用装配应显式传入 provider，旧直接夹具兼容 local。""" self.settings = settings self.project_root = (project_root or Path(__file__).resolve().parent.parent).resolve() self.runtime_root = (settings.opencode_runtime_root or settings.data_root / "opencode").expanduser().resolve() self.workspace = self.runtime_root / "workspace" self.db_path = self.runtime_root / "opencode.db" self.slots_root = self.runtime_root / "slots" self.lock_path = self.runtime_root / "worker.lock" self.log_root = self.runtime_root / "logs" self.concurrency = validate_agent_concurrency(settings.opencode_concurrency) self._slot_semaphore = Semaphore(self.concurrency) self._slot_ids: Queue[int] = Queue() for slot_id in range(self.concurrency): self._slot_ids.put(slot_id) self._prepare_lock = Lock() self._runtime_ready = False self._closing = Event() self._process: subprocess.Popen[bytes] | None = None self._processes: set[subprocess.Popen[bytes]] = set() # 结果目录创建、活动 task 标记和清理必须共享可重入锁，避免清理与启动交错。 self._process_lock = RLock() self._process_tasks: dict[subprocess.Popen[bytes], str | None] = {} self._active_task_ids: set[str] = set() self._active_executor_task_ids: set[str] = set() self._last_executor_attempt_ids: dict[str, str] = {} self._executor_attempt_context = local() self._analysis_observations: dict[tuple[str, str], dict[str, object]] = {} self._cancelled_task_ids: set[str] = set() # 未显式传入 provider 只为既有开源直接构造调用保留 local 兼容；应用工厂 # 对 non-local scope 必须显式装配 provider，不能在这里猜测路径。 self.workspace_provider: WorkspaceProvider = workspace_provider or LocalWorkspaceProvider( self.runtime_root, image_root=self._configured_image_root(), skill_root=self.project_root / "skills" / "research-meme-context", ) self.model_capability_provider = model_capability_provider self._workspace_by_task: dict[str, ResolvedWorkspace] = {} self._workspace_selectors: dict[str, str] = {} self._workspace_capabilities: dict[str, str] = {} # 进程内 fencing 只做旧回调快速拒绝；跨进程 claim 仍由任务数据库负责。 self._attempt_fence = AttemptFence() self.result_store = OpenCodeResultStore( self.runtime_root / "task-results", result_name=RESULT_FILE_NAME, draft_name=RESULT_DRAFT_NAME, max_bytes=int(getattr(settings, "agent_result_max_bytes", RESULT_DEFAULT_MAX_BYTES)), ) self.executor = AgentExecutorClient( getattr(settings, "agent_executor_url", None), getattr(settings, "agent_executor_token", None), timeout=int(getattr(settings, "agent_executor_request_timeout_seconds", 1810)), ) def resolve_workspace(self, context: TrustedWorkspaceContext) -> ResolvedWorkspace: """从可信任务上下文解析并缓存一次 workspace 描述。""" try: resolved = self.workspace_provider.resolve(context) except (WorkspaceResolutionError, WorkspaceCapabilityError) as exc: raise OpenCodeError(exc.code, str(exc)) from exc except Exception as exc: # noqa: BLE001 - provider 是安全边界，异常统一失败关闭 raise OpenCodeError("opencode_workspace_invalid", "workspace provider 无法解析任务") from exc if not isinstance(resolved, ResolvedWorkspace): raise OpenCodeError("opencode_workspace_invalid", "workspace provider 返回值无效") if context.selector is not None and context.selector != resolved.selector: raise OpenCodeError("opencode_workspace_mismatch", "workspace selector 与可信任务事实不一致") try: validated = self._validate_resolved_workspace(context, resolved) # 候选物化属于 provider 的受信适配层，必须在任何 OpenCode 进程或 # 结果文件副作用前完成；物化后再次校验固定 task 根和 manifest 节点。 prepare_candidates_for_provider(self.workspace_provider, context, validated) materializer = callable(getattr(self.workspace_provider, "prepare_candidates", None)) if validated.candidate_root.exists() or validated.candidate_root.is_symlink(): validate_directory_path(validated.candidate_root, code="opencode_workspace_invalid") else: validate_directory_path(validated.candidate_root, allow_missing_leaf=True, code="opencode_workspace_invalid") if context.visual_match_snapshot is not None: if not materializer and not validated.local: raise WorkspaceResolutionError("visual_candidate_materialization_failed", "外部 workspace 缺少候选物化能力") manifest = self._validate_candidate_manifest(validated, context.visual_match_snapshot) self._validate_candidate_tree(validated, manifest) return validated except WorkspaceResolutionError as exc: raise OpenCodeError(exc.code, str(exc)) from exc except (OSError, TypeError, ValueError) as exc: raise OpenCodeError("opencode_workspace_invalid", "workspace provider 返回值无法校验") from exc @staticmethod def _validate_candidate_manifest(workspace: ResolvedWorkspace, snapshot: Mapping[str, object]) -> dict[str, object]: """在启动外部 Agent 前校验 manifest 文件内容与 Task snapshot 完全一致。""" path = workspace.candidate_manifest_path try: validate_file_path(path, code="visual_candidate_materialization_failed") info = path.lstat() if info.st_size > CANDIDATE_MANIFEST_MAX_BYTES: raise WorkspaceResolutionError("visual_candidate_materialization_failed", "候选 manifest 超过大小限制") descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) try: raw = os.read(descriptor, CANDIDATE_MANIFEST_MAX_BYTES + 1) finally: os.close(descriptor) if len(raw) > CANDIDATE_MANIFEST_MAX_BYTES: raise WorkspaceResolutionError("visual_candidate_materialization_failed", "候选 manifest 超过大小限制") document = json.loads(raw.decode("utf-8")) expected = visual_match_snapshot_manifest(snapshot) validated = validate_visual_match_snapshot_manifest( document, expected_sha256=str(expected["snapshot_sha256"]), ) except WorkspaceResolutionError: raise except (OSError, UnicodeDecodeError, json.JSONDecodeError, RecursionError, TypeError, ValueError, VisualMatchSnapshotError) as exc: raise WorkspaceResolutionError("visual_candidate_materialization_failed", "候选 manifest 无法校验") from exc if validated != expected: raise WorkspaceResolutionError("visual_candidate_materialization_failed", "候选 manifest 与 snapshot 不一致") return validated @staticmethod def _validate_candidate_tree(workspace: ResolvedWorkspace, manifest: Mapping[str, object]) -> None: """按 manifest 校验候选目录，拒绝额外节点、符号链接和可写文件。""" root = workspace.candidate_root try: validate_directory_path(root, code="visual_candidate_materialization_failed") root_info = root.lstat() if root_info.st_mode & (stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH): raise ValueError("visual_candidate_materialization_failed") candidates = manifest.get("candidates") if not isinstance(candidates, list): raise ValueError("visual_candidate_materialization_failed") expected_files = {"manifest.json"} expected_directories: set[str] = set() declared_bytes = 0 for candidate in candidates: if not isinstance(candidate, Mapping): raise ValueError("visual_candidate_materialization_failed") relative = candidate.get("relative_path") if not isinstance(relative, str): raise ValueError("visual_candidate_materialization_failed") size_bytes = candidate.get("size_bytes") if not isinstance(size_bytes, int) or isinstance(size_bytes, bool) or size_bytes < 0 or size_bytes > MAX_CANDIDATE_FILE_BYTES: raise ValueError("visual_candidate_materialization_failed") declared_bytes += size_bytes if declared_bytes > MAX_CANDIDATE_ROOT_BYTES: raise ValueError("visual_candidate_materialization_failed") expected_files.add(relative) parts = PurePosixPath(relative).parts expected_directories.update("/".join(parts[:index]) for index in range(1, len(parts))) for item in root.rglob("*"): info = item.lstat() relative = item.relative_to(root).as_posix() if stat.S_ISLNK(info.st_mode) or info.st_mode & (stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH): raise ValueError("visual_candidate_materialization_failed") if stat.S_ISDIR(info.st_mode): if relative not in expected_directories: raise ValueError("visual_candidate_materialization_failed") elif stat.S_ISREG(info.st_mode): if relative not in expected_files: raise ValueError("visual_candidate_materialization_failed") else: raise ValueError("visual_candidate_materialization_failed") total_bytes = sum(item.lstat().st_size for item in root.rglob("*") if item.is_file()) if total_bytes > MAX_CANDIDATE_ROOT_BYTES: raise ValueError("visual_candidate_materialization_failed") for candidate in candidates: relative = candidate.get("relative_path") image_sha256 = candidate.get("image_sha256") size_bytes = candidate.get("size_bytes") if ( not isinstance(relative, str) or not isinstance(image_sha256, str) or not isinstance(size_bytes, int) or isinstance(size_bytes, bool) or size_bytes < 0 or size_bytes > MAX_CANDIDATE_FILE_BYTES ): raise ValueError("visual_candidate_materialization_failed") candidate_path = root.joinpath(*PurePosixPath(relative).parts) validate_file_path(candidate_path, code="visual_candidate_materialization_failed") info = candidate_path.lstat() if info.st_size != size_bytes: raise ValueError("visual_candidate_materialization_failed") digest = hashlib.sha256() descriptor = os.open(candidate_path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) try: while True: chunk = os.read(descriptor, 1024 * 1024) if not chunk: break digest.update(chunk) finally: os.close(descriptor) if digest.hexdigest() != image_sha256.lower(): raise ValueError("visual_candidate_materialization_failed") except WorkspaceResolutionError: raise except (OSError, TypeError, ValueError) as exc: raise WorkspaceResolutionError("visual_candidate_materialization_failed", "候选文件无法校验") from exc def prepare_candidates_for_task( self, *, task_id: str, attempt_id: str, scope_id: str, snapshot: Mapping[str, object], selector: str | None = None, session_id: str | None = None, resume_of_attempt_id: str | None = None, image_relative_path: str | None = None, ) -> ResolvedWorkspace: """在外部执行前按可信 claim 物化当前 Task 的候选 manifest。""" try: context = TrustedWorkspaceContext( task_id=task_id, attempt_id=attempt_id, scope_id=scope_id, selector=selector, session_id=session_id, resume_of_attempt_id=resume_of_attempt_id, image_relative_path=image_relative_path, visual_match_snapshot=snapshot, ) except WorkspaceResolutionError as exc: raise OpenCodeError(exc.code, str(exc)) from exc return self.resolve_workspace(context) def _validate_resolved_workspace(self, context: TrustedWorkspaceContext, resolved: ResolvedWorkspace) -> ResolvedWorkspace: """固定 provider 输出的结果/DB 布局，并重建服务端权限规则。 provider 可以把图片、metadata 和 Skill 放在容器外部受控挂载，但配置、Task 临时目录和结果路径不能变成新的任意文件写入接口。这里的检查只验证已存在 的输入视图；Task 目录和结果目录在 capability 校验后才创建。 """ def absolute(path: Path) -> Path: """规范化路径而不跟随最终符号链接。""" return Path(os.path.abspath(Path(path).expanduser())) directory = absolute(resolved.directory) scratch = absolute(resolved.task_scratch_root) config_file = absolute(resolved.config_file) config_dir = absolute(resolved.config_dir) images_root = absolute(resolved.images_root) metadata_root = absolute(resolved.metadata_root) skill_root = absolute(resolved.skill_root) candidate_root = absolute(resolved.candidate_root) task_results = absolute(resolved.task_results_root) expected_results = absolute(self.runtime_root / "task-results" / context.task_id) expected_db = absolute(self.db_path) if not isinstance(resolved.selector, str) or resolved.selector != resolved.selector.strip() or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", resolved.selector): raise WorkspaceResolutionError("opencode_workspace_invalid", "workspace selector 无效") if absolute(resolved.db_path) != expected_db: raise WorkspaceResolutionError("opencode_workspace_invalid", "workspace DB 路径不受支持") validate_file_path(resolved.db_path, allow_missing=True, code="opencode_workspace_invalid") if task_results != expected_results: raise WorkspaceResolutionError("opencode_workspace_invalid", "workspace 结果路径不受支持") if scratch != directory / "tasks" / context.task_id: raise WorkspaceResolutionError("opencode_workspace_invalid", "workspace 临时路径不受支持") if candidate_root != directory.parent / "candidates" / context.task_id: raise WorkspaceResolutionError("opencode_workspace_invalid", "workspace 候选路径不受支持") if config_file != scratch / "opencode.json" or config_dir != scratch / ".opencode": raise WorkspaceResolutionError("opencode_workspace_invalid", "workspace 配置路径不受支持") if resolved.local and context.scope_id != "local": raise WorkspaceResolutionError("opencode_workspace_mismatch", "non-local 任务不能使用 local workspace") if resolved.local and resolved.selector != "local": raise WorkspaceResolutionError("opencode_workspace_invalid", "local workspace selector 无效") if not resolved.local and resolved.selector == "local": raise WorkspaceResolutionError("opencode_workspace_invalid", "external workspace selector 保留字无效") for path in (directory, images_root, metadata_root, candidate_root.parent): validate_directory_path(path, code="opencode_workspace_invalid") if candidate_root.exists() or candidate_root.is_symlink(): validate_directory_path(candidate_root, code="opencode_workspace_invalid") if resolved.local and skill_root.is_symlink(): # local provider 为既有 checkout 保留相对 Skill 链接；链接目标仍须是普通目录。 validate_directory_path(skill_root.resolve(strict=True), code="opencode_workspace_invalid") else: validate_directory_path(skill_root, code="opencode_workspace_invalid") for path in (scratch, config_dir, task_results): if path.exists() or path.is_symlink(): validate_directory_path(path, code="opencode_workspace_invalid") validated = replace( resolved, directory=directory, config_file=config_file, config_dir=config_dir, images_root=images_root, metadata_root=metadata_root, skill_root=skill_root, candidate_root=candidate_root, task_scratch_root=scratch, task_results_root=task_results, db_path=expected_db, ) return replace(validated, _permission_rules=build_external_directory_rules(validated)) def workspace_for_task(self, task_id: str) -> str | None: """返回任务最近解析的 opaque selector，供诊断写回。""" with self._process_lock: return self._workspace_selectors.get(task_id) @property def executor_mode(self) -> bool: """判断当前任务是否通过已选定的 executor 执行。""" mode = getattr(self.settings, "agent_runtime_mode", "auto") if mode not in {"auto", "executor", "host"}: raise OpenCodeError("agent_runtime_mode_invalid", "Agent 运行模式不受支持") if mode == "executor": return True return mode == "auto" and self.executor.configured @property def last_executor_attempt_id(self) -> str | None: """返回最近一次当前线程任务的 executor attempt 摘要。""" # 该属性只作为成功写回的诊断补充；真正的绑定事实仍来自 executor 响应。 with self._process_lock: if not self._last_executor_attempt_ids: return None return next(reversed(self._last_executor_attempt_ids.values())) def executor_attempt_id_for(self, task_id: str) -> str | None: """按业务 task 读取最近 attempt，避免并发任务互相串接诊断。""" thread_binding = getattr(self._executor_attempt_context, "binding", None) if isinstance(thread_binding, tuple) and thread_binding[0] == task_id: return thread_binding[1] with self._process_lock: return self._last_executor_attempt_ids.get(task_id) def analysis_observation_for(self, task_id: str, executor_attempt_id: str | None = None) -> dict[str, object] | None: """按当前 executor attempt 取出一次分析摘要，读取后立即清理。""" with self._process_lock: expected_attempt_id = executor_attempt_id or self._last_executor_attempt_ids.get(task_id) if not isinstance(expected_attempt_id, str) or not expected_attempt_id: return None value = self._analysis_observations.pop((task_id, expected_attempt_id), None) return dict(value) if value is not None else None def _remember_executor_attempt(self, task_id: str, executor_attempt_id: str) -> None: """保存任务 attempt 摘要并限制长期运行进程的内存历史。""" self._executor_attempt_context.binding = (task_id, executor_attempt_id) with self._process_lock: self._last_executor_attempt_ids[task_id] = executor_attempt_id while len(self._last_executor_attempt_ids) > 5000: self._last_executor_attempt_ids.pop(next(iter(self._last_executor_attempt_ids))) def _remember_analysis_observation(self, task_id: str, response: ExecutorTaskResponse) -> None: """暂存成功响应的分析摘要，等待 API 在同一次调用中取出并持久化。""" if response.observed_cost is None: return with self._process_lock: thread_binding = getattr(self._executor_attempt_context, "binding", None) expected_attempt_id = thread_binding[1] if isinstance(thread_binding, tuple) and thread_binding[0] == task_id else None response_attempt_id = response.executor_attempt_id or expected_attempt_id if not isinstance(expected_attempt_id, str) or response_attempt_id != expected_attempt_id: return self._analysis_observations[(task_id, expected_attempt_id)] = { "observed_cost": response.observed_cost, "usage_checked_at": response.usage_checked_at, "reminder_sent": response.reminder_sent, "process_reaped": response.process_reaped, } while len(self._analysis_observations) > 5000: self._analysis_observations.pop(next(iter(self._analysis_observations))) @staticmethod def _executor_error_code(code: str, *, health: bool = False) -> str: """把 executor 客户端异常收口为任务服务可识别的稳定错误码。""" if health and code in {"agent_timeout", "agent_executor_unavailable", "agent_executor_http_error"}: return "agent_runtime_unavailable" known = { "agent_analysis_policy_missing", "agent_analysis_policy_invalid", "agent_analysis_usage_unavailable", "agent_analysis_reminder_plugin_unavailable", "agent_maximum_analysis_depth_exceeded", "model_capability_invalid", "model_capability_unavailable", "model_broker_endpoint_invalid", "agent_timeout", "task_interrupted", "service_shutdown", "agent_attempt_metadata_write_failed", "agent_process_failed", "unknown_execution", "agent_output_invalid_json", "agent_result_file_missing", "agent_result_file_unreadable", "agent_result_file_too_large", "agent_result_file_invalid_json", "agent_result_file_schema_invalid", "agent_result_path_invalid", "agent_image_path_forbidden", "agent_timeout_limit_exceeded", "agent_runtime_unavailable", "opencode_not_configured", "invalid_task", "invalid_reverse_image_policy", "agent_backpressure", "task_exists", "agent_provider_rate_limited", "agent_provider_server_error", "agent_connection_interrupted", "session_binding_mismatch", "session_not_resumable", "agent_executor_not_configured", "agent_executor_unavailable", "agent_executor_unauthorized", "agent_executor_invalid_response", "opencode_workspace_invalid", "opencode_workspace_mismatch", "opencode_workspace_capability_invalid", "opencode_workspace_capability_expired", "opencode_workspace_capability_unavailable", "opencode_workspace_provider_missing", } return code if code in known else "agent_executor_invalid_response" def _configured_image_root(self) -> Path: """解析设置中的图片根目录，统一相对路径基准以匹配 Compose 挂载。""" configured = Path(self.settings.image_root or self.project_root / "data" / "images").expanduser() if not configured.is_absolute(): configured = self.project_root / configured return configured.resolve() def _link(self, target: Path, source: Path) -> None: """创建或核验相对链接，拒绝覆盖不属于 runtime 的真实目录。""" if target.is_symlink(): if target.resolve() == source.resolve(): return target.unlink() elif target.exists(): raise OpenCodeError("opencode_not_configured", f"runtime 路径冲突：{target}") target.parent.mkdir(parents=True, exist_ok=True) target.symlink_to(os.path.relpath(source, target.parent), target_is_directory=True) def _config_for_runtime(self, workspace: ResolvedWorkspace | None = None) -> dict[str, Any]: """生成当前运行模式的无密钥配置。 生产 executor 只使用 broker/capability 占位符；local/host 夹具继续使用 旧 endpoint/key 环境引用，保证开发模式与已安装 OpenCode 的兼容回归。 """ config = _workspace_opencode_config(workspace) if workspace is not None else json.loads(json.dumps(RUNTIME_OPENCODE_CONFIG, ensure_ascii=False)) production_executor = str(getattr(self.settings, "public_release_profile", "local")).casefold() in {"production", "public", "1", "true", "yes", "on"} and self.executor_mode if not production_executor: options = config.setdefault("provider", {}).setdefault("mememeow", {}).setdefault("options", {}) options["baseURL"] = "{env:MEMEMEOW_OPENCODE_BASE_URL}" options["apiKey"] = "{env:MEMEMEOW_OPENCODE_API_KEY}" return config def _write_workspace_config(self, workspace: ResolvedWorkspace) -> None: """原子写入当前 workspace 的无密钥配置和外部目录权限。""" target = workspace.config_file validate_directory_path(target.parent, create=True, code="opencode_workspace_invalid") validate_file_path(target, allow_missing=True, code="opencode_workspace_invalid") content = json.dumps(self._config_for_runtime(workspace), ensure_ascii=False, indent=2) + "\n" try: if target.read_text(encoding="utf-8") == content: os.chmod(target, 0o600) return except FileNotFoundError: pass temporary = target.with_name(f".{target.name}.tmp.{os.getpid()}.{id(self)}") validate_file_path(temporary, allow_missing=True, code="opencode_workspace_invalid") descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600) try: with os.fdopen(descriptor, "w", encoding="utf-8") as handle: descriptor = -1 handle.write(content) finally: if descriptor >= 0: os.close(descriptor) os.replace(temporary, target) def _write_runtime_config(self) -> None: """兼容旧诊断入口，写入既有 local workspace 基础配置。""" target = self.workspace / "opencode.json" content = json.dumps(self._config_for_runtime(), ensure_ascii=False, indent=2) + "\n" try: if target.read_text(encoding="utf-8") == content: return except FileNotFoundError: pass temporary = target.with_name(f".{target.name}.tmp.{os.getpid()}.{id(self)}") temporary.write_text(content, encoding="utf-8") os.chmod(temporary, 0o600) os.replace(temporary, target) def prepare_runtime(self) -> None: """准备可复用 workspace，不执行任何包管理器或依赖下载。""" with self._prepare_lock: if self._runtime_ready: return if not self.settings.opencode_model: raise OpenCodeError("opencode_not_configured", "未配置 OpenCode 模型") production = str(getattr(self.settings, "public_release_profile", "local")).casefold() in {"production", "public", "1", "true", "yes", "on"} if production and self.executor_mode: if not getattr(self.settings, "model_broker_url", None): raise OpenCodeError("model_broker_endpoint_invalid", "生产模型 broker 未配置") elif not self.settings.opencode_base_url or not self.settings.opencode_api_key: raise OpenCodeError("opencode_not_configured", "未配置 OpenCode 服务地址或密钥") if self.executor_mode: if not self.executor.configured: raise OpenCodeError("agent_executor_not_configured", "Agent executor 地址或凭据 token 未配置") try: health = self.executor.health() except AgentExecutorError as exc: # 健康探针失败也必须进入任务稳定错误协议，不能把客户端异常 # 直接交给长任务服务并退化成 task_failed。 code = self._executor_error_code(exc.code, health=True) raise OpenCodeError(code, str(exc)[:500], reason_code=exc.reason_code) from exc if not bool(health.get("ready")): raise OpenCodeError("agent_runtime_unavailable", "Agent executor 健康检查未通过") else: executable = self.settings.opencode_executable if not executable or (not Path(executable).is_file() and shutil.which(executable) is None): raise OpenCodeError("opencode_not_configured", "未找到 OpenCode 可执行文件") skills_source = self.project_root / "skills" / "research-meme-context" if not isinstance(self.workspace_provider, LocalWorkspaceProvider): # 外部 provider 已在 resolve 阶段验证自己的只读 Skill 视图， # host 模式不应再要求项目根存在同一份副本。 skills_source = None # 项目自己的 OpenCode 插件依赖与前端依赖隔离；环境变量可覆盖这一默认共享目录。 shared_modules = self.settings.opencode_node_modules or self.project_root / ".opencode" / "node_modules" if (skills_source is not None and not skills_source.is_dir()) or not shared_modules.is_dir() or not (shared_modules / "@ai-sdk" / "openai").is_dir(): raise OpenCodeError("opencode_not_configured", "OpenCode skill、共享 node_modules 或 Responses provider 未预先安装") self.runtime_root.mkdir(parents=True, exist_ok=True) (self.runtime_root / "home").mkdir(parents=True, exist_ok=True) if isinstance(self.workspace_provider, LocalWorkspaceProvider): self.workspace.mkdir(parents=True, exist_ok=True) self.slots_root.mkdir(parents=True, exist_ok=True) self.log_root.mkdir(parents=True, exist_ok=True) if isinstance(self.workspace_provider, LocalWorkspaceProvider): self._write_runtime_config() if not self.executor_mode and isinstance(self.workspace_provider, LocalWorkspaceProvider): skills_source = self.project_root / "skills" / "research-meme-context" self._link(self.workspace / ".opencode" / "skills" / "research-meme-context", skills_source) self._link(self.workspace / "node_modules", shared_modules) self._runtime_ready = True def runtime_probe(self) -> dict[str, object]: """返回共享 runtime、skill 和依赖探针结果，供启动诊断使用。""" if self.executor_mode: error_code: str | None = None try: health = self.executor.health() except AgentExecutorError as exc: health = {} error_code = self._executor_error_code(exc.code, health=True) ready = bool(health.get("ready")) runtime_ready = bool(health.get("runtime_read_write")) images_ready = bool(health.get("images_read_only")) skills_ready = bool(health.get("skills_read_only")) executable_ready = bool(health.get("opencode")) socket_absent = bool(health.get("docker_socket_absent")) return { "mode": "executor", "executor_url_configured": bool(self.executor.url), "executor_token_configured": bool(self.executor.token), "executor_running": ready, "runtime_root_ready": self.runtime_root.is_dir(), "workspace_ready": self.workspace.is_dir(), "executable_ready": executable_ready, "skills_ready": skills_ready, "dependencies_ready": executable_ready, "mounts_ready": runtime_ready and images_ready and skills_ready, "runtime_read_write": runtime_ready, "images_read_only": images_ready, "skill_read_only": skills_ready, "non_root": True, "network_ready": ready, "docker_socket_absent": socket_absent, "concurrency": self.concurrency, "error_code": error_code, "verified": bool(ready and self.executor.configured and runtime_ready and images_ready and skills_ready and executable_ready and socket_absent), } executable = self.settings.opencode_executable executable_ready = bool(executable and (Path(executable).is_file() or shutil.which(executable))) skills_ready = (self.project_root / "skills" / "research-meme-context").is_dir() modules = self.settings.opencode_node_modules or self.project_root / ".opencode" / "node_modules" dependencies_ready = modules.is_dir() and (modules / "@ai-sdk" / "openai").is_dir() return { "runtime_root_ready": self.runtime_root.is_dir(), "workspace_ready": self.workspace.is_dir(), "db_path": self.db_path.name, "executable_ready": executable_ready, "skills_ready": skills_ready, "dependencies_ready": dependencies_ready, "concurrency": self.concurrency, "mode": "host-runtime-slot-lock", "verified": bool(executable_ready and skills_ready and dependencies_ready), } def probe_runtime(self) -> dict[str, object]: """提供兼容命名的 runtime 探针入口，供诊断脚本和设置页调用。""" return self.runtime_probe() def skill_hash(self) -> str: """计算 skill 文件内容哈希，用于把运行时契约写入任务记录。""" source = self.project_root / "skills" / "research-meme-context" digest = hashlib.sha256() for path in sorted((item for item in source.rglob("*") if item.is_file()), key=lambda item: item.as_posix()): digest.update(path.relative_to(source).as_posix().encode("utf-8")) digest.update(path.read_bytes()) return digest.hexdigest() def _agent_reverse_image_url(self) -> str: """返回 Agent 使用的反向图片回调地址，兼容旧版设置夹具。""" return str(getattr(self.settings, "agent_reverse_image_internal_url", None) or self.settings.reverse_image_internal_url) def build_environment( self, slot_id: int | None = None, task_id: str | None = None, callback_token: str | None = None, *, workspace: ResolvedWorkspace | None = None, ) -> dict[str, str]: """构造隔离的 OpenCode 进程环境，供后台任务和交互检查入口共同使用。""" active_workspace = workspace or ResolvedWorkspace( selector="local", directory=self.workspace, config_file=self.workspace / "opencode.json", config_dir=self.workspace / ".opencode", images_root=self._configured_image_root(), metadata_root=self._configured_image_root(), skill_root=self.project_root / "skills" / "research-meme-context", task_scratch_root=self.workspace / "tasks" / (task_id or "interactive"), task_results_root=self.runtime_root / "task-results" / (task_id or "interactive"), db_path=self.db_path, candidate_root=self.runtime_root / "candidates" / (task_id or "interactive"), local=True, ) if self.executor_mode: # API 不启动 OpenCode 子进程；该快照仅供诊断，绝不把 executor token、 # callback 根 secret 或长期模型凭据传给 Agent。候选 manifest 是唯一 # 的本地视觉输入入口。 values = { "OPENCODE_DB": "/runtime/opencode.db", "OPENCODE_CONFIG": str(active_workspace.config_file), "OPENCODE_CONFIG_DIR": str(active_workspace.config_dir), "OPENCODE_DISABLE_PROJECT_CONFIG": "1", "MEMEMEOW_REVERSE_IMAGE_INTERNAL_URL": self._agent_reverse_image_url(), "MEMEMEOW_AGENT_CANDIDATE_MANIFEST": f"/runtime/candidates/{task_id or 'interactive'}/manifest.json", "MEMEMEOW_DATA_ROOT": "/runtime", } if slot_id is not None: values["MEMEMEOW_OPENCODE_SLOT"] = str(slot_id) if task_id: values["MEMEMEOW_AGENT_TASK_ID"] = str(task_id) if callback_token: values["MEMEMEOW_AGENT_CALLBACK_TOKEN"] = callback_token return values # host 回滚模式也必须使用白名单；继承整个 API 环境会把 callback # 根 secret、executor token、数据库配置或其它 scope 配置带进 Agent 子进程。 runtime_root = self.runtime_root environment = { "PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin"), # 使用 runtime 专属 HOME，避免 host 兼容模式读取调用用户的凭据目录。 "HOME": str(self.runtime_root / "home"), "OPENCODE_DB": str(runtime_root / "opencode.db"), "OPENCODE_CONFIG": str(active_workspace.config_file), "OPENCODE_CONFIG_DIR": str(active_workspace.config_dir), # 禁止向上合并项目根配置，避免任务意外使用其他 provider 或本地凭据。 "OPENCODE_DISABLE_PROJECT_CONFIG": "1", "MEMEMEOW_DATA_ROOT": str(runtime_root), "MEMEMEOW_REVERSE_IMAGE_CACHE_ROOT": str(runtime_root / "reverse_image_cache" / "web_detection"), "MEMEMEOW_REVERSE_IMAGE_INTERNAL_URL": self._agent_reverse_image_url(), # 视觉 Skill 只接收 provider 物化的 task-scoped manifest，不获得 # 数据库、模型运行时或视觉 callback 权限。 "MEMEMEOW_AGENT_CANDIDATE_MANIFEST": str(active_workspace.candidate_manifest_path), } environment["MEMEMEOW_OPENCODE_BASE_URL"] = str(self.settings.opencode_base_url or "") environment["MEMEMEOW_OPENCODE_API_KEY"] = str(self.settings.opencode_api_key or "") shared_modules = self.settings.opencode_node_modules or self.project_root / ".opencode" / "node_modules" environment["NODE_PATH"] = str(shared_modules) if slot_id is not None: environment["MEMEMEOW_OPENCODE_SLOT"] = str(slot_id) if task_id: environment["MEMEMEOW_AGENT_TASK_ID"] = str(task_id) if callback_token: environment["MEMEMEOW_AGENT_CALLBACK_TOKEN"] = callback_token return environment def _acquire_slot(self) -> tuple[int, Any]: """获取进程内 semaphore 和跨进程 slot 文件锁。""" acquired = False while not self._closing.is_set(): if self._slot_semaphore.acquire(timeout=0.2): acquired = True break if not acquired or self._closing.is_set(): # 关闭信号可能与 semaphore 获取同时到达，不能遗留已占用配额。 if acquired: self._slot_semaphore.release() raise OpenCodeError("task_interrupted", "OpenCode runner 正在关闭") try: import fcntl except ImportError: # pragma: no cover - Windows 退化为进程内互斥 if self._closing.is_set(): self._slot_semaphore.release() raise OpenCodeError("task_interrupted", "OpenCode runner 正在关闭") try: slot_id = self._slot_ids.get(timeout=0.2) except Empty as exc: self._slot_semaphore.release() raise OpenCodeError("opencode_slot_unavailable", "无法获取 OpenCode slot") from exc if self._closing.is_set(): self._slot_ids.put(slot_id) self._slot_semaphore.release() raise OpenCodeError("task_interrupted", "OpenCode runner 正在关闭") return slot_id, None while not self._closing.is_set(): try: slot_id = self._slot_ids.get(timeout=0.2) except Empty: continue if self._closing.is_set(): self._slot_ids.put(slot_id) self._slot_semaphore.release() raise OpenCodeError("task_interrupted", "OpenCode runner 正在关闭") lock_path = self.slots_root / f"slot-{slot_id}.lock" try: lock_path.parent.mkdir(parents=True, exist_ok=True) handle = lock_path.open("a+") except OSError as exc: self._slot_ids.put(slot_id) self._slot_semaphore.release() raise OpenCodeError("opencode_slot_unavailable", "无法创建 OpenCode slot 锁") from exc try: fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: handle.close() self._slot_ids.put(slot_id) time.sleep(0.02) continue except OSError as exc: handle.close() self._slot_ids.put(slot_id) self._slot_semaphore.release() raise OpenCodeError("opencode_slot_unavailable", "无法获取 OpenCode slot 锁") from exc return slot_id, handle self._slot_semaphore.release() raise OpenCodeError("task_interrupted", "OpenCode runner 正在关闭") def _release_slot(self, slot_id: int, handle: Any) -> None: """释放跨进程 slot 文件锁和进程内 semaphore。""" if handle is not None: try: import fcntl fcntl.flock(handle.fileno(), fcntl.LOCK_UN) except (ImportError, OSError): pass handle.close() self._slot_ids.put(slot_id) self._slot_semaphore.release() def _track_process(self, process: subprocess.Popen[bytes], task_id: str | None = None) -> None: """登记由当前 runner 管理的进程组，便于并行 shutdown。""" with self._process_lock: self._process = process self._processes.add(process) self._process_tasks[process] = task_id def _mark_task_active(self, task_id: str) -> None: """登记当前进程仍在执行的 task，结果清理时保护其专属目录。""" with self._process_lock: self._active_task_ids.add(task_id) def _unmark_task_active(self, task_id: str) -> None: """移除已结束 task 的活动标记，允许后续保留策略清理产物。""" with self._process_lock: self._active_task_ids.discard(task_id) def _untrack_process(self, process: subprocess.Popen[bytes]) -> None: """从受管理进程集合移除已收束进程。""" with self._process_lock: self._processes.discard(process) self._process_tasks.pop(process, None) self._process = next(iter(self._processes), None) def _take_pre_cancelled(self, task_id: str) -> bool: """消费任务开始前收到的取消标记，阻止取消竞态继续启动外部进程。""" with self._process_lock: if task_id not in self._cancelled_task_ids: return False self._cancelled_task_ids.discard(task_id) return True @staticmethod def _event_session(event: object) -> str | None: """从公开 JSONL 事件中递归找出 session 标识。""" if isinstance(event, dict): for key in ("session_id", "sessionID", "sessionId", "id"): value = event.get(key) if isinstance(value, str) and value and (key != "id" or "session" in str(event.get("type", "")).lower()): return value for value in event.values(): found = OpenCodeRunner._event_session(value) if found: return found elif isinstance(event, list): for value in event: found = OpenCodeRunner._event_session(value) if found: return found return None @staticmethod def _process_error_diagnostic(stdout: bytes, stderr: bytes, *, secrets: tuple[str, ...] = ()) -> str: """从 CLI 的 stderr 或 JSONL error 事件提取有限且可展示的诊断。""" return OpenCodeRunner._process_error_diagnostic_stream(BytesIO(stdout), BytesIO(stderr), secrets=secrets) @staticmethod def _classify_process_failure(stdout: BinaryIO, stderr: BinaryIO) -> str: """按有限错误事件区分 provider、网络和普通进程失败。""" raw = _stream_sample(stdout, STREAM_COPY_CHUNK_BYTES * 4) raw += _stream_sample(stderr, STREAM_COPY_CHUNK_BYTES) text = raw.decode("utf-8", errors="replace").lower() statuses = [int(match) for match in re.findall(r"(?:statuscode|status|http)[^0-9]{0,8}(\d{3})", text)] if any(status == 429 for status in statuses): return "agent_provider_rate_limited" if any(500 <= status <= 599 for status in statuses): return "agent_provider_server_error" if any(token in text for token in ("econnreset", "connection reset", "connection refused", "socket hang up", "network error", "timed out")): return "agent_connection_interrupted" return "agent_process_failed" @staticmethod def _process_http_status(stdout: BinaryIO, stderr: BinaryIO) -> int | None: """从有限进程输出提取 provider HTTP 状态，供脱敏错误摘要使用。""" raw = _stream_sample(stdout, STREAM_COPY_CHUNK_BYTES * 4) raw += _stream_sample(stderr, STREAM_COPY_CHUNK_BYTES) text = raw.decode("utf-8", errors="replace").lower() statuses = [int(match) for match in re.findall(r"(?:statuscode|status|http)[^0-9]{0,8}(\d{3})", text)] return next((status for status in statuses if status == 429 or 500 <= status <= 599), None) @staticmethod def _process_error_diagnostic_stream(stdout: BinaryIO, stderr: BinaryIO, *, secrets: tuple[str, ...] = ()) -> str: """从临时输出文件提取诊断，读取量受控且不影响完整输出处理。""" stderr.seek(0) stderr_text = stderr.read(STREAM_COPY_CHUNK_BYTES).decode("utf-8", errors="replace").strip() if stderr_text: return _redact_runtime_diagnostic(stderr_text, secrets) stdout.seek(0) diagnostic: str | None = None for line in stdout: try: event = json.loads(line) except json.JSONDecodeError: continue if not isinstance(event, dict) or event.get("type") != "error": continue error = event.get("error") data = error.get("data") if isinstance(error, dict) else None if not isinstance(data, dict): continue message = data.get("message") status = data.get("statusCode") if isinstance(message, str) and message.strip(): suffix = f" (HTTP {status})" if isinstance(status, int) else "" diagnostic = _redact_runtime_diagnostic(f"{message.strip()}{suffix}", secrets) return diagnostic[:500] if diagnostic else "OpenCode 进程以非零状态退出" @staticmethod def _last_assistant_text(exported: object) -> str: """仅抽取完成 session 中最后一条 assistant 消息的 text parts。""" messages: list[object] | None = None if isinstance(exported, dict): for key in ("messages", "data"): value = exported.get(key) if isinstance(value, list) and all(isinstance(item, dict) for item in value): messages = value break elif isinstance(exported, list): messages = exported if not messages: raise OpenCodeError("agent_output_invalid_json", "完成 session 中不存在消息") def message_role(item: dict[str, Any]) -> str: info = item.get("info") role = item.get("role") if role is None and isinstance(info, dict): role = info.get("role") return str(role or "").lower() candidates = [item for item in messages if isinstance(item, dict) and message_role(item) == "assistant"] if not candidates: raise OpenCodeError("agent_output_invalid_json", "完成 session 中不存在 assistant 消息") message = candidates[-1] parts = message.get("parts", message.get("content", [])) if isinstance(parts, str): return parts if not isinstance(parts, list): raise OpenCodeError("agent_output_invalid_json", "assistant 消息不包含文本部分") texts: list[str] = [] for part in parts: if isinstance(part, str): texts.append(part) elif isinstance(part, dict) and str(part.get("type", "text")).lower() == "text" and isinstance(part.get("text"), str): texts.append(part["text"]) if not texts: raise OpenCodeError("agent_output_invalid_json", "assistant 消息不包含文本") return "".join(texts) @staticmethod def extract_candidate(text: str) -> dict[str, Any]: """严格解析唯一业务 JSON，仅兼容一个纯 JSON fenced block。""" raw = text.strip() try: value = json.loads(raw) except json.JSONDecodeError: match = re.fullmatch(r"```json\s*\n?([\s\S]*?)\n?```", raw, flags=re.IGNORECASE) if not match: raise OpenCodeError("agent_output_invalid_json", "assistant 输出不是唯一 JSON 对象") from None try: value = json.loads(match.group(1).strip()) except json.JSONDecodeError as exc: raise OpenCodeError("agent_output_invalid_json", "JSON fenced block 无法解析") from exc if not isinstance(value, dict): raise OpenCodeError("agent_output_invalid_json", "候选输出必须是 JSON 对象") return value def validate_candidate(self, candidate: dict[str, Any]) -> dict[str, Any]: """执行输出 schema、字段白名单和敏感数据边界校验。""" required = {"title", "summary", "subjects", "visible_text", "references", "meaning", "keywords", "search_queries", "uncertainties"} try: candidate = validate_agent_result(candidate, secret_inventory=secret_inventory_from_settings(self.settings)) except PublicDataError as exc: raise OpenCodeError("agent_output_schema_invalid", "候选 JSON 超出公开结果边界", reason_code=exc.code) from exc if not required.issubset(candidate): raise OpenCodeError("agent_output_schema_invalid", "候选 JSON 缺少必填字段") schema_path = self.project_root / "skills" / "research-meme-context" / "references" / "output-schema.json" try: schema = json.loads(schema_path.read_text(encoding="utf-8")) if Draft202012Validator is not None: errors = list(Draft202012Validator(schema, format_checker=FormatChecker()).iter_errors(candidate)) if errors: raise ValueError(errors[0].message) except (OSError, ValueError, TypeError, json.JSONDecodeError) as exc: raise OpenCodeError("agent_output_schema_invalid", "候选 JSON 不符合输出 schema") from exc try: context = MemeContext.model_validate(candidate) except Exception as exc: # noqa: BLE001 raise OpenCodeError("agent_output_schema_invalid", "候选 JSON 字段不符合约束") from exc return context.model_dump(mode="json", exclude_none=False) def _terminate(self, process: subprocess.Popen[bytes]) -> None: """终止任务独占的进程组，避免超时后留下子进程。""" if process.poll() is not None: return try: os.killpg(process.pid, signal.SIGTERM) process.wait(timeout=3) except (OSError, subprocess.TimeoutExpired): try: os.killpg(process.pid, signal.SIGKILL) process.wait(timeout=2) except OSError: pass except subprocess.TimeoutExpired: # 无法继续等待时由上层任务状态收束，避免关闭流程无限阻塞。 pass def _terminate_task(self, task_id: str | None) -> None: """终止宿主模式下属于指定 task 的本地进程组，不影响其它任务。""" if not task_id: return with self._process_lock: processes = [process for process, active_task_id in self._process_tasks.items() if active_task_id == task_id] for process in processes: self._terminate(process) @staticmethod def _iter_json_array_values(source: BinaryIO) -> Iterator[object]: """增量解析顶层 JSON 数组，每次只在内存保留一个元素。""" decoder = json.JSONDecoder() text_source = io.TextIOWrapper(source, encoding="utf-8") buffer = "" position = 0 eof = False def fill() -> bool: """补充固定大小文本块，避免读取器一次申请完整响应。""" nonlocal buffer, eof if eof: return False chunk = text_source.read(STREAM_COPY_CHUNK_BYTES) if not chunk: eof = True return False buffer += chunk return True def compact() -> None: """丢弃已消费前缀，避免增量缓冲区随元素数量增长。""" nonlocal buffer, position if position: buffer = buffer[position:] position = 0 def skip_whitespace() -> None: """定位下一个 JSON 符号。""" nonlocal position while True: while position < len(buffer) and buffer[position].isspace(): position += 1 if position < len(buffer) or eof or not fill(): return def consume(expected: str) -> None: """消费一个已知 JSON 符号并校验结构。""" nonlocal position skip_whitespace() if position >= len(buffer): raise json.JSONDecodeError("JSON 响应提前结束", buffer, position) if buffer[position] != expected: raise json.JSONDecodeError("JSON 数组结构无效", buffer, position) position += 1 compact() def decode_value() -> object: """解析单个数组元素；跨块时继续填充而不读取后续元素。""" nonlocal position while True: skip_whitespace() try: value, end = decoder.raw_decode(buffer, position) except json.JSONDecodeError as exc: incomplete = exc.msg.startswith("Unterminated string") or exc.pos >= len(buffer) - 1 if eof or not incomplete or not fill(): raise continue position = end compact() return value skip_whitespace() consume("[") while True: skip_whitespace() if position >= len(buffer): raise json.JSONDecodeError("JSON 数组提前结束", buffer, position) if buffer[position] == "]": consume("]") skip_whitespace() if position < len(buffer) or not eof and fill(): skip_whitespace() if position < len(buffer): raise json.JSONDecodeError("JSON 响应包含额外内容", buffer, position) return yield decode_value() skip_whitespace() if position >= len(buffer): raise json.JSONDecodeError("JSON 数组缺少分隔符", buffer, position) if buffer[position] == ",": position += 1 compact() continue if buffer[position] == "]": continue raise json.JSONDecodeError("JSON 数组缺少分隔符", buffer, position) def _session_messages( self, session_id: str, environment: dict[str, str], task_id: str | None = None, *, workspace: ResolvedWorkspace | None = None, ) -> object: """经临时 loopback server 读取完整 session，规避 export 内联附件的 CLI 截断。""" self.log_root.mkdir(parents=True, exist_ok=True) workspace_directory = workspace.directory if workspace is not None else self.workspace with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: probe.bind(("127.0.0.1", 0)) port = probe.getsockname()[1] server = subprocess.Popen( [ str(self.settings.opencode_executable), "serve", "--hostname", "127.0.0.1", "--port", str(port), ], cwd=workspace_directory, env=environment, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, ) self._track_process(server, task_id) endpoint = f"http://127.0.0.1:{port}/session/{session_id}/message" deadline = time.monotonic() + min(15, self.settings.opencode_timeout_seconds) try: while time.monotonic() < deadline: if server.poll() is not None: raise OpenCodeError("agent_export_failed", "OpenCode session server 未能启动") try: # 响应先流式落到临时文件，避免把完整 transcript 一次性放入内存。 with tempfile.TemporaryFile(mode="w+b", prefix="session-", dir=self.log_root) as payload: with urlopen(endpoint, timeout=1) as response: # noqa: S310 - endpoint 固定为本机 loopback。 shutil.copyfileobj(response, payload, length=STREAM_COPY_CHUNK_BYTES) payload.seek(0) first = payload.read(1) payload.seek(0) if first == b"[": # API 返回数组时只保留最后一条 assistant，丢弃大型工具消息。 assistant_messages: list[dict[str, Any]] = [] for item in self._iter_json_array_values(payload): if not isinstance(item, dict): continue role = item.get("role") info = item.get("info") if role is None and isinstance(info, dict): role = info.get("role") if str(role or "").lower() == "assistant": assistant_messages = [item] return {"messages": assistant_messages} return json.load(payload) except OpenCodeError: raise except (OSError, TimeoutError, URLError, json.JSONDecodeError): time.sleep(0.1) raise OpenCodeError("agent_export_failed", "无法读取 OpenCode session 消息") finally: self._terminate(server) self._untrack_process(server) def map_image_path(self, image: Path) -> Path: """把后端图片路径映射到 executor 的只读 `/images`，拒绝根目录外文件。""" root = self._configured_image_root() if self.executor_mode and root != (self.project_root / "data" / "images").resolve(): raise OpenCodeError("agent_image_root_mismatch", "executor 图片挂载与图片根配置不一致") raw_candidate = image.expanduser() # 先检查原始路径；若先 resolve，指向图片根目录内的符号链接会失去可识别性。 if raw_candidate.is_symlink(): raise OpenCodeError("agent_image_path_forbidden", "图片路径不可作为符号链接读取") candidate = raw_candidate.resolve() try: relative = candidate.relative_to(root) except ValueError as exc: raise OpenCodeError("agent_image_path_forbidden", "图片路径不在受控图片目录内") from exc if not candidate.is_file(): raise OpenCodeError("agent_image_path_forbidden", "图片路径不可作为普通文件读取") return EXECUTOR_IMAGE_ROOT / Path(relative.as_posix()) if self.executor_mode else candidate def map_host_image_path(self, image: Path) -> Path: """兼容诊断和测试入口，返回图片在当前运行时中的路径。""" return self.map_image_path(image) def task_result_paths(self, task_id: str, *, workspace: ResolvedWorkspace | None = None) -> tuple[Path, Path]: """创建任务专属结果目录并返回草稿、最终临时 JSON 路径。""" with self._process_lock: if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", task_id or ""): raise OpenCodeError("agent_result_path_invalid", "任务结果标识非法") root = workspace.task_results_root.parent if workspace is not None else self.runtime_root / "task-results" store = OpenCodeResultStore( root, result_name=RESULT_FILE_NAME, draft_name=RESULT_DRAFT_NAME, max_bytes=int(getattr(self.settings, "agent_result_max_bytes", RESULT_DEFAULT_MAX_BYTES)), ) expected_draft, expected_result = store.paths(task_id) self._ensure_no_symlink_path(self.runtime_root.parent, create=True) self._ensure_no_symlink_path(self.runtime_root, create=True) self._ensure_no_symlink_path(root, create=True) directory = root / task_id self._ensure_no_symlink_path(directory, create=True) self._ensure_no_symlink_path(directory / RESULT_DRAFT_NAME, allow_file=True) self._ensure_no_symlink_path(directory / RESULT_FILE_NAME, allow_file=True) if (directory / RESULT_DRAFT_NAME, directory / RESULT_FILE_NAME) != (expected_draft, expected_result): raise OpenCodeError("agent_result_path_invalid", "结果文件路径不受支持") return expected_draft, expected_result @staticmethod def _ensure_no_symlink_path(path: Path, *, create: bool = False, allow_file: bool = False) -> None: """逐级使用 lstat 检查 runtime 结果路径，拒绝符号链接和节点类型劫持。""" current = Path(path.anchor) if path.is_absolute() else Path() parts = path.parts[1:] if path.is_absolute() else path.parts for part in parts: current = current / part try: info = current.lstat() except FileNotFoundError: if not create: return try: current.mkdir() except FileExistsError: # 目录创建与攻击者替换可能交错，重新 lstat 后继续统一类型校验。 try: info = current.lstat() except OSError as exc: raise OpenCodeError("agent_result_path_invalid", "结果路径无法检查") from exc except OSError as exc: raise OpenCodeError("agent_result_path_invalid", "结果目录无法创建") from exc else: continue except OSError as exc: raise OpenCodeError("agent_result_path_invalid", "结果路径无法检查") from exc if stat.S_ISLNK(info.st_mode): raise OpenCodeError("agent_result_path_invalid", "结果路径包含符号链接") if current == path and allow_file and stat.S_ISREG(info.st_mode): return if not stat.S_ISDIR(info.st_mode): raise OpenCodeError("agent_result_path_invalid", "结果路径包含非目录节点") @staticmethod def _reset_task_result_files(draft_path: Path, result_path: Path) -> None: """清理本次重试的确定结果文件，避免沿用上一次尝试的成功产物。""" OpenCodeRunner._ensure_no_symlink_path(draft_path.parent) for path in (draft_path, result_path): try: info = path.lstat() if stat.S_ISLNK(info.st_mode): raise OpenCodeError("agent_result_path_invalid", "结果文件路径包含符号链接") if stat.S_ISDIR(info.st_mode): raise OpenCodeError("agent_result_path_invalid", "结果文件路径被目录占用") path.unlink(missing_ok=True) except FileNotFoundError: continue except OpenCodeError: raise except OSError as exc: raise OpenCodeError("agent_result_file_unreadable", "无法准备 Agent 结果文件路径") from exc def create_task_result_paths(self, task_id: str, *, workspace: ResolvedWorkspace | None = None) -> tuple[Path, Path]: """兼容调用方名称，创建并返回任务专属结果文件路径。""" return self.task_result_paths(task_id, workspace=workspace) def _read_result_file(self, result_path: Path, *, workspace: ResolvedWorkspace | None = None) -> dict[str, Any]: """读取有限大小的 Agent 结果文件并执行 JSON/schema/业务字段校验。""" # 结果读取只允许任务目录内的固定最终文件，避免该通用入口被拿来读取 runtime 外文件。 result_path = Path(os.path.abspath(Path(result_path).expanduser())) result_root = Path(os.path.abspath((workspace.task_results_root.parent if workspace is not None else self.runtime_root / "task-results"))) try: relative = result_path.relative_to(result_root) except ValueError as exc: raise OpenCodeError("agent_result_path_invalid", "结果文件不在受控任务目录内") from exc if len(relative.parts) != 2 or relative.name != RESULT_FILE_NAME: raise OpenCodeError("agent_result_path_invalid", "结果文件名称或层级非法") limit = int(getattr(self.settings, "agent_result_max_bytes", RESULT_DEFAULT_MAX_BYTES)) # O_NOFOLLOW 只保护最终节点；父级路径也必须逐级确认不是符号链接。 self._ensure_no_symlink_path(result_path.parent) try: info = result_path.lstat() if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode): raise OpenCodeError("agent_result_file_unreadable", "Agent 结果文件不是普通文件") size = info.st_size except FileNotFoundError as exc: raise OpenCodeError("agent_result_file_missing", "Agent 未生成结果文件") from exc except OpenCodeError: raise except OSError as exc: raise OpenCodeError("agent_result_file_unreadable", "Agent 结果文件不可读取") from exc if size > limit: raise OpenCodeError("agent_result_file_too_large", "Agent 结果文件超过大小限制") try: descriptor = os.open(result_path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) try: payload = os.read(descriptor, limit + 1) finally: os.close(descriptor) except OSError as exc: raise OpenCodeError("agent_result_file_unreadable", "Agent 结果文件不可读取") from exc if len(payload) > limit: raise OpenCodeError("agent_result_file_too_large", "Agent 结果文件超过大小限制") try: value = json.loads(payload.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError, RecursionError) as exc: raise OpenCodeError("agent_result_file_invalid_json", "Agent 结果文件不是有效 JSON") from exc if not isinstance(value, dict): raise OpenCodeError( "agent_result_file_schema_invalid", "Agent 结果文件必须是 JSON 对象", reason_code="result_object_required", ) try: return self.validate_candidate(value) except OpenCodeError as exc: raise OpenCodeError("agent_result_file_schema_invalid", str(exc), reason_code=exc.reason_code) from exc def read_result_file(self, result_path: Path, *, workspace: ResolvedWorkspace | None = None) -> dict[str, Any]: """读取并校验任务结果文件，供任务处理器和测试直接调用。""" return self._read_result_file(result_path, workspace=workspace) def cleanup_task_results(self, *, keep_task_id: str | None = None) -> int: """按保留天数和最大任务数清理结果及候选目录。""" roots = (self.runtime_root / "task-results", self.runtime_root / "candidates") with self._process_lock: # 清理逻辑不能让 root 或祖先的符号链接把删除范围带到 runtime 外部。 entries_by_root: list[tuple[Path, list[tuple[Path, float]]]] = [] for root in roots: try: self._ensure_no_symlink_path(root) info = root.lstat() except (FileNotFoundError, OSError, OpenCodeError): continue if not stat.S_ISDIR(info.st_mode): continue try: entries_info: list[tuple[Path, float]] = [] for item in root.iterdir(): try: item_info = item.lstat() except OSError: continue if stat.S_ISDIR(item_info.st_mode) and not stat.S_ISLNK(item_info.st_mode): entries_info.append((item, item_info.st_mtime)) entries_by_root.append((root, entries_info)) except OSError: continue if not entries_by_root: return 0 now = time.time() retention = int(getattr(self.settings, "agent_result_retention_days", 14)) * 86400 active_task_ids = set(self._active_task_ids) active_task_ids.update(task_id for task_id in self._process_tasks.values() if task_id) protected_task_ids = active_task_ids | ({keep_task_id} if keep_task_id else set()) max_tasks = int(getattr(self.settings, "agent_result_max_tasks", 500)) removed_task_ids: set[str] = set() for _root, entries_info in entries_by_root: entries_info = [(item, mtime) for item, mtime in entries_info if item.name not in protected_task_ids] entries_info.sort(key=lambda item: item[1], reverse=True) for index, (entry, mtime) in enumerate(entries_info): if index >= max_tasks or now - mtime > retention: shutil.rmtree(entry, ignore_errors=True) removed_task_ids.add(entry.name) return len(removed_task_ids) def _run_command( self, image: Path, prompt: str, *, slot_id: int | None = None, task_id: str | None = None, result_path: Path | None = None, callback_token: str | None = None, resume_session_id: str | None = None, workspace: ResolvedWorkspace | None = None, ) -> list[str]: """构造宿主模式单图研究 CLI 参数；续跑只能显式传入 session。""" # host 回滚支持开发者直接检查临时图片；业务 API 在提交前仍通过 map_image_path # 验证受控图片根，executor 路径则单独执行同一边界校验。 image_path = image.expanduser().resolve() active_workspace = workspace or ResolvedWorkspace( selector="local", directory=self.workspace, config_file=self.workspace / "opencode.json", config_dir=self.workspace / ".opencode", images_root=self._configured_image_root(), metadata_root=self._configured_image_root(), skill_root=self.project_root / "skills" / "research-meme-context", task_scratch_root=self.workspace / "tasks" / (task_id or "interactive"), task_results_root=self.runtime_root / "task-results" / (task_id or "interactive"), db_path=self.db_path, candidate_root=self.runtime_root / "candidates" / (task_id or "interactive"), local=True, ) executable = str(self.settings.opencode_executable or "opencode") command = [ executable, "run", # 非交互任务无法回答 OpenCode 的外部目录询问；runtime 边界负责限制可见范围。 "--auto", "--dir", str(active_workspace.directory), "--format", "json", "--file", str(image_path), "--model", str(self.settings.opencode_model), "--variant", OPENCODE_REASONING_VARIANT, "--title", f"mememeow-task-{task_id or 'interactive'}", ] if resume_session_id: command.extend(("--session", resume_session_id)) command.append(prompt) return command def run( self, image: Path, progress: Callable[[float | None, str | None], None], *, task_id: str | None = None, reverse_image_policy: str = "forbid", callback_token: str | None = None, resume_session_id: str | None = None, resume_of_attempt_id: str | None = None, processing_config_hash: str | None = None, workspace_context: TrustedWorkspaceContext | None = None, model_capability: str | None = None, analysis_observed_cost: object = None, analysis_policy: dict[str, object] | None = None, ) -> tuple[dict[str, Any], str]: """执行单张图片研究；失败也尽力返回可验证 session 诊断。""" task_id = task_id or uuid.uuid4().hex if analysis_policy is not None and not self.executor_mode: raise OpenCodeError("agent_analysis_policy_invalid", "启用分析用量控制需要 Executor 模式") normalized_observed_cost = None if analysis_policy is not None: try: normalized_observed_cost = normalize_observed_cost(analysis_observed_cost) except ModelCapabilityProviderError as exc: raise OpenCodeError(exc.code, str(exc)) from exc # host 回滚模式也需要 attempt 级诊断标识；executor 模式提交后会以 # 服务端生成的独立 attempt 覆盖该临时值。 local_executor_attempt_id = f"host-attempt-{uuid.uuid4().hex}" self._remember_executor_attempt(task_id, local_executor_attempt_id) if self._take_pre_cancelled(task_id) or self._closing.is_set(): raise OpenCodeError("task_interrupted", "OpenCode runner 正在关闭或任务已取消", executor_attempt_id=local_executor_attempt_id) context = workspace_context or TrustedWorkspaceContext( task_id=task_id, attempt_id=local_executor_attempt_id, scope_id="local", selector="local", session_id=resume_session_id, resume_of_attempt_id=resume_of_attempt_id, ) if workspace_context is not None: if workspace_context.task_id != task_id: raise OpenCodeError("opencode_workspace_mismatch", "workspace 上下文与任务不一致", executor_attempt_id=local_executor_attempt_id) # capability 的 attempt_id 必须与本次 executor 请求使用的独立 attempt # 完全相同；scope/selector/session 等业务事实仍来自调用方 claim。 context = replace(context, task_id=task_id, attempt_id=local_executor_attempt_id) if context.task_id != task_id: raise OpenCodeError("opencode_workspace_mismatch", "workspace 上下文与任务不一致", executor_attempt_id=local_executor_attempt_id) resolved_workspace = self.resolve_workspace(context) if analysis_policy is not None and model_capability is None and self.model_capability_provider is not None: policy_model = analysis_policy.get("model") policy_variant = analysis_policy.get("variant") if not isinstance(policy_model, str) or not isinstance(policy_variant, str): raise OpenCodeError("agent_analysis_policy_invalid", "分析用量策略缺少模型事实", executor_attempt_id=local_executor_attempt_id) request = ModelCapabilityRequest( task_id=task_id, attempt_id=local_executor_attempt_id, scope_id=context.scope_id, model=policy_model, variant=policy_variant, session_id=resume_session_id, resume_of_attempt_id=resume_of_attempt_id, analysis_policy=dict(analysis_policy), observed_cost=normalized_observed_cost, ) try: model_capability = capability_for_model_provider(self.model_capability_provider, request) except ModelCapabilityProviderError as exc: raise OpenCodeError(exc.code, str(exc), executor_attempt_id=local_executor_attempt_id) from exc if analysis_policy is not None and not model_capability: raise OpenCodeError( "model_capability_unavailable", "启用分析用量控制需要当前 attempt 的模型 capability", executor_attempt_id=local_executor_attempt_id, ) try: binding = ExecutionBinding( task_id=task_id, attempt_id=local_executor_attempt_id, scope_id=context.scope_id, workspace_selector=resolved_workspace.selector, input_digest=stable_input_digest( task_id, str(image), reverse_image_policy, processing_config_hash, resume_session_id, ), session_id=resume_session_id, ) except ExecutionBindingError as exc: raise OpenCodeError(exc.code, "OpenCode 执行绑定无效", executor_attempt_id=local_executor_attempt_id) from exc self._attempt_fence.bind(binding) with self._process_lock: # 即使 capability 签发失败，也保留已解析的 opaque selector 供失败诊断和 # claim fencing 写回；目录与 OpenCode 进程仍不会在此处创建。 self._workspace_selectors[task_id] = resolved_workspace.selector while len(self._workspace_selectors) > 5000: self._workspace_selectors.pop(next(iter(self._workspace_selectors))) capability = capability_for_provider(self.workspace_provider, context, resolved_workspace) if self.executor_mode and not resolved_workspace.local and not capability: raise OpenCodeError("opencode_workspace_capability_unavailable", "非 local executor workspace 缺少 capability", executor_attempt_id=local_executor_attempt_id) with self._process_lock: self._workspace_by_task[task_id] = resolved_workspace if capability: self._workspace_capabilities[task_id] = capability # provider 解析在任何 OpenCode 子进程和结果文件副作用前完成；此时才准备 # runtime 和当前 workspace 配置。 try: self.prepare_runtime() if not self.executor_mode: # 外部 provider 的 Task 临时目录和配置只在 capability 已经验证后 # 创建；解析失败不能留下可写任务目录或结果目录。 validate_directory_path(resolved_workspace.task_scratch_root, create=True, code="opencode_workspace_invalid") validate_directory_path(resolved_workspace.config_dir, create=True, code="opencode_workspace_invalid") self._write_workspace_config(resolved_workspace) slot_id, lock_handle = self._acquire_slot() except Exception: with self._process_lock: self._workspace_by_task.pop(task_id, None) self._workspace_capabilities.pop(task_id, None) raise try: with self._process_lock: self._active_task_ids.add(task_id) if self.executor_mode: self._active_executor_task_ids.add(task_id) _draft_path, result_path = self.task_result_paths(task_id, workspace=resolved_workspace) if resume_session_id and not normalize_identifier(resume_session_id, kind="session"): raise OpenCodeError("session_binding_mismatch", "续跑 session 标识无效", executor_attempt_id=local_executor_attempt_id) # 续跑必须保留既有 draft/中间产物；新业务 attempt 才初始化结果文件。 if not resume_session_id: self._reset_task_result_files(_draft_path, result_path) result_runtime_root = self.runtime_root active_image = Path(image) if context.image_relative_path: try: active_image = resolved_workspace.image_path(context.image_relative_path) except WorkspaceResolutionError as exc: raise OpenCodeError(exc.code, str(exc), executor_attempt_id=local_executor_attempt_id) from exc # 保留原始输入路径，供 Agent 读取 GIF 全部帧。 original_image = active_image.expanduser().resolve() if self.executor_mode: if context.image_relative_path: relative_image = context.image_relative_path.replace("\\", "/") else: mapped_image = self.map_image_path(active_image) try: relative_image = mapped_image.relative_to(EXECUTOR_IMAGE_ROOT).as_posix() except ValueError as exc: raise OpenCodeError("agent_image_path_forbidden", "图片路径不在 executor 受控目录内") from exc if self._take_pre_cancelled(task_id): raise OpenCodeError("task_interrupted", "Agent 任务已取消", executor_attempt_id=local_executor_attempt_id) progress(0.1, "正在提交 Agent executor 任务") try: response = self.executor.run( task_id=task_id, image_relative_path=relative_image, reverse_image_policy=reverse_image_policy if reverse_image_policy in {"forbid", "auto"} else "forbid", timeout_seconds=min(int(self.settings.opencode_timeout_seconds), int(getattr(self.settings, "agent_executor_max_timeout_seconds", 1800))), callback_token=callback_token, session_id=resume_session_id, resume_of_attempt_id=resume_of_attempt_id, processing_config_hash=processing_config_hash, visual_snapshot_sha256=( str(context.visual_match_snapshot.get("snapshot_sha256")) if isinstance(context.visual_match_snapshot, Mapping) else None ), executor_attempt_id=local_executor_attempt_id, workspace_selector=None if resolved_workspace.local else resolved_workspace.selector, workspace_capability=capability, model_capability=model_capability, analysis_policy=analysis_policy, shutdown_requested=self._closing.is_set, ) if response.executor_attempt_id: self._remember_executor_attempt(task_id, response.executor_attempt_id) self._remember_analysis_observation(task_id, response) except AgentExecutorError as exc: code = self._executor_error_code(exc.code) if exc.executor_attempt_id: self._remember_executor_attempt(task_id, exc.executor_attempt_id) cancelled_response: ExecutorTaskResponse | None = None if code in {"agent_timeout", "agent_executor_unavailable", "agent_runtime_unavailable", "agent_executor_invalid_response"}: try: cancelled_response = self.executor.cancel(exc.executor_attempt_id or task_id) except AgentExecutorError: pass decision = classify_resume_error(code, session_id=exc.session_id, target_unchanged=True, grant_state="committed") if code == "agent_timeout": raise OpenCodeError( "agent_timeout", "OpenCode 执行超时", session_id=exc.session_id or getattr(cancelled_response, "session_id", None), executor_attempt_id=exc.executor_attempt_id or getattr(cancelled_response, "executor_attempt_id", None), http_status=exc.http_status, reason_code=exc.reason_code, process_reaped=exc.process_reaped if exc.process_reaped is not None else getattr(cancelled_response, "process_reaped", None), observed_cost=exc.observed_cost or getattr(cancelled_response, "observed_cost", None), usage_checked_at=exc.usage_checked_at or getattr(cancelled_response, "usage_checked_at", None), reminder_sent=exc.reminder_sent or getattr(cancelled_response, "reminder_sent", False), termination_reason=exc.termination_reason, termination_signal=exc.termination_signal or getattr(cancelled_response, "termination_signal", None), check_stage=exc.check_stage or getattr(cancelled_response, "check_stage", None), trigger_reason=exc.trigger_reason or getattr(cancelled_response, "trigger_reason", None), ) from exc raise OpenCodeError( code, str(exc)[:500], session_id=exc.session_id or getattr(cancelled_response, "session_id", None), executor_attempt_id=exc.executor_attempt_id or getattr(cancelled_response, "executor_attempt_id", None), retryable=decision.retryable, http_status=exc.http_status, reason_code=exc.reason_code, process_reaped=exc.process_reaped if exc.process_reaped is not None else getattr(cancelled_response, "process_reaped", None), observed_cost=exc.observed_cost or getattr(cancelled_response, "observed_cost", None), usage_checked_at=exc.usage_checked_at or getattr(cancelled_response, "usage_checked_at", None), reminder_sent=exc.reminder_sent or getattr(cancelled_response, "reminder_sent", False), termination_reason=exc.termination_reason, termination_signal=exc.termination_signal or getattr(cancelled_response, "termination_signal", None), check_stage=exc.check_stage or getattr(cancelled_response, "check_stage", None), trigger_reason=exc.trigger_reason or getattr(cancelled_response, "trigger_reason", None), ) from exc if self._take_pre_cancelled(task_id): try: self.executor.cancel(response.executor_attempt_id or task_id) except AgentExecutorError: pass raise OpenCodeError("task_interrupted", "Agent 任务已取消", executor_attempt_id=response.executor_attempt_id or local_executor_attempt_id) if response.status != "succeeded": raise OpenCodeError("agent_executor_invalid_response", "Agent executor 任务未返回成功状态", session_id=response.session_id, executor_attempt_id=response.executor_attempt_id) if not response.session_id: raise OpenCodeError( "agent_output_invalid_json", "Agent executor 未返回 session 标识", executor_attempt_id=response.executor_attempt_id or local_executor_attempt_id, ) progress(0.65, "正在读取研究结果文件") candidate = self._read_result_file(result_path, workspace=resolved_workspace) progress(0.9, "正在校验并写入语境") return candidate, response.session_id resume_instruction = ( "这是同一 OpenCode session 的受控续跑；保留并检查已有草稿和中间产物，只完成尚未完成的工作。" if resume_session_id else "这是首次执行；先建立任务草稿并逐步完成研究。" ) prompt = ( "使用 research-meme-context skill 分析这张表情包；不要使用本地以图搜图。" f"原始图片路径（只读）：{json.dumps(str(original_image), ensure_ascii=False)}。" "必须直接通过视觉模型查看输入图片并识别图片文字，禁止调用任何本地文字识别程序、命令或脚本。" "遇到错误时自行尝试可行的替代方案；确认无法解决时，简短说明原因并退出。" f"本任务 reverse_image_policy={reverse_image_policy if reverse_image_policy in {'forbid', 'auto'} else 'forbid'}；只能通过项目内部反向图片接口使用能力，绝不读取或请求供应商密钥。" f"{resume_instruction}" f"结果必须写入 {resolved_workspace.result_path}。" f"先在同目录写入 {resolved_workspace.draft_path}，" "使用 output-schema.json 校验完整 JSON 对象后，使用同一文件系统的原子 rename/mv 将草稿替换为最终文件。" "不要把业务 JSON 作为 assistant 文本交付，不要写入数据库；完成后简短说明即可。" ) command = self._run_command(active_image, prompt, slot_id=slot_id, task_id=task_id, result_path=result_path, callback_token=callback_token, resume_session_id=resume_session_id, workspace=resolved_workspace) execution_command = command environment = self.build_environment(slot_id, task_id, callback_token, workspace=resolved_workspace) progress(0.1, f"正在启动语境研究（slot {slot_id}）") # 由临时文件承接完整事件流，避免按总字节数拒绝合法的大输出，也避免管道缓存堆积在内存。 with ( tempfile.TemporaryFile(mode="w+b", prefix="stdout-", dir=self.log_root) as stdout_stream, tempfile.TemporaryFile(mode="w+b", prefix="stderr-", dir=self.log_root) as stderr_stream, ): process = subprocess.Popen( execution_command, cwd=resolved_workspace.directory, env=environment, stdin=subprocess.DEVNULL, stdout=stdout_stream, stderr=stderr_stream, start_new_session=True, ) self._track_process(process, task_id) timed_out = False interrupted = False try: interrupted = self._take_pre_cancelled(task_id) if interrupted or self._closing.is_set(): self._terminate(process) else: # stdout/stderr 已重定向到临时文件；communicate 只等待进程。 process.communicate(timeout=self.settings.opencode_timeout_seconds) except subprocess.TimeoutExpired: self._terminate(process) timed_out = True finally: self._untrack_process(process) interrupted = interrupted or self._take_pre_cancelled(task_id) log_path = self.log_root / f"{hashlib.sha256(str(image).encode()).hexdigest()[:16]}.jsonl" stdout_stream.flush() stderr_stream.flush() self._write_diagnostic_log(stdout_stream, log_path) session_id = None stdout_stream.seek(0) for line in stdout_stream: if not line.strip(): continue try: event = json.loads(line) except json.JSONDecodeError: # 失败路径只要已有合法 session 就继续保留它；成功路径仍由 # 结果文件/schema 校验收束，不把噪声日志当成 session。 continue found = self._event_session(event) if found: normalized_found = normalize_identifier(found, kind="session") if normalized_found is None: continue if resume_session_id and normalized_found != resume_session_id: raise OpenCodeError("session_binding_mismatch", "OpenCode 返回了不匹配的 session", session_id=normalized_found, executor_attempt_id=local_executor_attempt_id) if session_id and normalized_found != session_id: raise OpenCodeError("session_binding_mismatch", "OpenCode 返回了多个不匹配的 session", session_id=normalized_found, executor_attempt_id=local_executor_attempt_id) session_id = normalized_found if not session_id: session_id = resume_session_id if interrupted: raise OpenCodeError("task_interrupted", "Agent 任务已取消", session_id=session_id, executor_attempt_id=local_executor_attempt_id, process_reaped=process.poll() is not None) if self._closing.is_set() and process.returncode != 0: raise OpenCodeError("service_shutdown", "服务正常关闭，Agent 执行已中断", session_id=session_id, executor_attempt_id=local_executor_attempt_id, process_reaped=process.poll() is not None) if timed_out: raise OpenCodeError("agent_timeout", "OpenCode 执行超时", session_id=session_id, executor_attempt_id=local_executor_attempt_id) if process.returncode != 0: diagnostic = self._process_error_diagnostic_stream( stdout_stream, stderr_stream, secrets=( str(getattr(self.settings, "opencode_api_key", "") or ""), str(getattr(self.settings, "agent_executor_token", "") or ""), callback_token or "", ), ) failure_code = self._classify_process_failure(stdout_stream, stderr_stream) http_status = self._process_http_status(stdout_stream, stderr_stream) decision = classify_resume_error(failure_code, session_id=session_id, target_unchanged=True, grant_state="committed") raise OpenCodeError(failure_code, diagnostic, session_id=session_id, executor_attempt_id=local_executor_attempt_id, retryable=decision.retryable, http_status=http_status) if not session_id: raise OpenCodeError("agent_output_invalid_json", "OpenCode 未返回 session 标识", executor_attempt_id=local_executor_attempt_id) progress(0.65, "正在读取研究结果文件") try: candidate = self._read_result_file(result_path, workspace=resolved_workspace) except OpenCodeError as exc: if resume_session_id or exc.code != "agent_result_file_missing": # 续跑或已有结果文件损坏时必须以文件校验为边界，不能退回 # transcript 解析，否则会绕过副作用和产物完整性判定。 raise # 仅在 host 未生成结果文件时兼容旧 session 读取；executor 任务在上方已直接失败。 try: data = self._session_messages(session_id, environment, task_id, workspace=resolved_workspace) except TypeError: # 保持旧测试/扩展覆写三参数 session reader 的兼容。 data = self._session_messages(session_id, environment, task_id) candidate = self.validate_candidate(self.extract_candidate(self._last_assistant_text(data))) progress(0.9, "正在校验并写入语境") return candidate, session_id finally: self._attempt_fence.release(task_id, local_executor_attempt_id) self.cleanup_task_results(keep_task_id=task_id) self._unmark_task_active(task_id) with self._process_lock: self._active_executor_task_ids.discard(task_id) self._cancelled_task_ids.discard(task_id) self._workspace_by_task.pop(task_id, None) self._workspace_capabilities.pop(task_id, None) self._release_slot(slot_id, lock_handle) @staticmethod def _write_diagnostic_log(source: BinaryIO, target: Path) -> None: """保存有限诊断前缀；完整输出仍只存在于本次运行的临时文件。""" source.seek(0) with target.open("wb") as destination: remaining = DIAGNOSTIC_LOG_BYTES while remaining > 0: chunk = source.read(min(STREAM_COPY_CHUNK_BYTES, remaining)) if not chunk: break destination.write(chunk) remaining -= len(chunk) def shutdown(self) -> None: """终止当前受管理进程，供应用生命周期收束调用。""" self._closing.set() with self._process_lock: executor_business_ids = list(self._active_executor_task_ids) for task_id in executor_business_ids: try: self.executor.cancel(self.executor.attempt_id_for(task_id) or task_id, reason="service_shutdown") except AgentExecutorError as exc: logger.warning("停机中断请求失败：task={} error={}", task_id, exc.code) with self._process_lock: processes = list(self._processes) if self._process is not None and self._process not in processes: processes.append(self._process) for process in processes: self._terminate(process) def cancel(self, task_id: str) -> None: """取消指定研究任务；executor 请求远端取消，host 终止对应本地进程组。""" if not task_id: return with self._process_lock: if len(self._cancelled_task_ids) >= 5000: for stale_id in list(self._cancelled_task_ids - self._active_task_ids): self._cancelled_task_ids.discard(stale_id) if len(self._cancelled_task_ids) < 4000: break self._cancelled_task_ids.add(task_id) if self.executor_mode: try: self.executor.cancel(self.executor.attempt_id_for(task_id) or task_id) except AgentExecutorError: # 数据库任务已被取消时，executor 可能已经重启或忘记任务；不阻塞 API 收束。 return return self._terminate_task(task_id)