MemeMeow/executor/server.py

"""受控 OpenCode executor HTTP 服务。

该模块运行在 ``mememeow-agent-runtime`` 容器内，是后端与 OpenCode 之间的
唯一执行边界。它只接受固定字段的研究任务，不接受 shell、任意命令、任意
环境变量或任意工作目录；任务结果仍通过共享 ``/runtime/task-results`` 文件
协议交付给后端。
"""

from __future__ import annotations

import hashlib
import hmac
import json
import os
import re
import signal
import socket
import stat
import subprocess
import tempfile
import threading
import time
from concurrent.futures import Future
from dataclasses import dataclass, field
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path, PurePosixPath
from typing import Any
from decimal import Decimal
from loguru import logger
from urllib.parse import unquote, urlsplit
from opentelemetry import metrics, trace
from runtime_telemetry import TelemetryConfig, extract_carrier, protected_content, start_telemetry, trace_carrier
from executor.sqlite_telemetry import SQLiteTelemetry

try:
    from backend.opencode import RUNTIME_OPENCODE_CONFIG
except ModuleNotFoundError:  # pragma: no cover - Agent 镜像只复制 executor 公共快照
    from executor.runtime_opencode_config import RUNTIME_OPENCODE_CONFIG
from executor.model_capability import (
    MODEL_BROKER_URL_ENV,
    MODEL_CAPABILITY_FIELD,
    MODEL_CAPABILITY_ENV,
    ModelCapabilityError,
    validate_model_broker_url,
    validate_model_capability,
    validate_model_name,
)
try:
    from backend.opencode_workspace import (
        SELECTOR_RE,
        WorkspaceCapabilityError,
        WorkspaceCapabilitySigner,
        build_edit_permission_rules,
        validate_directory_path,
        validate_file_path,
    )
except ModuleNotFoundError as exc:  # pragma: no cover - Agent 镜像只复制 executor 快照
    if exc.name != "backend":
        raise
    from executor.opencode_workspace import (
        SELECTOR_RE,
        WorkspaceCapabilityError,
        WorkspaceCapabilitySigner,
        build_edit_permission_rules,
        validate_directory_path,
        validate_file_path,
    )
from executor.agent_limits import validate_agent_concurrency
from executor.analysis_policy import AnalysisPolicyError, parse_analysis_policy
from executor.analysis_monitor import AnalysisMonitor, AnalysisControlError
from executor.attempt_diagnostics import AttemptOutput, configure_executor_diagnostics, stream_sample as _stream_sample
from executor.process_supervisor import ProcessSupervisor
from executor.database_maintenance import DatabaseMaintenance
from executor.result_store import ExecutorResultStore, ExecutorResultStoreError
from executor.task_queue import ExecutionQueue
from executor.token import ExecutorTokenError, ensure_token_file, read_token_file
try:
    from backend.public_dto import PublicDataError, secret_inventory_from_mapping, validate_agent_result
except ModuleNotFoundError as exc:  # pragma: no cover - Agent 镜像只复制 executor 快照
    if exc.name != "backend":
        raise
    from executor.public_dto import PublicDataError, secret_inventory_from_mapping, validate_agent_result
try:
    from backend.visual_snapshot import VisualMatchSnapshotError, validate_visual_match_snapshot_manifest
except ModuleNotFoundError as exc:  # pragma: no cover - Agent 镜像只复制无状态协议快照
    if exc.name not in {"backend", "backend.visual_snapshot"}:
        raise
    try:
        from executor.visual_snapshot import VisualMatchSnapshotError, validate_visual_match_snapshot_manifest
    except ModuleNotFoundError:
        # 最小验证镜像可能只复制 executor 公共快照；此时加载同一份无状态校验
        # 实现，避免 executor 因可选视觉 manifest 文件缺失而无法启动。
        class VisualMatchSnapshotError(ValueError):
            """最小 executor 镜像中的视觉 manifest 校验错误。"""

        def validate_visual_match_snapshot_manifest(value: object, *, expected_sha256: str | None = None) -> dict[str, object]:
            """在无完整 backend 源码时拒绝无法验证的视觉 manifest。"""
            del value, expected_sha256
            raise VisualMatchSnapshotError("visual_match_snapshot_invalid")

TASK_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}\Z")
SESSION_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,254}\Z")
CONFIG_HASH_RE = re.compile(r"[0-9a-fA-F]{64}\Z")
RESULT_FILE_NAME = "result.json.tmp"
RESULT_DRAFT_NAME = "result.json.draft"
ATTEMPT_METADATA_NAME = ".executor-attempts.json"
RUNTIME_ROOT = Path(os.getenv("MEMEMEOW_EXECUTOR_RUNTIME_ROOT", "/runtime"))
IMAGE_ROOT = Path(os.getenv("MEMEMEOW_EXECUTOR_IMAGE_ROOT", "/images"))
WORKSPACE = RUNTIME_ROOT / "workspace"
RESULT_ROOT = RUNTIME_ROOT / "task-results"
LOG_ROOT = RUNTIME_ROOT / "logs"
ANALYSIS_SOCKET_ROOT = RUNTIME_ROOT / "analysis-sockets"
SKILL_ROOT = Path(os.getenv("MEMEMEOW_EXECUTOR_SKILL_ROOT", "/skills/research-meme-context"))
WORKSPACE_ROOT = Path(os.getenv("MEMEMEOW_EXECUTOR_WORKSPACE_ROOT", str(RUNTIME_ROOT / "workspaces")))
DEFAULT_MAX_RESULT_BYTES = 1024 * 1024
MAX_CANDIDATE_FILE_BYTES = 64 * 1024 * 1024
MAX_CANDIDATE_ROOT_BYTES = 256 * 1024 * 1024
# session 只解析完整 JSONL 行；单行超过上限时跳过，避免异常输出占满内存。
SESSION_CAPTURE_LINE_BYTES = 1024 * 1024
ALLOWED_REQUEST_FIELDS = frozenset(
    {
        "task_id",
        "business_task_id",
        "executor_attempt_id",
        "resume_of_attempt_id",
        "session_id",
        "processing_config_hash",
        "image_relative_path",
        "reverse_image_policy",
        "timeout_seconds",
        "wait",
        "callback_token",
        "trace_context",
        "trace_attributes",
        "workspace_selector",
        "workspace_capability",
        "visual_snapshot_sha256",
        "analysis_policy",
        MODEL_CAPABILITY_FIELD,
    }
)
REQUIRED_RESULT_FIELDS = frozenset(
    {"title", "summary", "subjects", "visible_text", "references", "meaning", "keywords", "search_queries", "uncertainties"}
)
TASK_HISTORY_LIMIT = 5000
_EXECUTOR_ERROR_CODES = frozenset(
    {
        "agent_analysis_policy_missing",
        "agent_analysis_policy_invalid",
        "agent_analysis_usage_unavailable",
        "agent_analysis_reminder_plugin_unavailable",
        "agent_maximum_analysis_depth_exceeded",
        "agent_timeout",
        "task_interrupted",
        "agent_process_failed",
        "agent_executor_internal_error",
        "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",
        "opencode_workspace_invalid",
        "opencode_workspace_mismatch",
        "opencode_workspace_capability_invalid",
        "opencode_workspace_capability_expired",
        "opencode_workspace_capability_unavailable",
        "model_capability_invalid",
        "model_capability_unavailable",
        "model_broker_endpoint_invalid",
        "model_name_invalid",
        "visual_candidate_materialization_failed",
        "visual_match_snapshot_invalid",
    }
)
_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"(?<![A-Za-z0-9:/])/(?:[^\s,;:'\"()\[\]{}]+/)+[^\s,;:'\"()\[\]{}]+")


class _ProcessFailure(RuntimeError):
    """保存 OpenCode 进程失败的稳定错误码和受保护诊断。"""

    def __init__(
        self,
        code: str,
        message: str,
        *,
        http_status: int | None = None,
        analysis_diagnostic: dict[str, object] | None = None,
    ):
        """初始化进程失败；诊断只在受信 Executor 调用链中传递。"""
        super().__init__(message)
        self.code = code
        self.http_status = http_status
        self.analysis_diagnostic = analysis_diagnostic


class _ResultFileValidationFailure(RuntimeError):
    """携带结果文件内部原因码的稳定 executor 异常。"""

    def __init__(self, code: str, *, reason_code: str | None = None) -> None:
        """初始化结果文件失败；外部错误码仍保持原有协议。"""
        super().__init__(code)
        self.reason_code = reason_code


def _env_int(name: str, default: int, minimum: int, maximum: int) -> int:
    """读取并限制 executor 的整数配置，非法值使用安全默认值。"""
    try:
        value = int(os.getenv(name, str(default)))
    except (TypeError, ValueError):
        return default
    return max(minimum, min(value, maximum))


def _env_agent_concurrency(name: str, default: int) -> int:
    """读取 executor 的 Agent 运行并发，只校验正整数。"""
    raw = os.getenv(name)
    if raw is None:
        value = default
    else:
        try:
            value = int(raw)
        except (TypeError, ValueError) as exc:
            raise ValueError("agent_concurrency_invalid") from exc
    return validate_agent_concurrency(value)


def _json_error(code: str, message: str, *, reason_code: str | None = None) -> dict[str, object]:
    """构造不包含本地路径、命令或秘密的稳定错误响应。"""
    error: dict[str, object] = {"error": code, "message": message}
    if isinstance(reason_code, str) and re.fullmatch(r"[a-z0-9][a-z0-9_.-]{0,127}", reason_code):
        error["reason_code"] = reason_code
    return error


def _redact_diagnostic(value: str, secrets: tuple[str, ...] = ()) -> str:
    """从上游诊断中移除已知密钥和常见凭据格式，保留有限可读错误。"""
    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)
    value = _GENERIC_PATH_PATTERN.sub("[PATH]", value)
    return value[:500]


def _safe_json(value: Any) -> bytes:
    """序列化受控响应，避免响应中出现非 JSON 值。"""
    return json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8")


def _session_id_from_event(value: Any) -> str | None:
    """从 OpenCode JSONL 事件递归提取 session 标识。"""
    if isinstance(value, dict):
        for key in ("session_id", "sessionID", "sessionId"):
            candidate = value.get(key)
            if isinstance(candidate, str) and candidate:
                return candidate
        for nested in value.values():
            found = _session_id_from_event(nested)
            if found:
                return found
    elif isinstance(value, list):
        for nested in value:
            found = _session_id_from_event(nested)
            if found:
                return found
    return None


def _diagnostic(stdout: bytes, stderr: bytes, *, secrets: tuple[str, ...] = ()) -> str:
    """从有限输出中提取安全诊断，不返回完整 transcript 或密钥。"""
    text = stderr[:2048].decode("utf-8", errors="replace").strip()
    if text:
        return _redact_diagnostic(text, secrets)
    for line in stdout.splitlines()[:128]:
        try:
            event = json.loads(line)
        except (TypeError, json.JSONDecodeError):
            continue
        if isinstance(event, dict) and event.get("type") == "error":
            error = event.get("error")
            data = error.get("data") if isinstance(error, dict) else None
            message = data.get("message") if isinstance(data, dict) else None
            if isinstance(message, str) and message.strip():
                return _redact_diagnostic(message.strip(), secrets)
    return "OpenCode 进程执行失败"


def _relative_image_path(value: object) -> Path:
    """验证图片相对路径，拒绝绝对路径、父级跳转和符号链接入口。"""
    if not isinstance(value, str) or not value or len(value) > 512 or "\\" in value or re.match(r"^[A-Za-z]:", value):
        raise ValueError("agent_image_path_forbidden")
    raw_parts = value.split("/")
    if any(not part or part in {".", ".."} for part in raw_parts) or any(ord(character) < 0x20 for character in value):
        raise ValueError("agent_image_path_forbidden")
    normalized = PurePosixPath(value)
    if normalized.is_absolute() or not normalized.parts:
        raise ValueError("agent_image_path_forbidden")
    return Path(*normalized.parts)


@dataclass
class TaskState:
    """一个 executor attempt 的受控状态和取消句柄。"""

    task_id: str
    business_task_id: str
    executor_attempt_id: str
    image_relative_path: str
    reverse_image_policy: str
    timeout_seconds: int
    processing_config_hash: str | None = None
    session_id: str | None = None
    resume_of_attempt_id: str | None = None
    is_resume: bool = False
    callback_token: str | None = field(default=None, repr=False)
    workspace_selector: str = "local"
    workspace_capability: str | None = field(default=None, repr=False)
    model_capability: str | None = field(default=None, repr=False)
    workspace_directory: Path | None = field(default=None, repr=False)
    images_root: Path | None = field(default=None, repr=False)
    metadata_root: Path | None = field(default=None, repr=False)
    skill_root: Path | None = field(default=None, repr=False)
    candidate_root: Path | None = field(default=None, repr=False)
    visual_snapshot_sha256: str | None = field(default=None, repr=False)
    task_scratch_root: Path | None = field(default=None, repr=False)
    config_file: Path | None = field(default=None, repr=False)
    config_dir: Path | None = field(default=None, repr=False)
    status: str = "queued"
    created_at: float = field(default_factory=time.time)
    started_at: float | None = None
    completed_at: float | None = None
    process_reaped: bool = True
    error: dict[str, object] | None = None
    result_path: str | None = None
    process: subprocess.Popen[bytes] | None = field(default=None, repr=False)
    cancel_event: threading.Event = field(default_factory=threading.Event, repr=False)
    cancel_reason: str | None = None
    done: threading.Event = field(default_factory=threading.Event, repr=False)
    analysis_policy: dict[str, object] | None = field(default=None, repr=False)
    observed_cost: str | None = None
    usage_checked_at: str | None = None
    reminder_sent: bool = False
    database_maintenance: dict[str, object] | None = None
    trace_context: dict[str, str] = field(default_factory=dict, repr=False)
    trace_attributes: dict[str, str] = field(default_factory=dict, repr=False)

    def public(self) -> dict[str, object]:
        """返回 API 可见状态，不暴露进程对象和 executor 本地实现细节。"""
        return {
            "task_id": self.business_task_id,
            "business_task_id": self.business_task_id,
            "executor_attempt_id": self.executor_attempt_id,
            "status": self.status,
            "image_relative_path": self.image_relative_path,
            "reverse_image_policy": self.reverse_image_policy,
            "session_id": self.session_id,
            "resume_of_attempt_id": self.resume_of_attempt_id,
            "processing_config_hash": self.processing_config_hash,
            "is_resume": self.is_resume,
            "result_path": self.result_path,
            "error": self.error,
            "created_at": self.created_at,
            "started_at": self.started_at,
            "completed_at": self.completed_at,
            "process_reaped": self.process_reaped,
            "workspace_selector": self.workspace_selector,
            "observed_cost": self.observed_cost,
            "usage_checked_at": self.usage_checked_at,
            "reminder_sent": self.reminder_sent,
            "database_maintenance": self.database_maintenance,
        }


@dataclass(frozen=True)
class WorkspaceLayout:
    """executor 内部已验证的 selector 目录布局。"""

    selector: str
    directory: Path
    images_root: Path
    metadata_root: Path
    skill_root: Path
    candidate_root: Path
    task_scratch_root: Path
    config_file: Path
    config_dir: Path
    local: bool = False

    @property
    def candidate_manifest_path(self) -> Path:
        """返回当前 executor Task 固定的候选 manifest 路径。"""
        return self.candidate_root / "manifest.json"


class Executor:
    """在容器内管理固定 OpenCode 任务接口的并发执行器。"""

    def __init__(self) -> None:
        """初始化共享目录、认证配置和有限并发池。"""
        # Agent 运行身份创建的 runtime 文件不应继承镜像默认 umask 的 group/other 位。
        os.umask(0o077)
        self.token_file = os.getenv("MEMEMEOW_AGENT_EXECUTOR_TOKEN_FILE", "").strip()
        # 环境变量兼容显式 host/测试夹具，但空白 token 必须和 token 文件一样失败关闭。
        self.token = os.getenv("MEMEMEOW_AGENT_EXECUTOR_TOKEN", "").strip()
        self.token_error: str | None = None
        if not self.token and self.token_file:
            try:
                self.token = ensure_token_file(self.token_file)
            except ExecutorTokenError as exc:
                self.token_error = str(exc)
        try:
            self.model = validate_model_name(os.getenv("MEMEMEOW_OPENCODE_MODEL", ""))
        except ModelCapabilityError:
            self.model = ""
        self.release_profile = os.getenv("MEMEMEOW_PUBLIC_RELEASE_PROFILE", "local").strip().casefold()
        self.model_broker_url = os.getenv(MODEL_BROKER_URL_ENV, "").strip()
        self.legacy_base_url = os.getenv("MEMEMEOW_OPENCODE_BASE_URL", "").strip()
        self.legacy_api_key = os.getenv("MEMEMEOW_OPENCODE_API_KEY", "").strip()
        self.opencode_executable = os.getenv("MEMEMEOW_OPENCODE_EXECUTABLE", "opencode")
        self.max_workers = _env_agent_concurrency("MEMEMEOW_OPENCODE_CONCURRENCY", 1)
        self.max_timeout = _env_int("MEMEMEOW_AGENT_EXECUTOR_MAX_TIMEOUT_SECONDS", 1800, 1, 7200)
        self.max_result_bytes = _env_int("MEMEMEOW_AGENT_RESULT_MAX_BYTES", DEFAULT_MAX_RESULT_BYTES, 1024, 16 * 1024 * 1024)
        capability_key = os.getenv("MEMEMEOW_WORKSPACE_CAPABILITY_KEY", os.getenv("MEMEMEOW_AGENT_WORKSPACE_CAPABILITY_KEY", ""))
        self.capability_signer = WorkspaceCapabilitySigner(capability_key)
        # workspace root 只允许由部署预装配；executor 不因请求 selector 创建新的
        # scope 目录。未配置时仍保留旧 local 请求兼容路径。
        self.workspace_root = Path(os.getenv("MEMEMEOW_EXECUTOR_WORKSPACE_ROOT", str(WORKSPACE_ROOT)))
        self.lock = threading.RLock()
        self.tasks: dict[str, TaskState] = {}
        # queue 只负责线程池生命周期；任务状态、attempt 和结果事实仍由 Executor
        # 控制面维护，避免 HTTP 入口直接持有线程池细节。
        self.pool = ExecutionQueue(self.max_workers, thread_name_prefix="mememeow-opencode")
        self.process_supervisor = ProcessSupervisor()
        self.database_maintenance = DatabaseMaintenance()
        self.futures: dict[str, Future[None]] = {}
        self.ready_error: str | None = None
        self._prepare_runtime()

    @property
    def model_broker_configured(self) -> bool:
        """判断生产模型 broker endpoint 是否已配置且格式有效。"""
        try:
            validate_model_broker_url(self.model_broker_url)
        except ModelCapabilityError:
            return False
        return True

    @property
    def model_configured(self) -> bool:
        """判断当前 profile 是否具备模型连接材料。

        生产 executor 只能使用 broker endpoint 和任务 capability；local profile
        保留旧 endpoint/key 夹具兼容，但该兼容值不会进入生产配置。
        """
        if not self.model:
            return False
        if self.release_profile in {"production", "public", "1", "true", "yes", "on"}:
            return self.model_broker_configured
        return self.model_broker_configured or bool(self.legacy_base_url and self.legacy_api_key)

    def _prepare_runtime(self) -> None:
        """创建共享 runtime 目录并确保 executor 以非 root 可写方式启动。"""
        try:
            for path in (RUNTIME_ROOT, WORKSPACE, RESULT_ROOT, LOG_ROOT, ANALYSIS_SOCKET_ROOT, RUNTIME_ROOT / "candidates", RUNTIME_ROOT / "home"):
                path.mkdir(parents=True, exist_ok=True)
            database_path = RUNTIME_ROOT / "opencode.db"
            database_path.touch(exist_ok=True)
            os.chmod(database_path, stat.S_IRUSR | stat.S_IWUSR)
            if any(not os.access(path, os.R_OK | os.W_OK) for path in (RUNTIME_ROOT, WORKSPACE, RESULT_ROOT, LOG_ROOT)):
                self.ready_error = "runtime_not_writable"
                return
            if self.token_error:
                self.ready_error = self.token_error
            elif not self.token:
                self.ready_error = "executor_token_not_configured"
            elif not self.model_configured:
                self.ready_error = "model_broker_not_configured"
            elif not shutil_which(self.opencode_executable):
                self.ready_error = "opencode_executable_missing"
        except OSError:
            self.ready_error = "runtime_not_ready"

    def health(self) -> dict[str, object]:
        """返回真实 executor 健康状态，供 Compose 和后端探针使用。"""
        runtime_ok = RUNTIME_ROOT.is_dir() and os.access(RUNTIME_ROOT, os.R_OK | os.W_OK)
        workspace_ok = WORKSPACE.is_dir() and os.access(WORKSPACE, os.R_OK | os.W_OK)
        result_ok = RESULT_ROOT.is_dir() and os.access(RESULT_ROOT, os.R_OK | os.W_OK)
        logs_ok = LOG_ROOT.is_dir() and os.access(LOG_ROOT, os.R_OK | os.W_OK)
        image_ok = IMAGE_ROOT.is_dir() and os.access(IMAGE_ROOT, os.R_OK) and not os.access(IMAGE_ROOT, os.W_OK)
        skill_ok = SKILL_ROOT.is_dir() and os.access(SKILL_ROOT, os.R_OK) and not os.access(SKILL_ROOT, os.W_OK)
        executable_ok = bool(shutil_which(self.opencode_executable))
        socket_absent = not Path("/var/run/docker.sock").exists()
        token_ok = bool(self.token)
        if self.token_file:
            try:
                token_ok = hmac.compare_digest(read_token_file(self.token_file), self.token)
            except ExecutorTokenError:
                token_ok = False
        ready = bool(not self.ready_error and token_ok and runtime_ok and workspace_ok and result_ok and logs_ok and image_ok and skill_ok and executable_ok and socket_absent)
        return {
            "status": "ok" if ready else "degraded",
            "ready": ready,
            "executor": "mememeow-agent-executor",
            "opencode": executable_ok,
            "runtime_read_write": runtime_ok and workspace_ok and result_ok and logs_ok,
            "images_read_only": image_ok,
            "skills_read_only": skill_ok,
            "docker_socket_absent": socket_absent,
            "token_configured": token_ok,
            "opencode_configured": self.model_configured,
            "model_broker_configured": self.model_broker_configured,
            "capacity": self.max_workers,
            "queued": self._queued_count(),
            "error": self.ready_error,
        }

    def _queued_count(self) -> int:
        """返回尚未开始的任务数量，调用方可在锁外使用。"""
        with self.lock:
            return sum(1 for task in self.tasks.values() if task.status == "queued")

    @staticmethod
    def _checked_directory(path: Path, *, create: bool = False, missing_code: str = "opencode_workspace_invalid") -> Path:
        """在 executor 边界逐级拒绝符号链接、文件和越界目录。"""
        try:
            return validate_directory_path(path, create=create, code=missing_code)
        except Exception as exc:  # noqa: BLE001 - executor 对外只返回稳定错误码
            raise ValueError(missing_code) from exc

    def _workspace_layout(self, *, selector: str | None, business_task_id: str) -> WorkspaceLayout:
        """把 selector 解析到固定布局；未知 selector 不创建任何 scope 目录。"""
        if selector is None:
            directory = self._checked_directory(WORKSPACE)
            images_root = self._checked_directory(IMAGE_ROOT)
            metadata_root = images_root
            skill_root = self._checked_directory(SKILL_ROOT)
            base = directory
            local = True
        else:
            if not isinstance(selector, str) or not SELECTOR_RE.fullmatch(selector):
                raise ValueError("opencode_workspace_invalid")
            if selector == "local":
                directory = self._checked_directory(WORKSPACE)
                images_root = self._checked_directory(IMAGE_ROOT)
                metadata_root = images_root
                skill_root = self._checked_directory(SKILL_ROOT)
                local = True
                task_scratch = directory / "tasks" / business_task_id
                candidate_root = RUNTIME_ROOT / "candidates" / business_task_id
                return WorkspaceLayout(
                    selector="local",
                    directory=directory,
                    images_root=images_root,
                    metadata_root=metadata_root,
                    skill_root=skill_root,
                    candidate_root=candidate_root,
                    task_scratch_root=task_scratch,
                    config_file=task_scratch / "opencode.json",
                    config_dir=task_scratch / ".opencode",
                    local=True,
                )
            root = self._checked_directory(self.workspace_root, missing_code="opencode_workspace_invalid")
            base = root / selector
            self._checked_directory(base, missing_code="opencode_workspace_invalid")
            directory = self._checked_directory(base / "workspace")
            images_root = self._checked_directory(base / "images")
            metadata_root = self._checked_directory(base / "metadata")
            skill_root = self._checked_directory(base / "skills")
            local = False
        task_scratch = directory / "tasks" / business_task_id
        # 只检查祖先；创建由 _run 在锁定任务后完成，校验失败不会留下目录。
        tasks_root = directory / "tasks"
        try:
            tasks_root.lstat()
        except FileNotFoundError:
            pass
        else:
            self._checked_directory(tasks_root)
        config_dir = task_scratch / ".opencode"
        candidate_parent = (base / "candidates") if not local else (base.parent / "candidates")
        self._checked_directory(candidate_parent, create=True)
        candidate_root = candidate_parent / business_task_id
        if candidate_root.exists() or candidate_root.is_symlink():
            self._checked_directory(candidate_root)
        return WorkspaceLayout(
            selector=selector or "local",
            directory=directory,
            images_root=images_root,
            metadata_root=metadata_root,
            skill_root=skill_root,
            candidate_root=candidate_root,
            task_scratch_root=task_scratch,
            config_file=task_scratch / "opencode.json",
            config_dir=config_dir,
            local=local,
        )

    def _verify_workspace_capability(self, values: dict[str, object]) -> None:
        """校验 selector capability 的签名、受众、期限和 attempt 绑定。"""
        selector = values.get("workspace_selector")
        capability = values.get("workspace_capability")
        if selector is None:
            if capability is not None:
                raise ValueError("opencode_workspace_mismatch")
            return
        if selector == "local" and capability is None:
            # 显式 local 仍兼容旧单用户调用；非 local selector 永远需要签名材料。
            return
        if not isinstance(capability, str) or not capability:
            raise ValueError("opencode_workspace_capability_unavailable")
        if not self.capability_signer.configured:
            raise ValueError("opencode_workspace_capability_unavailable")
        try:
            self.capability_signer.verify(
                capability,
                task_id=str(values["business_task_id"]),
                attempt_id=str(values["executor_attempt_id"]),
                selector=str(selector),
                session_id=str(values["session_id"]) if isinstance(values.get("session_id"), str) else None,
                resume_of_attempt_id=str(values["resume_of_attempt_id"]) if isinstance(values.get("resume_of_attempt_id"), str) else None,
            )
        except WorkspaceCapabilityError as exc:
            raise ValueError(exc.code) from exc

    def _verify_task_workspace_capability(self, task: TaskState) -> None:
        """在排队任务真正启动前再次验证 capability，避免长队列越过有效期。"""
        if task.workspace_selector == "local":
            return
        if not task.workspace_capability or not self.capability_signer.configured:
            raise RuntimeError("opencode_workspace_capability_unavailable")
        try:
            self.capability_signer.verify(
                task.workspace_capability,
                task_id=task.business_task_id,
                attempt_id=task.executor_attempt_id,
                selector=task.workspace_selector,
                session_id=task.session_id,
                resume_of_attempt_id=task.resume_of_attempt_id,
            )
        except WorkspaceCapabilityError as exc:
            raise RuntimeError(exc.code) from exc

    def _prune_history_locked(self) -> None:
        """限制内存中的终态历史，避免长期运行的 executor 被任务标识耗尽内存。"""
        overflow = len(self.tasks) - TASK_HISTORY_LIMIT
        if overflow <= 0:
            return
        terminal = sorted(
            (task for task in self.tasks.values() if task.status in {"succeeded", "failed", "cancelled"}),
            key=lambda task: (task.completed_at or task.created_at, task.created_at, task.task_id),
        )
        for task in terminal[:overflow]:
            self.tasks.pop(task.executor_attempt_id, None)
            self.futures.pop(task.executor_attempt_id, None)

    def _attempt_metadata_path(self, business_task_id: str) -> Path:
        """返回任务专属 attempt 元数据路径，并验证业务标识可安全拼接。"""
        if not TASK_ID_RE.fullmatch(business_task_id):
            raise ValueError("agent_result_path_invalid")
        root_info = RESULT_ROOT.lstat()
        if stat.S_ISLNK(root_info.st_mode) or not stat.S_ISDIR(root_info.st_mode):
            raise ValueError("agent_result_path_invalid")
        result_dir = RESULT_ROOT / business_task_id
        if result_dir.is_symlink() or (result_dir.exists() and not result_dir.is_dir()):
            raise ValueError("agent_result_path_invalid")
        return result_dir / ATTEMPT_METADATA_NAME

    def _metadata_signature(self, attempts: list[dict[str, object]]) -> str:
        """使用 executor token 对可恢复 attempt 事实做完整性签名。"""
        raw = json.dumps(attempts, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
        return hmac.new(self.token.encode("utf-8"), raw, hashlib.sha256).hexdigest()

    def _load_attempt_metadata(self, business_task_id: str) -> list[dict[str, object]]:
        """读取并验证 runtime 中持久化的 attempt 绑定，损坏时返回空集合。"""
        try:
            path = self._attempt_metadata_path(business_task_id)
            metadata = path.lstat()
            if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode) or metadata.st_size > 256 * 1024:
                return []
            descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
            try:
                raw = os.read(descriptor, 256 * 1024 + 1)
            finally:
                os.close(descriptor)
            document = json.loads(raw.decode("utf-8"))
            attempts = document.get("attempts") if isinstance(document, dict) else None
            signature = document.get("signature") if isinstance(document, dict) else None
            if not isinstance(attempts, list) or not isinstance(signature, str):
                return []
            values = [item for item in attempts if isinstance(item, dict)]
            if len(values) != len(attempts) or not hmac.compare_digest(signature, self._metadata_signature(values)):
                return []
            return values[-32:]
        except (OSError, UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError, RecursionError):
            return []

    def _persist_attempt_metadata(self, task: TaskState) -> None:
        """原子保存终态 attempt 的最小绑定事实，供 executor 重启后恢复。"""
        # 同一业务任务的两个 attempt 可能几乎同时收束；锁住读-改-写序列，
        # 避免后完成的 attempt 覆盖前一个失败事实，导致重启后无法续跑。
        with self.lock:
            self._persist_attempt_metadata_locked(task)

    def _persist_attempt_metadata_locked(self, task: TaskState) -> None:
        """在 executor 锁内原子保存 attempt 元数据，调用者负责持有锁。"""
        path = self._attempt_metadata_path(task.business_task_id)
        path.parent.mkdir(parents=True, exist_ok=True)
        existing = self._load_attempt_metadata(task.business_task_id)
        entry: dict[str, object] = {
            "business_task_id": task.business_task_id,
            "executor_attempt_id": task.executor_attempt_id,
            "image_relative_path": task.image_relative_path,
            "reverse_image_policy": task.reverse_image_policy,
            "processing_config_hash": task.processing_config_hash,
            "visual_snapshot_sha256": task.visual_snapshot_sha256,
            "workspace_selector": task.workspace_selector,
            "session_id": task.session_id,
            "resume_of_attempt_id": task.resume_of_attempt_id,
            "status": task.status,
            "error": task.error,
            "created_at": task.created_at,
            "completed_at": task.completed_at,
            "process_reaped": task.process_reaped,
            "analysis_policy": task.analysis_policy,
            "observed_cost": task.observed_cost,
            "usage_checked_at": task.usage_checked_at,
            "reminder_sent": task.reminder_sent,
            "database_maintenance": task.database_maintenance,
        }
        values = [item for item in existing if item.get("executor_attempt_id") != task.executor_attempt_id]
        values.append(entry)
        document = {"attempts": values[-32:], "signature": self._metadata_signature(values[-32:])}
        raw = _safe_json(document)
        descriptor, temporary_name = tempfile.mkstemp(prefix=".executor-attempts.", dir=path.parent)
        try:
            with os.fdopen(descriptor, "wb") as handle:
                handle.write(raw)
            os.chmod(temporary_name, stat.S_IRUSR | stat.S_IWUSR)
            os.replace(temporary_name, path)
        except Exception:
            try:
                os.unlink(temporary_name)
            except OSError:
                pass
            raise

    def _persisted_resume_source(self, values: dict[str, object]) -> TaskState | None:
        """从签名元数据恢复失败 attempt，供 executor 服务重启后续跑。"""
        business_task_id = str(values["business_task_id"])
        session_id = values.get("session_id")
        resume_of = values.get("resume_of_attempt_id")
        if not isinstance(session_id, str):
            return None
        for entry in reversed(self._load_attempt_metadata(business_task_id)):
            if (
                entry.get("business_task_id") != business_task_id
                or entry.get("session_id") != session_id
                or entry.get("image_relative_path") != values.get("image_relative_path")
                or entry.get("reverse_image_policy") != values.get("reverse_image_policy")
                or entry.get("processing_config_hash") != values.get("processing_config_hash")
                or entry.get("visual_snapshot_sha256") != values.get("visual_snapshot_sha256")
                or entry.get("workspace_selector", "local") != (values.get("workspace_selector") or "local")
                or (resume_of is not None and entry.get("executor_attempt_id") != resume_of)
            ):
                continue
            if entry.get("status") != "failed" or entry.get("process_reaped") is not True or not isinstance(entry.get("executor_attempt_id"), str):
                continue
            error = entry.get("error") if isinstance(entry.get("error"), dict) else {}
            if error.get("error") not in {
                "service_shutdown",
                "agent_provider_rate_limited",
                "agent_provider_server_error",
                "agent_connection_interrupted",
                "agent_process_failed",
            }:
                continue
            return TaskState(
                task_id=business_task_id,
                business_task_id=business_task_id,
                executor_attempt_id=str(entry["executor_attempt_id"]),
                image_relative_path=str(entry["image_relative_path"]),
                reverse_image_policy=str(entry["reverse_image_policy"]),
                timeout_seconds=int(values["timeout_seconds"]),
                processing_config_hash=values.get("processing_config_hash") if isinstance(values.get("processing_config_hash"), str) else None,
                visual_snapshot_sha256=values.get("visual_snapshot_sha256") if isinstance(values.get("visual_snapshot_sha256"), str) else None,
                session_id=session_id,
                workspace_selector=str(values.get("workspace_selector") or "local"),
                status="failed",
                error={str(key): value for key, value in error.items()},
                created_at=float(entry.get("created_at")) if isinstance(entry.get("created_at"), (int, float)) else time.time(),
                completed_at=float(entry.get("completed_at")) if isinstance(entry.get("completed_at"), (int, float)) else time.time(),
                process_reaped=True,
                analysis_policy=entry.get("analysis_policy"),
                observed_cost=entry.get("observed_cost"),
                usage_checked_at=entry.get("usage_checked_at"),
            )
        return None

    def _validate_request(self, payload: object) -> dict[str, object]:
        """校验固定任务字段并返回业务任务与 attempt 的规范化参数。"""
        if not isinstance(payload, dict):
            raise ValueError("invalid_task")
        unknown = set(payload) - ALLOWED_REQUEST_FIELDS
        if unknown:
            raise ValueError("invalid_task")
        task_id = payload.get("task_id")
        if not isinstance(task_id, str) or not TASK_ID_RE.fullmatch(task_id):
            raise ValueError("agent_result_path_invalid")
        business_task_id = payload.get("business_task_id", task_id)
        if not isinstance(business_task_id, str) or not TASK_ID_RE.fullmatch(business_task_id) or business_task_id != task_id:
            raise ValueError("invalid_task")
        raw_attempt_id = payload.get("executor_attempt_id")
        # 旧客户端没有 attempt 字段时使用业务 ID 作为兼容别名；新客户端总是
        # 发送独立 attempt，因而终态历史不会被恢复请求复用。
        executor_attempt_id = raw_attempt_id if raw_attempt_id is not None else business_task_id
        if not isinstance(executor_attempt_id, str) or not TASK_ID_RE.fullmatch(executor_attempt_id):
            raise ValueError("invalid_task")
        raw_resume_of = payload.get("resume_of_attempt_id")
        if raw_resume_of is not None and (
            not isinstance(raw_resume_of, str) or not TASK_ID_RE.fullmatch(raw_resume_of)
        ):
            raise ValueError("invalid_task")
        session_id = payload.get("session_id")
        if session_id is not None and (not isinstance(session_id, str) or not SESSION_ID_RE.fullmatch(session_id)):
            raise ValueError("session_binding_mismatch")
        if raw_resume_of is not None and session_id is None:
            raise ValueError("session_binding_mismatch")
        processing_config_hash = payload.get("processing_config_hash")
        if processing_config_hash is not None and (
            not isinstance(processing_config_hash, str) or not CONFIG_HASH_RE.fullmatch(processing_config_hash)
        ):
            raise ValueError("invalid_task")
        visual_snapshot_sha256 = payload.get("visual_snapshot_sha256")
        if visual_snapshot_sha256 is not None and (
            not isinstance(visual_snapshot_sha256, str) or not CONFIG_HASH_RE.fullmatch(visual_snapshot_sha256)
        ):
            raise ValueError("visual_match_snapshot_invalid")
        relative = _relative_image_path(payload.get("image_relative_path"))
        raw_selector = payload.get("workspace_selector")
        if raw_selector is None or raw_selector == "local":
            image = IMAGE_ROOT / relative
            try:
                current = IMAGE_ROOT
                for part in relative.parts:
                    current = current / part
                    if current.is_symlink():
                        raise ValueError("agent_image_path_forbidden")
                root = IMAGE_ROOT.resolve()
                candidate = image.resolve(strict=True)
                candidate.relative_to(root)
                if not candidate.is_file():
                    raise ValueError("agent_image_path_forbidden")
            except (OSError, RuntimeError, ValueError) as exc:
                if isinstance(exc, ValueError) and str(exc) == "agent_image_path_forbidden":
                    raise
                raise ValueError("agent_image_path_forbidden") from exc
        policy = payload.get("reverse_image_policy", "forbid")
        if policy not in {"forbid", "auto"}:
            raise ValueError("invalid_reverse_image_policy")
        raw_timeout = payload.get("timeout_seconds", self.max_timeout)
        if isinstance(raw_timeout, bool):
            raise ValueError("invalid_task")
        try:
            timeout = int(raw_timeout)
        except (TypeError, ValueError) as exc:
            raise ValueError("invalid_task") from exc
        if timeout < 1 or timeout > self.max_timeout:
            raise ValueError("agent_timeout_limit_exceeded")
        wait = payload.get("wait", True)
        if not isinstance(wait, bool):
            raise ValueError("invalid_task")
        callback_token = payload.get("callback_token")
        if callback_token is not None and (
            not isinstance(callback_token, str)
            or len(callback_token) > 4096
            or any(character.isspace() or ord(character) < 0x20 for character in callback_token)
        ):
            raise ValueError("invalid_task")
        workspace_selector = raw_selector
        if workspace_selector is not None and (not isinstance(workspace_selector, str) or not SELECTOR_RE.fullmatch(workspace_selector)):
            raise ValueError("opencode_workspace_invalid")
        workspace_capability = payload.get("workspace_capability")
        if workspace_capability is not None and (not isinstance(workspace_capability, str) or not workspace_capability or len(workspace_capability) > 4096):
            raise ValueError("opencode_workspace_capability_invalid")
        model_capability = payload.get(MODEL_CAPABILITY_FIELD)
        if model_capability is not None:
            try:
                model_capability = validate_model_capability(model_capability)
            except ModelCapabilityError as exc:
                raise ValueError(str(exc)) from exc
        elif self.release_profile in {"production", "public", "1", "true", "yes", "on"} or self.model_broker_configured:
            raise ValueError("model_capability_unavailable")
        values = {
            "task_id": business_task_id,
            "business_task_id": business_task_id,
            "executor_attempt_id": executor_attempt_id,
            "resume_of_attempt_id": raw_resume_of,
            "session_id": session_id,
            "processing_config_hash": processing_config_hash.lower() if isinstance(processing_config_hash, str) else None,
            "visual_snapshot_sha256": visual_snapshot_sha256.lower() if isinstance(visual_snapshot_sha256, str) else None,
            "image_relative_path": relative.as_posix(),
            "reverse_image_policy": str(policy),
            "timeout_seconds": timeout,
            "wait": wait,
            "callback_token": callback_token,
            "workspace_selector": workspace_selector,
            "workspace_capability": workspace_capability,
            MODEL_CAPABILITY_FIELD: model_capability,
            "trace_context": {key: value for key, value in (payload.get("trace_context") or {}).items() if key in {"traceparent", "tracestate"} and isinstance(value, str) and len(value) <= 512} if isinstance(payload.get("trace_context"), dict) else {},
            "trace_attributes": {key: value for key, value in (payload.get("trace_attributes") or {}).items() if key in {"account.id", "scope.id"} and isinstance(value, str) and len(value) <= 128} if isinstance(payload.get("trace_attributes"), dict) else {},
        }
        self._verify_workspace_capability(values)
        policy = payload.get("analysis_policy")
        if policy is not None:
            try:
                parsed = parse_analysis_policy(policy)
            except AnalysisPolicyError as exc:
                raise ValueError(exc.code) from exc
            if parsed.model != self.model or parsed.variant != "max":
                raise ValueError("agent_analysis_policy_invalid")
            if self.release_profile in {"production", "public", "1", "true", "yes", "on"} and not self.model_broker_configured:
                raise ValueError("model_broker_endpoint_invalid")
            if model_capability is not None and not self.model_broker_configured:
                raise ValueError("model_broker_endpoint_invalid")
            values["analysis_policy"] = parsed.model_dump(mode="json")
        layout = self._workspace_layout(selector=workspace_selector if isinstance(workspace_selector, str) else None, business_task_id=business_task_id)
        image = layout.images_root / relative
        try:
            current = layout.images_root
            for part in relative.parts:
                current = current / part
                info = current.lstat()
                if stat.S_ISLNK(info.st_mode):
                    raise ValueError("agent_image_path_forbidden")
            info = image.lstat()
            if not stat.S_ISREG(info.st_mode):
                raise ValueError("agent_image_path_forbidden")
        except (OSError, ValueError) as exc:
            if isinstance(exc, ValueError) and str(exc) == "agent_image_path_forbidden":
                raise
            raise ValueError("agent_image_path_forbidden") from exc
        values["workspace_layout"] = layout
        return values

    def _resume_source(self, values: dict[str, object]) -> TaskState | None:
        """查找与续跑 session 完全绑定的失败 attempt，拒绝跨任务猜测。"""
        session_id = values.get("session_id")
        if not isinstance(session_id, str):
            return None
        business_task_id = values["business_task_id"]
        resume_of = values.get("resume_of_attempt_id")
        matches = [
            task
            for task in self.tasks.values()
            if task.session_id == session_id
            and task.business_task_id == business_task_id
            and task.image_relative_path == values["image_relative_path"]
            and task.reverse_image_policy == values["reverse_image_policy"]
            and task.processing_config_hash == values.get("processing_config_hash")
            and task.visual_snapshot_sha256 == values.get("visual_snapshot_sha256")
            and task.workspace_selector == (values.get("workspace_selector") or "local")
            and (resume_of is None or task.executor_attempt_id == resume_of)
        ]
        if not matches:
            persisted = self._persisted_resume_source(values)
            if persisted is not None:
                return persisted
            # 同一 session 若属于其它业务任务，明确返回绑定错误；完全未知的
            # session 则返回不可续跑，避免把最近会话当作恢复目标。
            if any(task.session_id == session_id for task in self.tasks.values()):
                raise RuntimeError("session_binding_mismatch")
            raise RuntimeError("session_not_resumable")
        source = max(matches, key=lambda task: (task.completed_at or task.created_at, task.created_at))
        if source.status != "failed":
            raise RuntimeError("session_not_resumable")
        if source.process_reaped is not True or not source.done.is_set():
            raise RuntimeError("session_not_resumable")
        source_error = (source.error or {}).get("error")
        if source_error not in {
            "service_shutdown",
            "agent_provider_rate_limited",
            "agent_provider_server_error",
            "agent_connection_interrupted",
            "agent_process_failed",
        }:
            raise RuntimeError("session_not_resumable")
        return source

    def submit(self, payload: object) -> tuple[TaskState, bool]:
        """创建固定研究任务并交给受限线程池，返回状态和同步等待标记。"""
        values = self._validate_request(payload)
        with self.lock:
            self._prune_history_locked()
            if not self.health().get("ready"):
                raise RuntimeError("agent_runtime_unavailable")
            if not self.model_configured:
                raise RuntimeError("opencode_not_configured")
            existing = self.tasks.get(values["executor_attempt_id"])
            if existing is not None:
                raise RuntimeError("task_exists")
            same_business = [
                task
                for task in self.tasks.values()
                if task.business_task_id == values["business_task_id"]
                and task.executor_attempt_id != values["executor_attempt_id"]
                and (task.status in {"queued", "running"} or task.process_reaped is not True)
            ]
            if same_business:
                # 同一业务 Task 只能有一个活动 attempt；未确认回收的旧进程更严格地
                # 收束为 unknown_execution，避免其晚到结果污染新 attempt。
                if not any(task.status in {"queued", "running"} for task in same_business) and any(task.process_reaped is not True for task in same_business):
                    raise RuntimeError("unknown_execution")
                raise RuntimeError("task_exists")
            source = self._resume_source(values) if values.get("session_id") else None
            if source is not None and source.analysis_policy != values.get("analysis_policy"):
                raise RuntimeError("agent_analysis_policy_invalid")
            task = TaskState(
                task_id=str(values["business_task_id"]),
                business_task_id=str(values["business_task_id"]),
                executor_attempt_id=str(values["executor_attempt_id"]),
                image_relative_path=str(values["image_relative_path"]),
                reverse_image_policy=str(values["reverse_image_policy"]),
                timeout_seconds=int(values["timeout_seconds"]),
                processing_config_hash=values.get("processing_config_hash") if isinstance(values.get("processing_config_hash"), str) else None,
                visual_snapshot_sha256=values.get("visual_snapshot_sha256") if isinstance(values.get("visual_snapshot_sha256"), str) else None,
                session_id=values.get("session_id") if isinstance(values.get("session_id"), str) else None,
                workspace_selector=str(values.get("workspace_selector") or "local"),
                workspace_capability=values.get("workspace_capability") if isinstance(values.get("workspace_capability"), str) else None,
                model_capability=values.get(MODEL_CAPABILITY_FIELD) if isinstance(values.get(MODEL_CAPABILITY_FIELD), str) else None,
                workspace_directory=(values.get("workspace_layout").directory if isinstance(values.get("workspace_layout"), WorkspaceLayout) else None),
                images_root=(values.get("workspace_layout").images_root if isinstance(values.get("workspace_layout"), WorkspaceLayout) else None),
                metadata_root=(values.get("workspace_layout").metadata_root if isinstance(values.get("workspace_layout"), WorkspaceLayout) else None),
                skill_root=(values.get("workspace_layout").skill_root if isinstance(values.get("workspace_layout"), WorkspaceLayout) else None),
                candidate_root=(values.get("workspace_layout").candidate_root if isinstance(values.get("workspace_layout"), WorkspaceLayout) else None),
                task_scratch_root=(values.get("workspace_layout").task_scratch_root if isinstance(values.get("workspace_layout"), WorkspaceLayout) else None),
                config_file=(values.get("workspace_layout").config_file if isinstance(values.get("workspace_layout"), WorkspaceLayout) else None),
                config_dir=(values.get("workspace_layout").config_dir if isinstance(values.get("workspace_layout"), WorkspaceLayout) else None),
                resume_of_attempt_id=source.executor_attempt_id if source else None,
                is_resume=source is not None,
                callback_token=values.get("callback_token") if isinstance(values.get("callback_token"), str) else None,
                result_path=f"task-results/{values['business_task_id']}/{RESULT_FILE_NAME}",
                analysis_policy=values.get("analysis_policy"),
                observed_cost=source.observed_cost if source else None,
                trace_context=values["trace_context"],
                trace_attributes=values["trace_attributes"],
            )
            self.tasks[task.executor_attempt_id] = task
            self.futures[task.executor_attempt_id] = self.pool.submit(self._run, task)
            return task, bool(values["wait"])

    @staticmethod
    def _workspace_permission_rules(task: TaskState) -> dict[str, str]:
        """生成 OpenCode 外部目录的最后匹配优先规则。"""
        images = task.images_root or IMAGE_ROOT
        metadata = task.metadata_root or IMAGE_ROOT
        skills = task.skill_root or SKILL_ROOT
        scratch = task.task_scratch_root or (WORKSPACE / "tasks" / task.business_task_id)
        candidate_root = task.candidate_root or (RUNTIME_ROOT / "candidates" / task.business_task_id)
        result_dir = RESULT_ROOT / task.business_task_id
        subtree = lambda path: f"{Path(os.path.abspath(path)).as_posix()}/**"
        return {
            "*": "deny",
            subtree(skills): "allow",
            subtree(images): "allow",
            subtree(metadata): "allow",
            subtree(candidate_root): "allow",
            subtree(scratch): "allow",
            # 外部文件工具先检查结果目录的 ``<dir>/*`` 父级模式；写入范围
            # 由同一配置中的 ``edit`` 规则收窄到两个固定结果文件。
            str(result_dir.absolute()) + "/*": "allow",
            str((result_dir / RESULT_DRAFT_NAME).absolute()): "allow",
            str((result_dir / RESULT_FILE_NAME).absolute()): "allow",
        }

    @staticmethod
    def _workspace_edit_permission_rules(task: TaskState) -> dict[str, str]:
        """生成只允许 Task 临时数据和两个结果文件的编辑规则。"""
        scratch = task.task_scratch_root or (WORKSPACE / "tasks" / task.business_task_id)
        candidate_root = task.candidate_root or (RUNTIME_ROOT / "candidates" / task.business_task_id)
        config_file = task.config_file or scratch / "opencode.json"
        config_dir = task.config_dir or scratch / ".opencode"
        result_dir = RESULT_ROOT / task.business_task_id
        return dict(
            build_edit_permission_rules(
                task_scratch_root=scratch,
                config_file=config_file,
                config_dir=config_dir,
                draft_path=result_dir / RESULT_DRAFT_NAME,
                result_path=result_dir / RESULT_FILE_NAME,
                candidate_root=candidate_root,
            )
        )

    def _prepare_task_workspace(self, task: TaskState) -> None:
        """创建当前 Task 临时目录并原子写入不含密钥的 workspace 配置。"""
        scratch = task.task_scratch_root or (WORKSPACE / "tasks" / task.business_task_id)
        config_file = task.config_file or scratch / "opencode.json"
        config_dir = task.config_dir or scratch / ".opencode"
        self._checked_directory(scratch.parent, create=True)
        self._checked_directory(scratch, create=True)
        self._checked_directory(scratch / "home", create=True)
        self._checked_directory(config_dir, create=True)
        try:
            validate_file_path(config_file, allow_missing=True, code="opencode_workspace_invalid")
        except Exception as exc:  # noqa: BLE001 - 配置路径是 executor 安全边界
            raise RuntimeError("opencode_workspace_invalid") from exc
        config = json.loads(json.dumps(RUNTIME_OPENCODE_CONFIG, ensure_ascii=False))
        if task.analysis_policy is not None:
            config["plugin"] = [["file:///opt/mememeow/plugins/analysis-reminder.mjs", {
                "attempt_id": task.executor_attempt_id,
                "policy": task.analysis_policy,
                "session_id": task.session_id,
                "socket_path": str(ANALYSIS_SOCKET_ROOT / f"{task.executor_attempt_id}.sock"),
            }]]
        config["permission"] = {
            "external_directory": self._workspace_permission_rules(task),
            "edit": self._workspace_edit_permission_rules(task),
        }
        content = json.dumps(config, ensure_ascii=False, indent=2) + "\n"
        temporary = config_file.with_name(f".{config_file.name}.tmp.{os.getpid()}.{id(task)}")
        try:
            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), stat.S_IRUSR | stat.S_IWUSR)
            try:
                with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
                    descriptor = -1
                    handle.write(content)
            finally:
                if descriptor >= 0:
                    os.close(descriptor)
        except (OSError, ValueError) as exc:
            raise RuntimeError("opencode_workspace_invalid") from exc
        try:
            os.replace(temporary, config_file)
        except OSError as exc:
            raise RuntimeError("opencode_workspace_invalid") from exc

    def _validate_candidate_view(self, task: TaskState) -> None:
        """在启动 OpenCode 前复核候选根、manifest 和只读权限。"""
        candidate_root = task.candidate_root or (RUNTIME_ROOT / "candidates" / task.business_task_id)
        expected_sha256 = task.visual_snapshot_sha256
        if not expected_sha256:
            if candidate_root.exists() or candidate_root.is_symlink():
                self._checked_directory(candidate_root, missing_code="visual_candidate_materialization_failed")
            return
        self._checked_directory(candidate_root, missing_code="visual_candidate_materialization_failed")
        manifest = candidate_root / "manifest.json"
        try:
            root_info = candidate_root.lstat()
            if root_info.st_mode & (stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH):
                raise ValueError("visual_candidate_materialization_failed")
            validate_file_path(manifest, code="visual_candidate_materialization_failed")
            info = manifest.lstat()
            if info.st_size > 4 * 1024 * 1024:
                raise ValueError("visual_candidate_materialization_failed")
            descriptor = os.open(manifest, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
            try:
                raw = os.read(descriptor, 4 * 1024 * 1024 + 1)
            finally:
                os.close(descriptor)
            if len(raw) > 4 * 1024 * 1024:
                raise ValueError("visual_candidate_materialization_failed")
            document = json.loads(raw.decode("utf-8"))
            validated = validate_visual_match_snapshot_manifest(document, expected_sha256=expected_sha256)
            candidates = validated.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, dict):
                    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 candidate_root.rglob("*"):
                node = item.lstat()
                relative = item.relative_to(candidate_root).as_posix()
                if stat.S_ISLNK(node.st_mode) or node.st_mode & (stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH):
                    raise ValueError("visual_candidate_materialization_failed")
                if stat.S_ISDIR(node.st_mode):
                    if relative not in expected_directories:
                        raise ValueError("visual_candidate_materialization_failed")
                elif stat.S_ISREG(node.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 candidate_root.rglob("*") if item.is_file())
            if total_bytes > MAX_CANDIDATE_ROOT_BYTES:
                raise ValueError("visual_candidate_materialization_failed")
            for candidate in candidates:
                if not isinstance(candidate, dict):
                    raise ValueError("visual_candidate_materialization_failed")
                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 = candidate_root.joinpath(*PurePosixPath(relative).parts)
                validate_file_path(candidate_path, code="visual_candidate_materialization_failed")
                candidate_info = candidate_path.lstat()
                if candidate_info.st_size != size_bytes:
                    raise ValueError("visual_candidate_materialization_failed")
                digest = hashlib.sha256()
                candidate_descriptor = os.open(candidate_path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
                try:
                    while True:
                        chunk = os.read(candidate_descriptor, 1024 * 1024)
                        if not chunk:
                            break
                        digest.update(chunk)
                finally:
                    os.close(candidate_descriptor)
                if digest.hexdigest() != image_sha256.lower():
                    raise ValueError("visual_candidate_materialization_failed")
        except (OSError, UnicodeDecodeError, json.JSONDecodeError, RecursionError, TypeError, ValueError, VisualMatchSnapshotError) as exc:
            raise RuntimeError("visual_candidate_materialization_failed") from exc

    def _task_environment(self, task: TaskState) -> dict[str, str]:
        """构造 OpenCode 最小环境白名单，不继承 executor 容器中的无关变量。"""
        scratch = task.task_scratch_root or (WORKSPACE / "tasks" / task.business_task_id)
        values = {
            "PATH": os.getenv("PATH", "/usr/local/bin:/usr/bin:/bin"),
            # HOME 使用当前 Task 的独立目录，避免多个 Agent 共享可写配置和缓存。
            "HOME": str(scratch / "home"),
            # 提醒和终止共同读取任务专属数据库中的主 session 金额。
            "OPENCODE_DB": str(scratch / "opencode.db" if task.analysis_policy is not None else RUNTIME_ROOT / "opencode.db"),
            "OPENCODE_CONFIG": str(task.config_file or scratch / "opencode.json"),
            "OPENCODE_CONFIG_DIR": str(task.config_dir or scratch / ".opencode"),
            "OPENCODE_DISABLE_PROJECT_CONFIG": "1",
            MODEL_BROKER_URL_ENV: self.model_broker_url or self.legacy_base_url,
            # 生产值来自本次 task 的短期 capability；local 旧夹具才使用旧 key。
            MODEL_CAPABILITY_ENV: task.model_capability or self.legacy_api_key,
            "MEMEMEOW_OPENCODE_SLOT": "0",
            "MEMEMEOW_AGENT_TASK_ID": task.business_task_id,
            "MEMEMEOW_AGENT_EXECUTOR_ATTEMPT_ID": task.executor_attempt_id,
            "MEMEMEOW_AGENT_TRACEPARENT": trace_carrier().get("traceparent", ""),
            **({"MEMEMEOW_AGENT_CALLBACK_TOKEN": task.callback_token} if task.callback_token else {}),
            "MEMEMEOW_REVERSE_IMAGE_INTERNAL_URL": os.getenv("MEMEMEOW_AGENT_REVERSE_IMAGE_INTERNAL_URL", ""),
            "MEMEMEOW_AGENT_CANDIDATE_MANIFEST": str((task.candidate_root or (RUNTIME_ROOT / "candidates" / task.business_task_id)) / "manifest.json"),
            "MEMEMEOW_DATA_ROOT": str(RUNTIME_ROOT),
            "MEMEMEOW_REVERSE_IMAGE_CACHE_ROOT": str(RUNTIME_ROOT / "reverse_image_cache" / "serpapi_google_lens"),
            # 依赖位于镜像只读目录，避免在 runtime volume 内创建 node_modules 链接。
            "NODE_PATH": "/opt/mememeow/node_modules",
        }
        return values

    @staticmethod
    def _process_directory(task: TaskState) -> Path:
        """返回 OpenCode 当前任务的工作目录；分析策略任务固定使用专属 scratch。"""
        if task.analysis_policy is not None:
            return task.task_scratch_root or (WORKSPACE / "tasks" / task.business_task_id)
        return task.workspace_directory or WORKSPACE

    def _sandbox_command(self, task: TaskState, command: list[str]) -> list[str]:
        """返回已经完成路径校验的 OpenCode 命令，不创建 Linux namespace。

        图片分析任务继续使用独立 scratch、任务数据库、候选 manifest 和 OpenCode
        文件权限规则；容器边界由 Compose 提供。保留该方法作为统一执行入口，避免
        调用方重新组装已经校验的命令。
        """
        del task
        return command

    def _prompt(self, task: TaskState) -> str:
        """生成唯一业务 prompt；调用方不能覆盖路径、命令或提示词。"""
        result_dir = RESULT_ROOT / task.task_id
        image = (task.images_root or IMAGE_ROOT) / _relative_image_path(task.image_relative_path)
        resume_instruction = (
            "这是同一 OpenCode session 的受控续跑；保留并检查已有草稿和中间产物，只完成尚未完成的工作。"
            if task.is_resume
            else "这是首次执行；先建立任务草稿并逐步完成研究。"
        )
        return (
            "使用 research-meme-context skill 分析这张表情包；先读取本任务固定候选 manifest，"
            "候选只作为参考证据；只通过项目内部接口使用可选反向图片能力。"
            f"原始图片路径（只读）：{json.dumps(str(image.absolute()), ensure_ascii=False)}。"
            "必须直接通过视觉模型查看输入图片并识别图片文字，禁止调用任何本地文字识别程序、命令或脚本。"
            f"本任务 reverse_image_policy={task.reverse_image_policy}。"
            f"{resume_instruction}"
            f"结果必须写入 {result_dir / RESULT_FILE_NAME}；先写 {result_dir / RESULT_DRAFT_NAME}，"
            "使用 output-schema.json 校验后通过同一文件系统的原子 rename/mv 替换最终文件。"
            "不要把业务 JSON 作为 assistant 文本交付，不要写入数据库。"
        )

    def _run(self, task: TaskState) -> None:
        """执行 attempt，并将线程入口异常保存为明确终态及诊断后继续抛出。"""
        started_at = time.monotonic()
        try:
            self._run_with_telemetry(task)
        except Exception as error:
            reason = _redact_diagnostic(f"{type(error).__name__}: {error}", secret_inventory_from_mapping(os.environ) + (self.token, task.callback_token or "", task.workspace_capability or "", task.model_capability or ""))
            logger.error("agent_executor_internal_error: task={} attempt={} exception={} reason={}", task.business_task_id, task.executor_attempt_id, type(error).__name__, reason)
            with self.lock:
                if not task.done.is_set():
                    task.status = "failed"
                    task.error = _json_error("agent_executor_internal_error", reason)
                    task.completed_at = time.time()
                    try:
                        self._persist_attempt_metadata_locked(task)
                    finally:
                        task.done.set()
            raise
        finally:
            if TelemetryConfig.from_env("mememeow-executor").enabled:
                meter = metrics.get_meter("mememeow.executor")
                meter.create_counter("agent_attempts", unit="{attempt}").add(1, {"result": task.status})
                meter.create_histogram("agent_attempt_duration", unit="s").record(time.monotonic() - started_at)

    def _run_with_telemetry(self, task: TaskState) -> None:
        """绑定当前 attempt 的观测上下文，业务结果由既有执行流程提供。"""
        if not TelemetryConfig.from_env("mememeow-executor").enabled:
            self._run_observed_task(task)
            return
        attributes = {**task.trace_attributes, "task.id": task.business_task_id, "attempt.id": task.executor_attempt_id, "workspace.selector": task.workspace_selector, "business.image_relative_path": task.image_relative_path}
        started_at = time.time()
        parent = extract_carrier(task.trace_context)
        links = [trace.Link(trace.get_current_span(parent).get_span_context())] if trace.get_current_span(parent).get_span_context().is_valid else []
        with trace.get_tracer("mememeow.executor").start_as_current_span("agent.attempt", context=extract_carrier({}), links=links, attributes=attributes) as span:
            observer = SQLiteTelemetry(database=Path(self._task_environment(task)["OPENCODE_DB"]), directory=self._process_directory(task), session=lambda: task.session_id, attributes=attributes, context=trace.set_span_in_context(span), started_at=started_at, secrets=(self.token, self.legacy_api_key, task.callback_token or "", task.workspace_capability or "", task.model_capability or ""))
            observer.start()
            try:
                self._run_observed_task(task)
            finally:
                observer.close()
                span.set_attribute("attempt.result", task.status)
                span.set_attribute("attempt.reason", protected_content(str(task.error or {}), observer.secrets))

    def _run_observed_task(self, task: TaskState) -> None:
        """执行 OpenCode 任务并更新状态，失败时保存脱敏后的有限诊断。"""
        with self.lock:
            if task.done.is_set():
                return
        process: subprocess.Popen[bytes] | None = None
        stdout = b""
        stderr = b""
        timed_out = False
        analysis_socket: socket.socket | None = None
        output = AttemptOutput()
        monitor = None
        database: Path | None = None
        final_status = "failed"
        final_error: dict[str, object] | None = None
        started = time.monotonic()
        diagnostic_secrets = secret_inventory_from_mapping(os.environ) + (
            self.legacy_api_key, self.token, task.callback_token or "",
            task.workspace_capability or "", task.model_capability or "",
            task.session_id or "",
        )
        try:
            with self.lock:
                if task.cancel_event.is_set():
                    final_status = "cancelled"
                    final_error = _json_error("task_interrupted", "任务已取消")
                    return
                self._verify_task_workspace_capability(task)
                task.status = "running"
                task.started_at = time.time()
                self._validate_candidate_view(task)
                self._prepare_task_workspace(task)
                result_dir = RESULT_ROOT / task.task_id
                if result_dir.is_symlink() or (result_dir.exists() and not result_dir.is_dir()):
                    raise RuntimeError("agent_result_path_invalid")
                result_dir.mkdir(parents=True, exist_ok=True)
                # 续跑必须保留原 attempt 的草稿和已验证结果，首次 attempt 才清理
                # 任务目录中的两个受控交付文件。
                if not task.is_resume:
                    for filename in (RESULT_FILE_NAME, RESULT_DRAFT_NAME):
                        path = result_dir / filename
                        if path.exists() or path.is_symlink():
                            path.unlink()
            env = self._task_environment(task)
            diagnostic_secrets += secret_inventory_from_mapping(env)
            image_root = task.images_root or IMAGE_ROOT
            image = image_root / _relative_image_path(task.image_relative_path)
            try:
                self._checked_directory(image_root)
                current = image_root
                for part in _relative_image_path(task.image_relative_path).parts:
                    current = current / part
                    info = current.lstat()
                    if stat.S_ISLNK(info.st_mode):
                        raise ValueError("agent_image_path_forbidden")
                info = image.lstat()
                if not stat.S_ISREG(info.st_mode):
                    raise ValueError("agent_image_path_forbidden")
            except (OSError, ValueError) as exc:
                if isinstance(exc, ValueError) and str(exc) == "agent_image_path_forbidden":
                    raise
                raise ValueError("agent_image_path_forbidden") from exc
            process_directory = self._process_directory(task)
            if task.analysis_policy is not None:
                ANALYSIS_SOCKET_ROOT.mkdir(parents=True, exist_ok=True)
                socket_path = ANALYSIS_SOCKET_ROOT / f"{task.executor_attempt_id}.sock"
                if socket_path.exists() or socket_path.is_symlink():
                    socket_path.unlink()
                analysis_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
                analysis_socket.bind(str(socket_path))
                os.chmod(socket_path, stat.S_IRUSR | stat.S_IWUSR)
                analysis_socket.listen(8)
            command = [
                self.opencode_executable,
                "run",
                "--auto",
                "--dir",
                str(process_directory),
                "--format",
                "json",
                "--file",
                str(image),
                "--model",
                self.model,
                "--variant",
                "max",
                "--title",
                f"mememeow-task-{task.task_id}",
            ]
            if task.session_id:
                command.extend(("--session", task.session_id))
            command.append(self._prompt(task))
            command = self._sandbox_command(task, command)
            output.phase = "process_start"
            with output.capture(LOG_ROOT, task.task_id) as (out, err):
                database = Path(env["OPENCODE_DB"])
                self.database_maintenance.acquire(database)
                process = subprocess.Popen(command, cwd=process_directory, env=env, stdin=subprocess.DEVNULL, stdout=out, stderr=err, start_new_session=True)
                output.phase = "process_running"
                with self.lock:
                    task.process = process
                    # 进程启动后，只有父进程确认 waitpid 收束才允许该 attempt 作为续跑源。
                    task.process_reaped = False
                deadline = time.monotonic() + task.timeout_seconds
                monitor = None
                if task.analysis_policy is not None:
                    scratch = task.task_scratch_root or (WORKSPACE / "tasks" / task.business_task_id)
                    monitor = AnalysisMonitor(
                        attempt_id=task.executor_attempt_id, policy=task.analysis_policy,
                        database=Path(env["OPENCODE_DB"]), directory=process_directory,
                        status_path=scratch / f"analysis-{task.executor_attempt_id}.json",
                        # 主 session 可能在 OpenCode 完成冷启动和首个模型请求后才出现。
                        # 使用任务自身的总期限等待绑定，避免把合法的慢启动误判为用量不可用。
                        startup_deadline=deadline,
                        status_socket=analysis_socket,
                        observed_cost=Decimal(task.observed_cost or "0"),
                    )
                    monitor.expected_pid = process.pid
                while process.poll() is None:
                    if task.cancel_event.is_set():
                        reaped = self.process_supervisor.terminate(process).reaped
                        with self.lock:
                            task.process_reaped = reaped
                        if not reaped:
                            raise _ProcessFailure("unknown_execution", "无法确认 OpenCode 进程已终止")
                        break
                    if time.monotonic() >= deadline:
                        reaped = self.process_supervisor.terminate(process).reaped
                        with self.lock:
                            task.process_reaped = reaped
                        if not reaped:
                            raise _ProcessFailure("unknown_execution", "无法确认 OpenCode 进程已终止")
                        timed_out = True
                        break
                    if monitor is not None:
                        output.phase = "analysis_monitor"
                        self._check_analysis(task, monitor, process, out)
                        output.phase = "process_running"
                    time.sleep(0.05)
                if process.poll() is not None:
                    with self.lock:
                        task.process_reaped = True
                out.flush()
                err.flush()
                # session 必须从完整 stdout 流解析；采样只保留给有限错误诊断。
                output.phase = "session_capture"
                self._capture_session(task, out)
                if monitor is not None and not timed_out and not task.cancel_event.is_set():
                    output.phase = "analysis_monitor"
                    self._check_analysis(task, monitor, process, out, exited=True)
                stdout = _stream_sample(out, 256 * 1024)
                stderr = _stream_sample(err, 16 * 1024)
            if task.cancel_event.is_set():
                output.phase = "process_cancelled"
                raise RuntimeError("task_interrupted")
            if timed_out:
                output.phase = "process_timeout"
                raise _ProcessFailure("agent_timeout", "OpenCode 执行超时")
            if process.returncode != 0:
                output.phase = "process_exit"
                code, http_status = self._classify_process_failure(stdout, stderr)
                raise _ProcessFailure(
                    code,
                    _diagnostic(stdout, stderr, secrets=(self.legacy_api_key, self.token, task.callback_token or "")),
                    http_status=http_status,
                )
            if not task.session_id:
                raise RuntimeError("agent_output_invalid_json")
            output.phase = "result_validation"
            self._validate_result_file(RESULT_ROOT / task.task_id / RESULT_FILE_NAME)
            with self.lock:
                if task.cancel_event.is_set():
                    raise RuntimeError("task_interrupted")
                final_status = "succeeded"
        except _ProcessFailure as exc:
            with self.lock:
                # 未确认进程已回收时，取消/超时也不能伪装成可安全重试的普通失败。
                code = exc.code
                if task.cancel_event.is_set() and code != "unknown_execution":
                    code = "task_interrupted"
                final_status = "cancelled" if code == "task_interrupted" else "failed"
                error: dict[str, object] = {
                    "error": code,
                    "message": _redact_diagnostic(exc.args[0], (self.legacy_api_key, self.token, task.callback_token or "")),
                }
                if exc.analysis_diagnostic is not None:
                    error["analysis_diagnostic"] = exc.analysis_diagnostic
                if exc.http_status is not None:
                    error["http_status"] = exc.http_status
                final_error = error
        except RuntimeError as exc:
            code, _, detail = str(exc).partition(":")
            reason_code = getattr(exc, "reason_code", None)
            with self.lock:
                # 取消与子进程自然退出可能同时发生；持锁后以取消为最终事实，避免把
                # 用户明确取消的任务误记为普通进程失败。
                if process is not None and process.poll() is not None:
                    task.process_reaped = True
                if task.process_reaped is not True:
                    code, detail = "unknown_execution", "无法确认 OpenCode 进程已终止"
                elif task.cancel_event.is_set():
                    code, detail = "task_interrupted", ""
                if code not in _EXECUTOR_ERROR_CODES:
                    code = "agent_process_failed"
                final_status = "cancelled" if code == "task_interrupted" else "failed"
                fallback = {
                    "agent_timeout": "OpenCode 执行超时",
                    "task_interrupted": "任务已取消",
                }.get(code, "任务执行失败")
                final_error = _json_error(
                    code,
                    _redact_diagnostic(detail, (self.legacy_api_key, self.token, task.callback_token or "")) if detail else fallback,
                    reason_code=reason_code,
                )
        except (OSError, ValueError, json.JSONDecodeError) as exc:
            with self.lock:
                if process is not None and process.poll() is not None:
                    task.process_reaped = True
                code = "unknown_execution" if task.process_reaped is not True else "task_interrupted" if task.cancel_event.is_set() else "agent_process_failed"
                final_status = "cancelled" if code == "task_interrupted" else "failed"
                final_error = _json_error(code, "任务已取消" if code == "task_interrupted" else "无法确认 OpenCode 进程已终止" if code == "unknown_execution" else _redact_diagnostic(str(exc), (self.legacy_api_key, self.token, task.callback_token or "")))
        except Exception as exc:  # noqa: BLE001 - 任务必须以终态收束，不能留下永久 running
            with self.lock:
                if process is not None and process.poll() is not None:
                    task.process_reaped = True
                code = "unknown_execution" if task.process_reaped is not True else "task_interrupted" if task.cancel_event.is_set() else "agent_process_failed"
                final_status = "cancelled" if code == "task_interrupted" else "failed"
                final_error = _json_error(code, "任务已取消" if code == "task_interrupted" else "无法确认 OpenCode 进程已终止" if code == "unknown_execution" else _redact_diagnostic(str(exc), (self.legacy_api_key, self.token, task.callback_token or "")))
        finally:
            if process is not None:
                task.process_reaped = self.process_supervisor.terminate(process).reaped
                if not task.process_reaped:
                    final_status = "failed"
                    final_error = _json_error("unknown_execution", "无法确认 OpenCode 进程已终止")
            if analysis_socket is not None:
                analysis_socket.close()
                socket_path = ANALYSIS_SOCKET_ROOT / f"{task.executor_attempt_id}.sock"
                try:
                    socket_path.unlink()
                except FileNotFoundError:
                    pass
            if database is not None:
                task.database_maintenance = self.database_maintenance.release(database, process_reaped=task.process_reaped)
                diagnostic = dict(task.database_maintenance)
                if "message" in diagnostic:
                    diagnostic["message"] = _redact_diagnostic(str(diagnostic["message"]), diagnostic_secrets)
                    task.database_maintenance = diagnostic
                logger.bind(attempt_id=task.executor_attempt_id, phase="database_checkpoint", **diagnostic).log(
                    "WARNING" if diagnostic["status"] in {"failed", "skipped"} else "INFO",
                    "OpenCode 数据库收尾：attempt={}；结果={}", task.executor_attempt_id, diagnostic,
                )
            with self.lock:
                task.process = None
                if task.cancel_event.is_set() and (final_error or {}).get("error") != "unknown_execution":
                    if task.cancel_reason != "service_shutdown":
                        final_status = "cancelled"
                        final_error = _json_error("task_interrupted", "任务已取消")
                    elif (final_error or {}).get("error") == "task_interrupted":
                        final_status = "failed"
                        final_error = _json_error("service_shutdown", "服务正常关闭，Agent 执行已中断")
                task.status = final_status
                task.error = final_error
                task.completed_at = time.time()
                try:
                    self._persist_attempt_metadata(task)
                except (OSError, ValueError, TypeError) as exc:
                    task.status = "failed"
                    task.error = _json_error("agent_attempt_metadata_write_failed", "无法保存 executor attempt 恢复信息")
                    logger.error("attempt 恢复信息保存失败：attempt={} error={}", task.executor_attempt_id, type(exc).__name__)
                task.done.set()
            if task.status in {"failed", "cancelled"}:
                secrets = diagnostic_secrets + (task.session_id or "",)
                redact = lambda value: _redact_diagnostic(str(value), secrets)
                output.log_failure(
                    redact=redact, task_id=task.business_task_id, attempt_id=task.executor_attempt_id,
                    resume=task.is_resume, status=task.status,
                    error_code=(task.error or {}).get("error"),
                    error_message=redact((task.error or {}).get("message", "")),
                    return_code=process.poll() if process is not None else None,
                    process_reaped=task.process_reaped, timed_out=timed_out,
                    duration_ms=int((time.monotonic() - started) * 1000),
                    failure_reason=redact(monitor.failure_reason or "") if monitor else None,
                    plugin_ready=monitor.ready if monitor else None,
                    plugin_error=redact(monitor.reminder_error or "") if monitor else None,
                    plugin_error_detail=redact(monitor.reminder_error_detail or "") if monitor else None,
                )

    def _check_analysis(self, task: TaskState, monitor: AnalysisMonitor, process: subprocess.Popen[bytes], stdout: Any, *, exited: bool = False) -> None:
        """检查冻结 attempt 的金额和插件，触发时确认进程回收后发布稳定错误。"""
        if not exited and time.monotonic() < monitor.next_check:
            return
        try:
            if task.session_id is None:
                # 独立打开文件描述符，读取位置不能改变子进程的 stdout 写入位置。
                with open(f"/proc/self/fd/{stdout.fileno()}", "rb") as snapshot:
                    self._capture_session(task, snapshot)
            monitor.check(task.session_id, exited=exited)
        except (AnalysisControlError, RuntimeError) as exc:
            code = exc.code if isinstance(exc, AnalysisControlError) else "agent_analysis_usage_unavailable"
            reason = exc.reason if isinstance(exc, AnalysisControlError) else str(exc)
            monitor.failure_reason = reason
            termination = self.process_supervisor.terminate(process)
            reaped = termination.reaped
            signal_name: str | None = None
            if termination.returncode in {-15, 143}:
                signal_name = "SIGTERM"
            elif termination.returncode in {-9, 137}:
                signal_name = "SIGKILL"
            elif termination.reaped and termination.returncode is not None:
                signal_name = "already_exited"
            diagnostic = {
                "threshold_cost": str(monitor.policy.get("termination_cost")),
                "final_observed_cost": str(monitor.observed_cost),
                "check_stage": "analysis_monitor",
                "termination_signal": signal_name,
                "process_reaped": reaped,
                "trigger_reason": code,
            }
            with self.lock:
                task.process_reaped = reaped
            logger.bind(
                attempt_id=task.executor_attempt_id, policy=task.analysis_policy,
                observed_cost=str(monitor.observed_cost), phase=reason,
                process_reaped=reaped, trigger=code,
            ).error(
                "Agent 分析用量控制终止：attempt={}；策略={}；观测金额={}；原因={}；阶段={}；进程回收={}",
                task.executor_attempt_id, task.analysis_policy, str(monitor.observed_cost), code, reason, reaped,
            )
            public_code = code if reaped else "unknown_execution"
            message = "超过最大分析程度" if public_code == "agent_maximum_analysis_depth_exceeded" else f"{code}:{reason}" if reaped else f"进程回收未确认；触发原因={code}；阶段={reason}"
            raise _ProcessFailure(public_code, message, analysis_diagnostic=diagnostic) from exc
        finally:
            with self.lock:
                task.observed_cost = str(monitor.observed_cost)
                task.usage_checked_at = monitor.usage_checked_at
                task.reminder_sent = monitor.reminder_sent

    def _capture_session(self, task: TaskState, stdout: Any) -> None:
        """从完整临时 stdout 流逐行绑定 session，避免采样截断 JSONL。"""
        found_session: str | None = None
        stdout.seek(0)
        while True:
            line = stdout.readline(SESSION_CAPTURE_LINE_BYTES + 1)
            if not line:
                break
            # BufferedIO.readline(size) 可能只返回超长行的前缀；丢弃该行剩余内容，
            # 让后续 JSONL 行仍能被解析，同时把单行内存限制在固定上限附近。
            if len(line) > SESSION_CAPTURE_LINE_BYTES:
                while line and not line.endswith(b"\n"):
                    line = stdout.readline(SESSION_CAPTURE_LINE_BYTES + 1)
                continue
            try:
                candidate = _session_id_from_event(json.loads(line))
            except (UnicodeDecodeError, json.JSONDecodeError, RecursionError):
                continue
            if candidate and SESSION_ID_RE.fullmatch(candidate):
                if task.session_id and candidate != task.session_id:
                    raise RuntimeError("session_binding_mismatch")
                if found_session and candidate != found_session:
                    raise RuntimeError("session_binding_mismatch")
                found_session = candidate
        if found_session:
            with self.lock:
                task.session_id = found_session

    @staticmethod
    def _classify_process_failure(stdout: bytes, stderr: bytes) -> tuple[str, int | None]:
        """按有限输出区分 provider 网关、网络和普通进程暂态错误。"""
        text = (stdout[:256 * 1024] + stderr[:16 * 1024]).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", 429
        server_status = next((status for status in statuses if 500 <= status <= 599), None)
        if server_status is not None:
            return "agent_provider_server_error", server_status
        if any(token in text for token in ("econnreset", "connection reset", "connection refused", "socket hang up", "network error", "timed out")):
            return "agent_connection_interrupted", None
        return "agent_process_failed", None

    def _validate_result_file(self, path: Path) -> None:
        """验证结果文件位置、大小和基本 JSON 结构；完整 schema 由后端复核。"""
        try:
            ExecutorResultStore(RESULT_ROOT, filename=RESULT_FILE_NAME, max_bytes=self.max_result_bytes).read(
                path,
                required_fields=set(REQUIRED_RESULT_FIELDS),
                validator=lambda value: validate_agent_result(value, secret_inventory=secret_inventory_from_mapping(os.environ)),
            )
        except ExecutorResultStoreError as exc:
            raise _ResultFileValidationFailure(exc.code, reason_code=exc.reason_code) from exc

    @staticmethod
    def _terminate(process: subprocess.Popen[bytes]) -> bool:
        """终止任务进程组并返回父进程是否已被 waitpid 确认回收。"""
        return ProcessSupervisor().terminate(process).reaped

    def cancel(self, task_id: str, *, reason: str = "task_interrupted") -> TaskState:
        """取消指定任务；只终止该任务进程，不影响 executor 或其他任务。"""
        if reason not in {"task_interrupted", "service_shutdown"}:
            raise ValueError("invalid_interruption_reason")
        with self.lock:
            task = self.tasks.get(task_id)
            if task is None:
                raise KeyError(task_id)
            if task.status in {"succeeded", "failed", "cancelled"}:
                return task
            # 用户取消优先；后续停机通知不能改变已经收到的取消原因。
            if task.cancel_reason is None or reason == "task_interrupted":
                task.cancel_reason = reason
            task.cancel_event.set()
            was_queued = task.status == "queued"
            process = task.process
            if process is None and was_queued:
                task.status = "failed" if task.cancel_reason == "service_shutdown" else "cancelled"
                task.error = _json_error(task.cancel_reason, "服务正常关闭，尚未开始执行" if task.cancel_reason == "service_shutdown" else "任务已取消")
                task.completed_at = time.time()
                try:
                    self._persist_attempt_metadata_locked(task)
                except (OSError, ValueError, TypeError) as exc:
                    task.status = "failed"
                    task.error = _json_error("agent_attempt_metadata_write_failed", "无法保存 executor attempt 恢复信息")
                    logger.error("attempt 恢复信息保存失败：attempt={} error={}", task.executor_attempt_id, type(exc).__name__)
                task.done.set()
        # 运行任务由执行线程终止并完成数据库维护，再统一发布最终状态。
        return task

    def close(self) -> None:
        """停止线程池并取消仍未结束的任务。"""
        with self.lock:
            ids = [task.executor_attempt_id for task in self.tasks.values() if task.status not in {"succeeded", "failed", "cancelled"}]
        for task_id in ids:
            self.cancel(task_id)
        self.pool.shutdown(wait=False, cancel_futures=True)


def shutil_which(executable: str) -> str | None:
    """在不导入 shell 的情况下检查固定 OpenCode 可执行文件。"""
    if not executable or "/" in executable:
        path = Path(executable)
        return str(path) if path.is_file() and os.access(path, os.X_OK) else None
    for directory in os.getenv("PATH", "").split(os.pathsep):
        candidate = Path(directory) / executable
        if candidate.is_file() and os.access(candidate, os.X_OK):
            return str(candidate)
    return None


class Handler(BaseHTTPRequestHandler):
    """executor HTTP 路由，只公开健康、任务提交、状态和取消接口。"""

    server: "ExecutorHTTPServer"

    def setup(self) -> None:
        """为请求体读取设置有限超时，避免慢速连接长期占用线程。"""
        super().setup()
        self.connection.settimeout(30)

    def log_message(self, _format: str, *_args: object) -> None:
        """禁止将认证头、任务输入或 OpenCode 诊断写入访问日志。"""
        return

    def _authorized(self) -> bool:
        """使用固定 Bearer token 验证内部 API 调用。"""
        expected = self.server.executor.token
        value = self.headers.get("Authorization", "")
        supplied = value[7:].strip() if value.startswith("Bearer ") else ""
        return bool(expected and supplied and hmac.compare_digest(supplied, expected))

    def _send(self, status: int, payload: object) -> None:
        """发送统一 JSON 响应并关闭连接。"""
        body = _safe_json(payload)
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def _read_json(self) -> object:
        """读取有限大小的 JSON 请求体，拒绝 multipart、流式 shell 等其他协议。"""
        if self.headers.get_content_type() != "application/json":
            raise ValueError("invalid_task")
        try:
            length = int(self.headers.get("Content-Length", "-1"))
        except ValueError as exc:
            raise ValueError("invalid_task") from exc
        if length < 0 or length > 64 * 1024:
            raise ValueError("invalid_task")
        try:
            raw = self.rfile.read(length)
            return json.loads(raw.decode("utf-8"))
        except (OSError, UnicodeDecodeError, json.JSONDecodeError, RecursionError) as exc:
            raise ValueError("invalid_task") from exc

    def do_GET(self) -> None:  # noqa: N802
        """处理健康和任务状态查询。"""
        path = urlsplit(self.path).path.rstrip("/") or "/"
        if path == "/health":
            if not self._authorized():
                self._send(401, _json_error("executor_unauthorized", "executor 认证失败"))
                return
            self._send(200, self.server.executor.health())
            return
        if not self._authorized():
            self._send(401, _json_error("executor_unauthorized", "executor 认证失败"))
            return
        match = re.fullmatch(r"/v1/tasks/([^/]+)", path)
        if match:
            task_id = unquote(match.group(1))
            with self.server.executor.lock:
                task = self.server.executor.tasks.get(task_id)
            if task is None:
                self._send(404, _json_error("task_not_found", "任务不存在"))
            else:
                self._send(200, task.public())
            return
        self._send(404, _json_error("not_found", "接口不存在"))

    def do_POST(self) -> None:  # noqa: N802
        """处理固定研究任务提交和任务取消。"""
        if not self._authorized():
            self._send(401, _json_error("executor_unauthorized", "executor 认证失败"))
            return
        path = urlsplit(self.path).path.rstrip("/") or "/"
        cancel_match = re.fullmatch(r"/v1/tasks/([^/]+)/(cancel|interrupt-for-shutdown)", path)
        if cancel_match:
            try:
                reason = "service_shutdown" if cancel_match.group(2) == "interrupt-for-shutdown" else "task_interrupted"
                task = self.server.executor.cancel(unquote(cancel_match.group(1)), reason=reason)
            except KeyError:
                self._send(404, _json_error("task_not_found", "任务不存在"))
                return
            self._send(200, task.public())
            return
        if path != "/v1/tasks":
            self._send(404, _json_error("not_found", "接口不存在"))
            return
        try:
            payload = self._read_json()
            task, wait = self.server.executor.submit(payload)
        except ValueError as exc:
            code = str(exc)
            messages = {
                "agent_image_path_forbidden": "图片路径不在受控图片目录内",
                "agent_result_path_invalid": "任务标识非法",
                "invalid_reverse_image_policy": "反向图片策略无效",
                "agent_analysis_policy_missing": "启用分析用量控制但缺少冻结策略",
                "agent_analysis_policy_invalid": "分析用量策略无效",
                "agent_timeout_limit_exceeded": "任务超时超过 executor 上限",
                "session_binding_mismatch": "续跑 session 与任务事实不匹配",
                "opencode_workspace_invalid": "workspace selector 或目录无效",
                "opencode_workspace_mismatch": "workspace capability 与任务事实不匹配",
                "opencode_workspace_capability_invalid": "workspace capability 无效",
                "opencode_workspace_capability_expired": "workspace capability 已过期",
                "opencode_workspace_capability_unavailable": "workspace capability 未配置",
                "model_capability_invalid": "模型 capability 无效",
                "model_capability_unavailable": "模型 capability 未配置",
                "model_broker_endpoint_invalid": "模型 broker 未配置",
                "model_name_invalid": "模型标识无效",
                "visual_match_snapshot_invalid": "视觉候选 snapshot 无效",
            }
            if code not in messages and code not in {"invalid_task"}:
                code = "invalid_task"
            self._send(400, _json_error(code, messages.get(code, "任务请求无效")))
            return
        except RuntimeError as exc:
            code = str(exc)
            if code not in {
                "agent_backpressure",
                "task_exists",
                "agent_runtime_unavailable",
                "opencode_not_configured",
                "session_binding_mismatch",
                "session_not_resumable",
                "unknown_execution",
                "opencode_workspace_invalid",
                "opencode_workspace_mismatch",
                "opencode_workspace_capability_invalid",
                "opencode_workspace_capability_expired",
                "opencode_workspace_capability_unavailable",
                "model_capability_invalid",
                "model_capability_unavailable",
                "model_broker_endpoint_invalid",
                "model_name_invalid",
                "agent_analysis_policy_missing",
                "agent_analysis_policy_invalid",
                "visual_match_snapshot_invalid",
                "visual_candidate_materialization_failed",
            }:
                code = "agent_runtime_unavailable"
            status = {
                "agent_backpressure": 429,
                "task_exists": 409,
                "session_binding_mismatch": 409,
                "session_not_resumable": 409,
                "unknown_execution": 409,
                "agent_runtime_unavailable": 503,
                "opencode_not_configured": 503,
            }.get(code, 503)
            self._send(status, _json_error(code, "任务当前无法执行"))
            return
        if wait:
            task.done.wait(task.timeout_seconds + 5)
        status = 200 if task.status in {"succeeded", "failed", "cancelled"} else 202
        self._send(status, task.public())


class ExecutorHTTPServer(ThreadingHTTPServer):
    """带 executor 状态对象的线程化 HTTP 服务。"""

    daemon_threads = True
    allow_reuse_address = True

    def __init__(self, address: tuple[str, int], executor: Executor):
        """绑定固定 loopback 监听地址供 Compose 内网转发。"""
        self.executor = executor
        super().__init__(address, Handler)


def main() -> None:
    """启动 executor；端口只由容器内部环境配置，不发布到宿主机。"""
    configure_executor_diagnostics(Path(os.getenv("MEMEMEOW_EXECUTOR_DIAGNOSTICS_ROOT", str(RUNTIME_ROOT / "executor-logs"))))
    host = os.getenv("MEMEMEOW_AGENT_EXECUTOR_HOST", "0.0.0.0")
    port = _env_int("MEMEMEOW_AGENT_EXECUTOR_PORT", 8277, 1, 65535)
    telemetry = start_telemetry("mememeow-executor")
    executor = Executor()
    server = ExecutorHTTPServer((host, port), executor)
    try:
        server.serve_forever(poll_interval=0.2)
    except KeyboardInterrupt:
        pass
    finally:
        executor.close()
        server.server_close()
        if telemetry is not None:
            telemetry.close()


if __name__ == "__main__":
    main()