mirror of
https://github.com/MemeMeow-Studio/MemeMeow.git
synced 2026-10-17 04:41:01 +08:00
infstellar
794 lines
33 KiB
Python
794 lines
33 KiB
Python
"""OpenCode 候选解析与持久任务恢复测试。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
from io import BytesIO
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from backend import opencode as opencode_module
|
||
from backend.config import Settings
|
||
from backend.opencode import DIAGNOSTIC_LOG_BYTES, OPENCODE_REASONING_VARIANT, OpenCodeError, OpenCodeRunner
|
||
from backend.tasks import PersistentTaskService
|
||
|
||
|
||
def make_settings(tmp_path: Path) -> Settings:
|
||
"""构造不依赖真实 OpenCode 的隔离配置。"""
|
||
return Settings(
|
||
data_root=tmp_path / "data",
|
||
image_root=tmp_path / "images",
|
||
embedding_api_key=None,
|
||
embedding_base_url=None,
|
||
embedding_model="embedding",
|
||
llm_enhance_model=None,
|
||
protected_mode=False,
|
||
allowed_endpoints=("/",),
|
||
rate_limit_enabled=False,
|
||
rate_limit_requests=1,
|
||
rate_limit_window=1,
|
||
max_upload_size=1,
|
||
opencode_executable=None,
|
||
opencode_model=None,
|
||
)
|
||
|
||
|
||
def candidate() -> dict[str, object]:
|
||
"""提供满足研究输出契约的最小候选。"""
|
||
return {
|
||
"title": "测试标题",
|
||
"summary": "一张用于测试的图片",
|
||
"subjects": ["主体"],
|
||
"visible_text": [],
|
||
"references": [],
|
||
"meaning": None,
|
||
"keywords": ["测试"],
|
||
"search_queries": [],
|
||
"uncertainties": [],
|
||
"source_urls": [],
|
||
}
|
||
|
||
|
||
def test_candidate_extraction_rejects_extra_text_and_accepts_single_fence(tmp_path: Path):
|
||
"""只接受完整 JSON 或唯一 fenced JSON，不猜测花括号边界。"""
|
||
runner = OpenCodeRunner(make_settings(tmp_path))
|
||
raw = __import__("json").dumps(candidate(), ensure_ascii=False)
|
||
assert runner.extract_candidate(raw)["title"] == "测试标题"
|
||
assert runner.extract_candidate(f"```json\n{raw}\n```")["summary"] == "一张用于测试的图片"
|
||
with pytest.raises(OpenCodeError) as error:
|
||
runner.extract_candidate(f"说明\n```json\n{raw}\n```")
|
||
assert error.value.code == "agent_output_invalid_json"
|
||
|
||
|
||
def test_candidate_validation_checks_required_fields_and_uri_format(tmp_path: Path):
|
||
"""输出必须通过 schema 和 Pydantic 双重字段约束。"""
|
||
runner = OpenCodeRunner(make_settings(tmp_path))
|
||
assert runner.validate_candidate(candidate())["title"] == "测试标题"
|
||
invalid = candidate()
|
||
invalid["source_urls"] = ["not-a-uri"]
|
||
with pytest.raises(OpenCodeError) as error:
|
||
runner.validate_candidate(invalid)
|
||
assert error.value.code == "agent_output_schema_invalid"
|
||
|
||
|
||
def test_candidate_validation_rejects_title_punctuation(tmp_path: Path):
|
||
"""Agent 即使输出结构合法，也不能把标点或符号写入标题。"""
|
||
runner = OpenCodeRunner(make_settings(tmp_path))
|
||
invalid = candidate()
|
||
invalid["title"] = "滑稽表情“认真！”"
|
||
with pytest.raises(OpenCodeError) as error:
|
||
runner.validate_candidate(invalid)
|
||
assert error.value.code == "agent_output_schema_invalid"
|
||
|
||
|
||
def test_candidate_validation_preserves_public_boundary_reason(tmp_path: Path):
|
||
"""host 候选和结果文件包装都应保留安全的公开边界原因码。"""
|
||
runner = OpenCodeRunner(make_settings(tmp_path))
|
||
invalid = candidate()
|
||
invalid["references"] = ["path=/runtime/secret"]
|
||
|
||
with pytest.raises(OpenCodeError) as candidate_error:
|
||
runner.validate_candidate(invalid)
|
||
assert candidate_error.value.code == "agent_output_schema_invalid"
|
||
assert candidate_error.value.reason_code == "result_sensitive_data"
|
||
|
||
_draft, result_path = runner.create_task_result_paths("reason-code")
|
||
result_path.write_text(json.dumps(invalid, ensure_ascii=False), encoding="utf-8")
|
||
with pytest.raises(OpenCodeError) as file_error:
|
||
runner.read_result_file(result_path)
|
||
assert file_error.value.code == "agent_result_file_schema_invalid"
|
||
assert file_error.value.reason_code == "result_sensitive_data"
|
||
|
||
_draft, object_result_path = runner.create_task_result_paths("object-reason-code")
|
||
object_result_path.write_text("[]", encoding="utf-8")
|
||
with pytest.raises(OpenCodeError) as object_error:
|
||
runner.read_result_file(object_result_path)
|
||
assert object_error.value.code == "agent_result_file_schema_invalid"
|
||
assert object_error.value.reason_code == "result_object_required"
|
||
|
||
|
||
def test_missing_opencode_configuration_has_stable_error(tmp_path: Path):
|
||
"""没有可执行文件或模型时 worker 返回稳定诊断。"""
|
||
with pytest.raises(OpenCodeError) as error:
|
||
OpenCodeRunner(make_settings(tmp_path)).prepare_runtime()
|
||
assert error.value.code == "opencode_not_configured"
|
||
|
||
|
||
def test_prepare_runtime_writes_common_config_without_secrets(tmp_path: Path):
|
||
"""runtime 配置固定在 workspace，provider 凭据只保留环境变量引用。"""
|
||
project = tmp_path / "project"
|
||
(project / "skills" / "research-meme-context").mkdir(parents=True)
|
||
modules = project / "node_modules"
|
||
(modules / "@ai-sdk" / "openai").mkdir(parents=True)
|
||
executable = tmp_path / "opencode"
|
||
executable.write_text("#!/bin/sh\n", encoding="utf-8")
|
||
settings = Settings(
|
||
**{
|
||
**make_settings(tmp_path).__dict__,
|
||
"opencode_executable": str(executable),
|
||
"opencode_model": "mememeow/gpt-5.6-luna",
|
||
"opencode_base_url": "https://example.invalid/v1",
|
||
"opencode_api_key": "not-written-to-config",
|
||
"opencode_node_modules": modules,
|
||
}
|
||
)
|
||
runner = OpenCodeRunner(settings, project_root=project)
|
||
runner.prepare_runtime()
|
||
|
||
payload = __import__("json").loads((runner.workspace / "opencode.json").read_text(encoding="utf-8"))
|
||
assert payload["experimental"] == {"continue_loop_on_deny": True}
|
||
provider = payload["provider"]["mememeow"]
|
||
assert provider["npm"] == "@ai-sdk/openai"
|
||
assert provider["options"] == {
|
||
"baseURL": "{env:MEMEMEOW_OPENCODE_BASE_URL}",
|
||
"apiKey": "{env:MEMEMEOW_OPENCODE_API_KEY}",
|
||
}
|
||
assert set(provider["models"]) == {"gpt-5.6-luna"}
|
||
assert provider["models"]["gpt-5.6-luna"]["variants"] == {
|
||
"max": {"reasoningEffort": "max"}
|
||
}
|
||
assert provider["models"]["gpt-5.6-luna"]["cost"]["input"] == 0.25
|
||
assert provider["models"]["gpt-5.6-luna"]["cost"]["output"] == 2
|
||
assert "not-written-to-config" not in (runner.workspace / "opencode.json").read_text(encoding="utf-8")
|
||
|
||
|
||
def test_runtime_environment_isolates_project_config(tmp_path: Path):
|
||
"""后台任务和检查脚本必须固定 DB、显式配置并禁止合并父目录配置。"""
|
||
settings = Settings(
|
||
**{
|
||
**make_settings(tmp_path).__dict__,
|
||
"opencode_model": "mememeow/gpt-5.6-luna",
|
||
"opencode_base_url": "https://example.invalid/v1",
|
||
"opencode_api_key": "test-key",
|
||
}
|
||
)
|
||
runner = OpenCodeRunner(settings)
|
||
environment = runner.build_environment()
|
||
assert environment["OPENCODE_DB"] == str(runner.db_path)
|
||
assert environment["OPENCODE_CONFIG"] == str(runner.workspace / "opencode.json")
|
||
assert environment["OPENCODE_CONFIG_DIR"] == str(runner.workspace / ".opencode")
|
||
assert environment["OPENCODE_DISABLE_PROJECT_CONFIG"] == "1"
|
||
assert "SERPAPI_API_KEY" not in environment
|
||
assert environment["MEMEMEOW_REVERSE_IMAGE_INTERNAL_URL"].endswith("/internal/reverse-image/search")
|
||
|
||
|
||
def test_host_runtime_environment_contains_only_claim_scoped_credentials(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
||
"""Host 模式也必须把当前 claim task id 传给薄客户端，避免请求落到错误任务。"""
|
||
monkeypatch.setenv("MEMEMEOW_AGENT_CALLBACK_SECRET", "root-callback-secret")
|
||
monkeypatch.setenv("MEMEMEOW_AGENT_EXECUTOR_TOKEN", "executor-token")
|
||
monkeypatch.setenv("MEMEMEOW_DATABASE_URL", "postgresql://should-not-enter-agent")
|
||
settings = Settings(
|
||
**{
|
||
**make_settings(tmp_path).__dict__,
|
||
"opencode_model": "mememeow/gpt-5.6-luna",
|
||
"opencode_base_url": "https://example.invalid/v1",
|
||
"opencode_api_key": "test-key",
|
||
"agent_runtime_mode": "host",
|
||
}
|
||
)
|
||
runner = OpenCodeRunner(settings)
|
||
environment = runner.build_environment(2, "claim-task-123")
|
||
assert environment["MEMEMEOW_OPENCODE_SLOT"] == "2"
|
||
assert environment["MEMEMEOW_AGENT_TASK_ID"] == "claim-task-123"
|
||
assert "MEMEMEOW_AGENT_CALLBACK_TOKEN" not in environment
|
||
assert "MEMEMEOW_AGENT_CALLBACK_SECRET" not in environment
|
||
assert "MEMEMEOW_AGENT_EXECUTOR_TOKEN" not in environment
|
||
assert "MEMEMEOW_DATABASE_URL" not in environment
|
||
assert "SERPAPI_API_KEY" not in environment
|
||
|
||
|
||
def test_prepare_runtime_uses_project_opencode_modules_by_default(tmp_path: Path):
|
||
"""未覆盖依赖路径时复用项目的 OpenCode 插件依赖而非前端依赖。"""
|
||
project = tmp_path / "project"
|
||
(project / "skills" / "research-meme-context").mkdir(parents=True)
|
||
modules = project / ".opencode" / "node_modules"
|
||
(modules / "@ai-sdk" / "openai").mkdir(parents=True)
|
||
executable = tmp_path / "opencode"
|
||
executable.write_text("#!/bin/sh\n", encoding="utf-8")
|
||
settings = Settings(
|
||
**{
|
||
**make_settings(tmp_path).__dict__,
|
||
"opencode_executable": str(executable),
|
||
"opencode_model": "mememeow/gpt-5.6-luna",
|
||
"opencode_base_url": "https://example.invalid/v1",
|
||
"opencode_api_key": "test-key",
|
||
}
|
||
)
|
||
runner = OpenCodeRunner(settings, project_root=project)
|
||
runner.prepare_runtime()
|
||
assert (runner.workspace / "node_modules").resolve() == modules.resolve()
|
||
|
||
|
||
def test_runtime_config_is_accepted_by_installed_opencode(tmp_path: Path):
|
||
"""已安装 CLI 必须展开 workspace 配置中的环境变量引用。"""
|
||
executable = shutil.which("opencode")
|
||
if executable is None:
|
||
pytest.skip("当前环境未安装 OpenCode CLI")
|
||
project = tmp_path / "project"
|
||
(project / "skills" / "research-meme-context").mkdir(parents=True)
|
||
modules = project / "node_modules"
|
||
(modules / "@ai-sdk" / "openai").mkdir(parents=True)
|
||
settings = Settings(
|
||
**{
|
||
**make_settings(tmp_path).__dict__,
|
||
"opencode_executable": executable,
|
||
"opencode_model": "mememeow/gpt-5.6-luna",
|
||
"opencode_base_url": "https://example.invalid/v1",
|
||
"opencode_api_key": "test-key",
|
||
"opencode_node_modules": modules,
|
||
}
|
||
)
|
||
runner = OpenCodeRunner(settings, project_root=project)
|
||
runner.prepare_runtime()
|
||
environment = runner.build_environment()
|
||
result = subprocess.run(
|
||
[executable, "debug", "config"],
|
||
cwd=runner.workspace,
|
||
env=environment,
|
||
capture_output=True,
|
||
text=True,
|
||
check=False,
|
||
)
|
||
assert result.returncode == 0, result.stderr
|
||
resolved = json.loads(result.stdout)
|
||
assert resolved["provider"]["mememeow"]["options"] == {
|
||
"baseURL": "https://example.invalid/v1",
|
||
"apiKey": "test-key",
|
||
}
|
||
|
||
|
||
def test_runner_fixes_luna_at_max_reasoning_variant(tmp_path: Path):
|
||
"""每个研究任务传绝对图片路径并显式要求 max 推理强度。"""
|
||
settings = Settings(
|
||
**{
|
||
**make_settings(tmp_path).__dict__,
|
||
"opencode_executable": "opencode",
|
||
"opencode_model": "mememeow/gpt-5.6-luna",
|
||
"opencode_base_url": "https://example.invalid/v1",
|
||
"opencode_api_key": "test-key",
|
||
}
|
||
)
|
||
command = OpenCodeRunner(settings)._run_command(tmp_path / "image.png", "prompt")
|
||
file_index = command.index("--file")
|
||
assert command[file_index + 1] == str((tmp_path / "image.png").resolve())
|
||
variant_index = command.index("--variant")
|
||
assert command[variant_index + 1] == OPENCODE_REASONING_VARIANT == "max"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("output", "expected"),
|
||
[
|
||
(b'{"error":{"statusCode":429}}\n', "agent_provider_rate_limited"),
|
||
(b'{"error":{"status":503}}\n', "agent_provider_server_error"),
|
||
(b"connection reset by peer\n", "agent_connection_interrupted"),
|
||
],
|
||
)
|
||
def test_runner_classifies_transient_process_failures(output: bytes, expected: str) -> None:
|
||
"""Runner 将 429、5xx 和连接中断收敛为可审计稳定错误。"""
|
||
assert OpenCodeRunner._classify_process_failure(BytesIO(output), BytesIO()) == expected
|
||
|
||
|
||
def test_runner_classifies_provider_error_in_large_output_tail() -> None:
|
||
"""大 JSONL 前缀超过采样上限时，尾部 521 仍必须映射为 provider 失败。"""
|
||
output = b"noise" * (opencode_module.STREAM_COPY_CHUNK_BYTES * 8) + b'\n{"error":{"statusCode":521}}\n'
|
||
assert OpenCodeRunner._classify_process_failure(BytesIO(output), BytesIO()) == "agent_provider_server_error"
|
||
assert OpenCodeRunner._process_http_status(BytesIO(output), BytesIO()) == 521
|
||
|
||
|
||
def test_runner_resume_command_requires_explicit_session(tmp_path: Path) -> None:
|
||
"""host 回滚续跑只追加明确 session，不使用全局 continue。"""
|
||
runner = OpenCodeRunner(make_settings(tmp_path))
|
||
command = runner._run_command(tmp_path / "image.png", "prompt", resume_session_id="session-1")
|
||
assert command[command.index("--session") + 1] == "session-1"
|
||
assert "--continue" not in command
|
||
|
||
|
||
def test_host_resume_prompt_requires_finishing_unfinished_work(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""host 回滚续跑 prompt 必须明确保留草稿并继续未完成工作。"""
|
||
settings = Settings(
|
||
**{
|
||
**make_settings(tmp_path).__dict__,
|
||
"opencode_model": "mememeow/gpt-5.6-luna",
|
||
"opencode_base_url": "https://example.invalid/v1",
|
||
"opencode_api_key": "test-key",
|
||
"agent_runtime_mode": "host",
|
||
}
|
||
)
|
||
runner = OpenCodeRunner(settings, project_root=tmp_path)
|
||
runner.runtime_root.mkdir(parents=True, exist_ok=True)
|
||
runner.workspace.mkdir(parents=True, exist_ok=True)
|
||
runner.slots_root.mkdir(parents=True, exist_ok=True)
|
||
runner.log_root.mkdir(parents=True, exist_ok=True)
|
||
image = tmp_path / "image.png"
|
||
image.write_bytes(b"image")
|
||
captured: dict[str, str] = {}
|
||
monkeypatch.setattr(runner, "prepare_runtime", lambda: None)
|
||
monkeypatch.setattr(runner, "_run_command", lambda _image, prompt, **_kwargs: captured.setdefault("prompt", prompt) or ["fake-opencode"])
|
||
monkeypatch.setattr(opencode_module.subprocess, "Popen", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("stop-after-prompt")))
|
||
with pytest.raises(RuntimeError, match="stop-after-prompt"):
|
||
runner.run(image, lambda _value, _message: None, task_id="host-resume", resume_session_id="session-1")
|
||
assert "同一 OpenCode session 的受控续跑" in captured["prompt"]
|
||
assert "只完成尚未完成的工作" in captured["prompt"]
|
||
|
||
|
||
def test_runner_accepts_large_cli_output_without_accumulating_pipe_bytes(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
||
"""CLI 事件流超过旧门禁时仍可解析，持久日志只保留诊断前缀。"""
|
||
project = tmp_path / "project"
|
||
(project / "skills" / "research-meme-context").mkdir(parents=True)
|
||
modules = project / "node_modules"
|
||
(modules / "@ai-sdk" / "openai").mkdir(parents=True)
|
||
executable = tmp_path / "fake-opencode.py"
|
||
executable.write_text(
|
||
"#!/usr/bin/env python3\n"
|
||
"import sys\n"
|
||
"if sys.argv[1] == 'run':\n"
|
||
" sys.stdout.write('{\\\"type\\\":\\\"noise\\\",\\\"payload\\\":\\\"' + ('x' * (3 * 1024 * 1024)) + '\\\"}\\n')\n"
|
||
" sys.stdout.write('{\\\"type\\\":\\\"session.created\\\",\\\"session_id\\\":\\\"large-session\\\"}\\n')\n",
|
||
encoding="utf-8",
|
||
)
|
||
executable.chmod(0o755)
|
||
settings = Settings(
|
||
**{
|
||
**make_settings(tmp_path).__dict__,
|
||
"opencode_executable": str(executable),
|
||
"opencode_model": "mememeow/gpt-5.6-luna",
|
||
"opencode_base_url": "https://example.invalid/v1",
|
||
"opencode_api_key": "test-key",
|
||
"opencode_node_modules": modules,
|
||
}
|
||
)
|
||
runner = OpenCodeRunner(settings, project_root=project)
|
||
monkeypatch.setattr(runner, "_session_messages", lambda session_id, environment, task_id=None: {"messages": [{"role": "assistant", "parts": [{"type": "text", "text": json.dumps(candidate(), ensure_ascii=False)}]}]})
|
||
monkeypatch.setattr(runner, "validate_candidate", lambda value: value)
|
||
image = tmp_path / "image.png"
|
||
image.write_bytes(b"image")
|
||
|
||
result, session_id = runner.run(image, lambda value, message: None)
|
||
|
||
assert session_id == "large-session"
|
||
assert result["title"] == "测试标题"
|
||
log_path = runner.log_root / f"{hashlib.sha256(str(image).encode()).hexdigest()[:16]}.jsonl"
|
||
assert log_path.stat().st_size == DIAGNOSTIC_LOG_BYTES
|
||
|
||
|
||
def test_host_does_not_fallback_to_transcript_for_invalid_result_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
||
"""结果文件存在但越过公开边界时，host 不能用合法 transcript 绕过拒绝。"""
|
||
project = tmp_path / "project"
|
||
(project / "skills" / "research-meme-context").mkdir(parents=True)
|
||
modules = project / "node_modules"
|
||
(modules / "@ai-sdk" / "openai").mkdir(parents=True)
|
||
executable = tmp_path / "fake-opencode.py"
|
||
executable.write_text(
|
||
"#!/usr/bin/env python3\n"
|
||
"import sys\n"
|
||
"if sys.argv[1] == 'run':\n"
|
||
" sys.stdout.write('{\\\"type\\\":\\\"session.created\\\",\\\"session_id\\\":\\\"boundary-session\\\"}\\n')\n",
|
||
encoding="utf-8",
|
||
)
|
||
executable.chmod(0o755)
|
||
settings = Settings(
|
||
**{
|
||
**make_settings(tmp_path).__dict__,
|
||
"opencode_executable": str(executable),
|
||
"opencode_model": "mememeow/gpt-5.6-luna",
|
||
"opencode_base_url": "https://example.invalid/v1",
|
||
"opencode_api_key": "test-key",
|
||
"opencode_node_modules": modules,
|
||
}
|
||
)
|
||
runner = OpenCodeRunner(settings, project_root=project)
|
||
invalid = candidate()
|
||
invalid["references"] = ["path=/runtime/secret"]
|
||
|
||
def write_invalid_result(_image: Path, _prompt: str, **kwargs: object) -> list[str]:
|
||
"""在 CLI 启动前写入越过公开边界的结果文件。"""
|
||
result_path = kwargs["result_path"]
|
||
assert isinstance(result_path, Path)
|
||
result_path.write_text(json.dumps(invalid, ensure_ascii=False), encoding="utf-8")
|
||
return [str(executable), "run"]
|
||
|
||
monkeypatch.setattr(runner, "_run_command", write_invalid_result)
|
||
monkeypatch.setattr(
|
||
runner,
|
||
"_session_messages",
|
||
lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("invalid result must not use transcript fallback")),
|
||
)
|
||
image = tmp_path / "image.png"
|
||
image.write_bytes(b"image")
|
||
|
||
with pytest.raises(OpenCodeError) as error:
|
||
runner.run(image, lambda _value, _message: None, task_id="result-boundary")
|
||
|
||
assert error.value.code == "agent_result_file_schema_invalid"
|
||
assert error.value.reason_code == "result_sensitive_data"
|
||
|
||
|
||
def test_session_messages_stream_large_response_without_read_limit(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
||
"""session API 大响应按块落盘读取，不再因固定字节门禁失败。"""
|
||
runner = OpenCodeRunner(make_settings(tmp_path))
|
||
body = json.dumps([{"role": "tool", "diagnostic": "x" * (3 * 1024 * 1024)}, {"role": "assistant", "parts": [{"type": "text", "text": "{}"}]}]).encode()
|
||
read_sizes: list[int] = []
|
||
|
||
class Response:
|
||
"""提供可记录 read 尺寸的最小 HTTP 响应夹具。"""
|
||
|
||
def __init__(self, payload: bytes):
|
||
self.payload = payload
|
||
self.position = 0
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, exc_type, exc_value, traceback):
|
||
return False
|
||
|
||
def read(self, size: int = -1) -> bytes:
|
||
read_sizes.append(size)
|
||
if size < 0:
|
||
size = len(self.payload) - self.position
|
||
chunk = self.payload[self.position : self.position + size]
|
||
self.position += len(chunk)
|
||
return chunk
|
||
|
||
class Server:
|
||
"""避免测试启动真实 OpenCode server。"""
|
||
|
||
pid = 0
|
||
|
||
def poll(self):
|
||
return None
|
||
|
||
monkeypatch.setattr(opencode_module.subprocess, "Popen", lambda *args, **kwargs: Server())
|
||
monkeypatch.setattr(opencode_module, "urlopen", lambda endpoint, timeout=1: Response(body))
|
||
monkeypatch.setattr(runner, "_terminate", lambda process: None)
|
||
|
||
result = runner._session_messages("large-session", runner.build_environment())
|
||
|
||
assert result["messages"][-1]["role"] == "assistant"
|
||
assert read_sizes
|
||
assert max(read_sizes) <= opencode_module.STREAM_COPY_CHUNK_BYTES
|
||
|
||
|
||
def test_opencode_launcher_reuses_runtime_for_session_list(tmp_path: Path):
|
||
"""会话检查入口必须复用同一 workspace 和 DB，并把 list 映射到公开 CLI。"""
|
||
project = Path(__file__).resolve().parent.parent
|
||
executable = tmp_path / "fake-opencode"
|
||
capture = tmp_path / "capture.txt"
|
||
executable.write_text(
|
||
"#!/usr/bin/env bash\n"
|
||
"printf '%s\\n' \"$PWD\" \"$OPENCODE_DB\" \"$OPENCODE_CONFIG\" "
|
||
f"\"$OPENCODE_CONFIG_DIR\" \"$OPENCODE_DISABLE_PROJECT_CONFIG\" \"$@\" > {str(capture)!r}\n",
|
||
encoding="utf-8",
|
||
)
|
||
executable.chmod(0o755)
|
||
modules = tmp_path / "node_modules"
|
||
(modules / "@ai-sdk" / "openai").mkdir(parents=True)
|
||
runtime = tmp_path / "runtime"
|
||
environment = {
|
||
**os.environ,
|
||
"MEMEMEOW_OPENCODE_EXECUTABLE": str(executable),
|
||
"MEMEMEOW_OPENCODE_MODEL": "mememeow/gpt-5.6-luna",
|
||
"MEMEMEOW_OPENCODE_BASE_URL": "https://example.invalid/v1",
|
||
"MEMEMEOW_OPENCODE_API_KEY": "test-key",
|
||
"MEMEMEOW_OPENCODE_RUNTIME_ROOT": str(runtime),
|
||
"MEMEMEOW_OPENCODE_NODE_MODULES": str(modules),
|
||
"MEMEMEOW_AGENT_RUNTIME_MODE": "host",
|
||
"MEMEMEOW_PYTHON": sys.executable,
|
||
}
|
||
result = subprocess.run(
|
||
[str(project / "scripts" / "open-opencode.sh"), "--list", "--format", "json"],
|
||
cwd=project,
|
||
env=environment,
|
||
capture_output=True,
|
||
text=True,
|
||
check=False,
|
||
)
|
||
assert result.returncode == 0, result.stderr
|
||
lines = capture.read_text(encoding="utf-8").splitlines()
|
||
workspace = str((runtime / "workspace").resolve())
|
||
assert lines[:5] == [
|
||
workspace,
|
||
str((runtime / "opencode.db").resolve()),
|
||
str((runtime / "workspace" / "opencode.json").resolve()),
|
||
str((runtime / "workspace" / ".opencode").resolve()),
|
||
"1",
|
||
]
|
||
assert lines[5:] == ["session", "list", "--format", "json"]
|
||
|
||
|
||
def test_opencode_launcher_fails_closed_for_unconfigured_executor(tmp_path: Path):
|
||
"""显式 executor 缺少凭据时，诊断入口不能绕过配置检查进入 Compose。"""
|
||
project = Path(__file__).resolve().parent.parent
|
||
compose_file = tmp_path / "compose.yml"
|
||
compose_file.write_text("services: {}\n", encoding="utf-8")
|
||
environment = {
|
||
**os.environ,
|
||
"MEMEMEOW_AGENT_RUNTIME_MODE": "executor",
|
||
"MEMEMEOW_AGENT_EXECUTOR_URL": "http://mememeow-agent-runtime:8277",
|
||
"MEMEMEOW_AGENT_EXECUTOR_TOKEN": " ",
|
||
"MEMEMEOW_COMPOSE_FILE": str(compose_file),
|
||
"MEMEMEOW_PYTHON": sys.executable,
|
||
}
|
||
environment.pop("MEMEMEOW_AGENT_EXECUTOR_TOKEN_FILE", None)
|
||
result = subprocess.run(
|
||
[str(project / "scripts" / "open-opencode.sh"), "--list"],
|
||
cwd=project,
|
||
env=environment,
|
||
capture_output=True,
|
||
text=True,
|
||
check=False,
|
||
)
|
||
assert result.returncode == 1
|
||
assert "OpenCode runtime 初始化失败 [agent_executor_not_configured]" in result.stderr
|
||
|
||
|
||
def test_opencode_launcher_uses_running_compose_runtime_for_session_list(tmp_path: Path):
|
||
"""Compose 服务运行时必须查询 named volume，而不是宿主旧数据库。"""
|
||
project = Path(__file__).resolve().parent.parent
|
||
capture = tmp_path / "capture-compose-list.txt"
|
||
compose_file = tmp_path / "compose file.yml"
|
||
compose_file.write_text("services: {}\n", encoding="utf-8")
|
||
fake_docker = tmp_path / "docker"
|
||
fake_docker.write_text(
|
||
"#!/usr/bin/env bash\n"
|
||
"if [[ \"$1\" == info ]]; then exit 0; fi\n"
|
||
"if [[ \" $* \" == *\" ps --status running --services mememeow-agent-runtime \"* ]]; then\n"
|
||
" printf 'mememeow-agent-runtime\\n'\n"
|
||
" exit 0\n"
|
||
"fi\n"
|
||
f"printf '%s\\n' \"$@\" > {str(capture)!r}\n",
|
||
encoding="utf-8",
|
||
)
|
||
fake_docker.chmod(0o755)
|
||
host_runtime = tmp_path / "host-runtime"
|
||
environment = {
|
||
**os.environ,
|
||
"PATH": f"{tmp_path}:{os.environ.get('PATH', '')}",
|
||
"MEMEMEOW_AGENT_RUNTIME_MODE": "executor",
|
||
"MEMEMEOW_AGENT_EXECUTOR_URL": "http://mememeow-agent-runtime:8277",
|
||
"MEMEMEOW_AGENT_EXECUTOR_TOKEN": "diagnostic-token",
|
||
"MEMEMEOW_COMPOSE_FILE": str(compose_file),
|
||
"MEMEMEOW_OPENCODE_RUNTIME_ROOT": str(host_runtime),
|
||
"MEMEMEOW_PYTHON": sys.executable,
|
||
}
|
||
environment.pop("MEMEMEOW_AGENT_EXECUTOR_TOKEN_FILE", None)
|
||
result = subprocess.run(
|
||
[str(project / "scripts" / "open-opencode.sh"), "--list", "--format", "json"],
|
||
cwd=tmp_path,
|
||
env=environment,
|
||
capture_output=True,
|
||
text=True,
|
||
check=False,
|
||
)
|
||
|
||
assert result.returncode == 0, result.stderr
|
||
lines = capture.read_text(encoding="utf-8").splitlines()
|
||
assert lines[:3] == ["compose", "-f", str(compose_file)]
|
||
assert "exec" in lines
|
||
assert "-T" in lines
|
||
assert "OPENCODE_DB=/runtime/opencode.db" in lines
|
||
assert lines[-4:] == ["session", "list", "--format", "json"]
|
||
assert str(host_runtime) not in "\n".join(lines)
|
||
|
||
|
||
def test_opencode_launcher_uses_container_model_for_compose_tui(tmp_path: Path):
|
||
"""Compose TUI 必须读取容器模型配置，并把未知参数原样交给 OpenCode。"""
|
||
project = Path(__file__).resolve().parent.parent
|
||
capture = tmp_path / "capture-compose-tui.txt"
|
||
compose_file = tmp_path / "docker-compose.yml"
|
||
compose_file.write_text("services: {}\n", encoding="utf-8")
|
||
fake_docker = tmp_path / "docker"
|
||
fake_docker.write_text(
|
||
"#!/usr/bin/env bash\n"
|
||
"if [[ \"$1\" == info ]]; then exit 0; fi\n"
|
||
"if [[ \" $* \" == *\" ps --status running --services mememeow-agent-runtime \"* ]]; then\n"
|
||
" printf 'mememeow-agent-runtime\\n'\n"
|
||
" exit 0\n"
|
||
"fi\n"
|
||
f"printf '%s\\n' \"$@\" > {str(capture)!r}\n",
|
||
encoding="utf-8",
|
||
)
|
||
fake_docker.chmod(0o755)
|
||
environment = {
|
||
**os.environ,
|
||
"PATH": f"{tmp_path}:{os.environ.get('PATH', '')}",
|
||
"MEMEMEOW_AGENT_RUNTIME_MODE": "auto",
|
||
"MEMEMEOW_AGENT_EXECUTOR_URL": "http://mememeow-agent-runtime:8277",
|
||
"MEMEMEOW_AGENT_EXECUTOR_TOKEN": "diagnostic-token",
|
||
"MEMEMEOW_COMPOSE_FILE": str(compose_file),
|
||
"MEMEMEOW_PYTHON": sys.executable,
|
||
}
|
||
environment.pop("MEMEMEOW_AGENT_EXECUTOR_TOKEN_FILE", None)
|
||
result = subprocess.run(
|
||
[str(project / "scripts" / "open-opencode.sh"), "--session", "ses_test"],
|
||
cwd=project,
|
||
env=environment,
|
||
capture_output=True,
|
||
text=True,
|
||
check=False,
|
||
)
|
||
|
||
assert result.returncode == 0, result.stderr
|
||
lines = capture.read_text(encoding="utf-8").splitlines()
|
||
assert "-T" not in lines
|
||
shell_index = lines.index("sh")
|
||
assert lines[shell_index : shell_index + 4] == [
|
||
"sh",
|
||
"-lc",
|
||
'exec opencode /runtime/workspace --model "$MEMEMEOW_OPENCODE_MODEL" "$@"',
|
||
"open-opencode",
|
||
]
|
||
assert lines[-2:] == ["--session", "ses_test"]
|
||
|
||
|
||
@pytest.mark.parametrize("runtime_mode", ["auto", "executor"])
|
||
def test_opencode_launcher_rejects_stale_host_runtime_when_compose_is_down(
|
||
tmp_path: Path,
|
||
runtime_mode: str,
|
||
):
|
||
"""Compose 模式不可在服务停机时静默回退到宿主历史 runtime。"""
|
||
project = Path(__file__).resolve().parent.parent
|
||
compose_file = tmp_path / "docker-compose.yml"
|
||
compose_file.write_text("services: {}\n", encoding="utf-8")
|
||
fake_docker = tmp_path / "docker"
|
||
fake_docker.write_text("#!/usr/bin/env bash\nexit 1\n", encoding="utf-8")
|
||
fake_docker.chmod(0o755)
|
||
environment = {
|
||
**os.environ,
|
||
"PATH": f"{tmp_path}:{os.environ.get('PATH', '')}",
|
||
"MEMEMEOW_AGENT_RUNTIME_MODE": runtime_mode,
|
||
"MEMEMEOW_AGENT_EXECUTOR_URL": "http://mememeow-agent-runtime:8277",
|
||
"MEMEMEOW_AGENT_EXECUTOR_TOKEN": "diagnostic-token",
|
||
"MEMEMEOW_COMPOSE_FILE": str(compose_file),
|
||
"MEMEMEOW_PYTHON": sys.executable,
|
||
}
|
||
environment.pop("MEMEMEOW_AGENT_EXECUTOR_TOKEN_FILE", None)
|
||
result = subprocess.run(
|
||
[str(project / "scripts" / "open-opencode.sh"), "--list"],
|
||
cwd=project,
|
||
env=environment,
|
||
capture_output=True,
|
||
text=True,
|
||
check=False,
|
||
)
|
||
|
||
assert result.returncode == 1
|
||
assert "Compose Agent 服务 mememeow-agent-runtime 未运行" in result.stderr
|
||
|
||
|
||
def test_last_assistant_message_excludes_tool_content(tmp_path: Path):
|
||
"""session 消息解析兼容公开 API 结构且只使用最后 assistant 的 text parts。"""
|
||
runner = OpenCodeRunner(make_settings(tmp_path))
|
||
text = runner._last_assistant_text({"messages": [
|
||
{"role": "assistant", "parts": [{"type": "text", "text": "旧内容"}]},
|
||
{"role": "tool", "parts": [{"type": "text", "text": "工具结果"}]},
|
||
{
|
||
"info": {"role": "assistant"},
|
||
"parts": [
|
||
{"type": "reasoning", "text": "内部推理"},
|
||
{"type": "text", "text": "{"},
|
||
{"type": "text", "text": "}"},
|
||
{"type": "step-finish"},
|
||
],
|
||
},
|
||
]})
|
||
assert text == "{}"
|
||
|
||
|
||
def test_process_error_diagnostic_reads_opencode_jsonl_error(tmp_path: Path):
|
||
"""非零退出时应从 OpenCode JSONL 中保留有限的上游错误诊断。"""
|
||
runner = OpenCodeRunner(make_settings(tmp_path))
|
||
stdout = json.dumps({
|
||
"type": "error",
|
||
"error": {"data": {"message": "Upstream request failed", "statusCode": 400}},
|
||
}).encode()
|
||
assert runner._process_error_diagnostic(stdout, b"") == "Upstream request failed (HTTP 400)"
|
||
assert runner._process_error_diagnostic(stdout, b"CLI failed\n") == "CLI failed"
|
||
|
||
|
||
def test_persistent_task_keeps_stable_error_code_and_limited_diagnostic(tmp_path: Path):
|
||
"""任务服务保留 Agent 失败码，同时允许有限的安全诊断。"""
|
||
service = PersistentTaskService(tmp_path / "tasks")
|
||
|
||
def fail(payload, progress):
|
||
raise RuntimeError("agent_process_failed: Upstream request failed (HTTP 400)")
|
||
|
||
service.register("meme_context_generation", fail)
|
||
service.start()
|
||
task = service.submit("meme_context_generation", {})
|
||
import time
|
||
|
||
for _ in range(50):
|
||
record = service.get(task.task_id)
|
||
if record and record.status == "failed":
|
||
break
|
||
time.sleep(0.01)
|
||
record = service.get(task.task_id)
|
||
assert record.error == {
|
||
"error": "agent_process_failed",
|
||
"message": "agent_process_failed: Upstream request failed (HTTP 400)",
|
||
}
|
||
service.shutdown()
|
||
|
||
|
||
def test_persistent_tasks_recover_queued_and_interrupt_running(tmp_path: Path):
|
||
"""服务重启会恢复 queued 并将旧 running 记录标为 task_interrupted。"""
|
||
root = tmp_path / "tasks"
|
||
first = PersistentTaskService(root)
|
||
queued = first.submit("known", {"value": 1})
|
||
running = first.submit("other", {"value": 2})
|
||
first.update(running.task_id, status="running")
|
||
first._executor.shutdown(wait=False, cancel_futures=True)
|
||
|
||
second = PersistentTaskService(root)
|
||
handled: list[int] = []
|
||
second.register("known", lambda payload, progress: handled.append(payload["value"]))
|
||
second.start()
|
||
import time
|
||
|
||
for _ in range(50):
|
||
if second.get(queued.task_id).status == "succeeded":
|
||
break
|
||
time.sleep(0.01)
|
||
assert second.get(queued.task_id).status == "succeeded"
|
||
assert handled == [1]
|
||
interrupted = second.get(running.task_id)
|
||
assert interrupted.status == "failed"
|
||
assert interrupted.error["error"] == "task_interrupted"
|
||
second.shutdown()
|
||
|
||
|
||
def test_slot_lock_is_cross_runner_and_does_not_reuse_busy_slot(tmp_path: Path):
|
||
"""两个应用 runner 不能同时持有同一个 slot 文件锁。"""
|
||
settings = Settings(
|
||
**{
|
||
**make_settings(tmp_path).__dict__,
|
||
"opencode_concurrency": 2,
|
||
}
|
||
)
|
||
first = OpenCodeRunner(settings)
|
||
second = OpenCodeRunner(settings)
|
||
first.slots_root.mkdir(parents=True, exist_ok=True)
|
||
first_slot, first_handle = first._acquire_slot()
|
||
second_slot, second_handle = second._acquire_slot()
|
||
assert first_slot != second_slot
|
||
first._release_slot(first_slot, first_handle)
|
||
second._release_slot(second_slot, second_handle)
|
||
|
||
|
||
def test_slot_acquire_after_shutdown_does_not_leak_semaphore(tmp_path: Path):
|
||
"""关闭中的 runner 拒绝新任务且不会遗留已占用的进程内配额。"""
|
||
runner = OpenCodeRunner(make_settings(tmp_path))
|
||
runner.shutdown()
|
||
with pytest.raises(OpenCodeError) as error:
|
||
runner._acquire_slot()
|
||
assert error.value.code == "task_interrupted"
|
||
assert runner._slot_semaphore.acquire(timeout=0.1)
|
||
runner._slot_semaphore.release()
|