MemeMeow/tests/test_api.py

"""FastAPI 公共契约、文件边界和上传行为测试。"""

from __future__ import annotations

import io
import json
import zipfile
import os
import threading
import time
from datetime import UTC, datetime, timedelta
from pathlib import Path
from uuid import UUID, uuid4

import pytest
from fastapi.testclient import TestClient
from PIL import Image
from pydantic import ValidationError
from sqlalchemy import select, text

from api import ConcurrencyUpdateRequest, app
from backend.collection_packages import CollectionManifest, CollectionManifestCollection, CollectionManifestMember, serialize_manifest, sha256_bytes
from backend.database import EMBEDDING_DIMENSIONS, ImageProcessingJob, Meme, MemeTextEmbedding, StorageOperation, Task, create_engine_for_url
from backend.image_processing import ImageProcessingWorker


def _clear_test_scope() -> None:
    """清理 API 测试使用的 local scope 业务行，保留安装标记和 scope 命名空间。"""
    url = os.getenv("MEMEMEOW_TEST_DATABASE_URL")
    if not url:
        pytest.skip("未设置 MEMEMEOW_TEST_DATABASE_URL，拒绝清理开发数据库")
    engine = create_engine_for_url(url, pool_size=1, max_overflow=0)
    statements = (
        "DELETE FROM agent_callback_requests",
        "DELETE FROM reverse_image_usage_events",
        "DELETE FROM operation_grants",
        "DELETE FROM image_processing_attempts",
        "DELETE FROM image_processing_stages",
        "DELETE FROM image_processing_jobs",
        "DELETE FROM search_migration_states",
        "DELETE FROM task_lane_resource_slots",
        "DELETE FROM task_lane_slots",
        "DELETE FROM task_batch_items",
        "DELETE FROM task_batches",
        "DELETE FROM tasks",
        "DELETE FROM meme_visual_embeddings",
        "DELETE FROM meme_text_embeddings",
        "DELETE FROM meme_embeddings",
        "DELETE FROM search_heads",
        "DELETE FROM search_generations",
        "DELETE FROM storage_operations",
        "DELETE FROM memes",
    )
    with engine.begin() as connection:
        for statement in statements:
            connection.execute(text(statement))
    engine.dispose()


def test_concurrency_update_request_accepts_large_positive_value() -> None:
    """设置接口请求模型保留大正整数，Agent 队列不设置数量门禁。"""
    assert ConcurrencyUpdateRequest.model_validate({"value": 128}).opencode_concurrency == 128
    with pytest.raises(ValidationError):
        ConcurrencyUpdateRequest.model_validate({"value": 0})


@pytest.fixture
def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    """使用隔离图片目录启动完整应用生命周期。"""
    test_url = os.getenv("MEMEMEOW_TEST_DATABASE_URL")
    if not test_url:
        pytest.skip("未设置 MEMEMEOW_TEST_DATABASE_URL，跳过 API PostgreSQL 测试")
    monkeypatch.setenv("MEMEMEOW_DATABASE_URL", test_url)
    monkeypatch.setenv("MEMEMEOW_DATA_ROOT", str(tmp_path / "data"))
    monkeypatch.setenv("MEMEMEOW_IMAGE_ROOT", str(tmp_path / "images"))
    monkeypatch.setenv("EMBEDDING_API_KEY", "")
    monkeypatch.setenv("EMBEDDING_BASE_URL", "")
    monkeypatch.setenv("MEMEMEOW_OPENCODE_MODEL", "")
    monkeypatch.setenv("MEMEMEOW_AGENT_CALLBACK_SECRET", "test-agent-callback-secret-1234")
    _clear_test_scope()
    with TestClient(app) as test_client:
        uploaded = test_client.post("/images/upload", files=[("files", ("deleted.png", png_bytes(), "image/png"))]).json()["results"][0]

        def fake_embed(content, *, filename="image"):
            """模拟视觉服务，确保夹具直接进入 Agent 阶段而不触发真实模型。"""
            del content, filename
            settings = test_client.app.state.settings
            return {
                "model": settings.visual_model,
                "dimensions": settings.visual_model_dimensions,
                "preprocess_version": settings.visual_preprocess_version,
                "embedding": [1.0] + [0.0] * (settings.visual_model_dimensions - 1),
            }

        monkeypatch.setattr(test_client.app.state.visual_inference, "embed", fake_embed)

        def fake_run(path, progress, *, task_id=None, **kwargs):
            """模拟 Agent 完成前目标图片被删除。"""
            del progress, kwargs
            assert task_id
            path.unlink()
            return {}, "deleted-session"

        monkeypatch.setattr(test_client.app.state.opencode, "run", fake_run)
        response = test_client.post("/images/context", json={"meme_id": uploaded["meme_id"]})
        assert response.status_code == 202
        task_id = response.json()["task_id"]
        deadline = time.monotonic() + 5
        while time.monotonic() < deadline:
            status = test_client.get(f"/tasks/{task_id}").json()
            if status["status"] in {"succeeded", "failed"}:
                break
            time.sleep(0.01)
        assert status["status"] == "failed"
        assert status["error"]["error"] == "target_changed"
    _clear_test_scope()
    with TestClient(app) as test_client:
        yield test_client, tmp_path
    _clear_test_scope()


def png_bytes(color: str = "red") -> bytes:
    """生成可解码的小型 PNG 测试数据。"""
    output = io.BytesIO()
    Image.new("RGB", (4, 4), color=color).save(output, format="PNG")
    return output.getvalue()


def collection_zip_bytes() -> bytes:
    """生成包含一张图片的最小合集包，验证导入任务衔接。"""
    content = png_bytes("green")
    source_id = uuid4().hex
    member = CollectionManifestMember(
        source_meme_id=source_id,
        filename_at_export="imported.png",
        path=f"images/{source_id}.png",
        extension=".png",
        size_bytes=len(content),
        sha256=sha256_bytes(content),
    )
    manifest = CollectionManifest(
        format="mememeow-collection",
        format_version=1,
        collection=CollectionManifestCollection(name=f"导入测试-{source_id[:8]}"),
        members=[member],
    )
    output = io.BytesIO()
    with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_STORED) as archive:
        archive.writestr("manifest.json", serialize_manifest(manifest))
        archive.writestr(member.path, content)
    return output.getvalue()


def test_config_is_masked_and_search_requires_cache(client):
    """配置响应不泄露密钥，缓存未就绪时搜索返回规范错误。"""
    test_client, _ = client
    config = test_client.get("/config")
    assert config.status_code == 200
    assert "embedding_api_key" not in config.json()
    assert "opencode_executable" not in config.json()
    assert config.json()["embedding_api_key_configured"] is False
    assert config.json()["embedding_cache_ready"] is False
    response = test_client.post("/search", json={"query": "开心"})
    assert response.status_code == 503
    assert response.json()["error"] == "cache_not_ready"


def test_startup_allows_root_orphan_files_without_listing_them(tmp_path, monkeypatch):
    """启动预检允许根目录孤立图片，并继续只从数据库列出可信 Meme。"""
    test_url = os.getenv("MEMEMEOW_TEST_DATABASE_URL")
    if not test_url:
        pytest.skip("未设置 MEMEMEOW_TEST_DATABASE_URL，跳过 API PostgreSQL 测试")
    image_root = tmp_path / "images"
    image_root.mkdir(parents=True)
    orphan = image_root / "legacy.png"
    orphan.write_bytes(png_bytes())
    monkeypatch.setenv("MEMEMEOW_DATABASE_URL", test_url)
    monkeypatch.setenv("MEMEMEOW_DATA_ROOT", str(tmp_path / "data"))
    monkeypatch.setenv("MEMEMEOW_IMAGE_ROOT", str(image_root))
    monkeypatch.setenv("EMBEDDING_API_KEY", "")
    monkeypatch.setenv("EMBEDDING_BASE_URL", "")
    monkeypatch.setenv("MEMEMEOW_OPENCODE_MODEL", "")
    _clear_test_scope()
    with TestClient(app) as test_client:
        health = test_client.get("/health")
        assert health.status_code == 200
        assert health.json()["storage_preflight"] == {"status": "warning", "orphan_files": 1, "blocking_errors": {}}
        assert test_client.get("/config").json()["storage_preflight"] == health.json()["storage_preflight"]
        assert test_client.get("/images").json()["total"] == 0
        assert orphan.is_file()
    _clear_test_scope()


@pytest.mark.parametrize("payload", [{"query": ""}, {"query": "x", "n_results": 0}, {"query": "x", "n_results": 31}, {"query": "x", "n_results": "5"}])
def test_search_validation_is_consistent(client, payload):
    """空查询和越界数量统一返回 400。"""
    test_client, _ = client
    response = test_client.post("/search", json=payload)
    assert response.status_code == 400
    assert response.json()["error"] == "invalid_request"


def test_get_search_is_not_an_alias(client):
    """旧 GET 检索入口不会执行检索逻辑。"""
    test_client, _ = client
    assert test_client.get("/search", params={"q": "x"}).status_code == 405


def test_upload_lists_and_serves_image(client):
    """上传后的图片可立即列出并通过受控媒体接口读取。"""
    test_client, _ = client
    response = test_client.post("/images/upload", files=[("files", ("hello.png", png_bytes(), "image/png"))])
    assert response.status_code == 200
    uploaded = response.json()["results"][0]
    assert uploaded["ok"] is True
    meme_id = uploaded["meme_id"]
    UUID(meme_id)
    listing = test_client.get("/images").json()
    assert listing["total"] == 1
    media_url = listing["items"][0]["media_url"]
    media = test_client.get(media_url)
    assert media.status_code == 200
    assert media.headers["content-type"] == "image/png"
    listed = test_client.get("/images").json()["items"][0]
    assert listed["meme_id"] == meme_id
    assert listed["media_url"] == f"/media/{meme_id}"
    assert listed["metadata"]["status"] == "pending"
    assert listed["embedding_status"] == "pending"


def test_upload_rejects_more_than_twenty_files_before_durable_write(client):
    """超过每请求文件数边界时在任何 Meme 或 operation 写入前拒绝。"""
    test_client, _ = client
    response = test_client.post(
        "/images/upload",
        files=[("files", (f"too-many-{index}.png", png_bytes(), "image/png")) for index in range(21)],
    )
    assert response.status_code == 413
    assert response.json()["error"] == "too_many_files"
    assert test_client.get("/images").json()["total"] == 0


def test_upload_accepts_twenty_files_and_exposes_default_limits(client):
    """恰好 20 个文件可以逐项入库，配置响应保持 20/客户端 2 并发提示/disabled。"""
    test_client, _ = client
    response = test_client.post(
        "/images/upload",
        files=[("files", (f"limit-{index}.png", png_bytes(), "image/png")) for index in range(20)],
    )
    assert response.status_code == 200
    assert len(response.json()["results"]) == 20
    assert all(item["ok"] for item in response.json()["results"])
    config = test_client.get("/config").json()
    assert config["max_files_per_request"] == 20
    assert config["max_concurrent_upload_requests"] == 2
    assert config["max_request_bytes"] is None


def test_upload_optional_total_budget_is_enforced_without_content_length_dependency(client):
    """启用总预算后按实际读取字节拒绝超限请求，而不是信任 Content-Length。"""
    test_client, _ = client
    settings = test_client.app.state.settings
    previous = settings.max_request_bytes
    settings.max_request_bytes = len(png_bytes())
    try:
        response = test_client.post(
            "/images/upload",
            files=[
                ("files", ("budget-a.png", png_bytes("red"), "image/png")),
                ("files", ("budget-b.png", png_bytes("blue"), "image/png")),
            ],
        )
    finally:
        settings.max_request_bytes = previous
    assert response.status_code == 413
    assert response.json()["error"] == "request_too_large"
    assert test_client.get("/images").json()["total"] == 0


def test_upload_retry_reuses_durable_meme_operation_and_processing_job(client):
    """响应丢失后的同内容重试只返回原 Meme 和处理 job，不重复 durable 事实。"""
    test_client, _ = client
    content = png_bytes("red")
    first = test_client.post("/images/upload", files=[("files", ("idempotent.png", content, "image/png"))])
    assert first.status_code == 200
    first_item = first.json()["results"][0]
    engine = test_client.app.state.database.engine
    with engine.connect() as connection:
        before = (
            connection.execute(select(Meme).where(Meme.scope_id == "local")).all(),
            connection.execute(select(StorageOperation).where(StorageOperation.scope_id == "local")).all(),
            connection.execute(select(ImageProcessingJob).where(ImageProcessingJob.scope_id == "local")).all(),
        )
    second = test_client.post("/images/upload", files=[("files", ("idempotent.png", content, "image/png"))])
    assert second.status_code == 200
    second_item = second.json()["results"][0]
    assert second_item["ok"] is True
    assert second_item["idempotent"] is True
    assert second_item["meme_id"] == first_item["meme_id"]
    with engine.connect() as connection:
        after = (
            connection.execute(select(Meme).where(Meme.scope_id == "local")).all(),
            connection.execute(select(StorageOperation).where(StorageOperation.scope_id == "local")).all(),
            connection.execute(select(ImageProcessingJob).where(ImageProcessingJob.scope_id == "local")).all(),
        )
    assert tuple(len(items) for items in after) == tuple(len(items) for items in before)


def test_upload_retry_reports_reconciliation_when_file_fact_changes(client):
    """数据库记录与实际文件指纹不一致时不盲目认领上传。"""
    test_client, tmp_path = client
    content = png_bytes("red")
    first = test_client.post("/images/upload", files=[("files", ("reconcile.png", content, "image/png"))])
    assert first.status_code == 200
    with test_client.app.state.database.environment("local") as environment:
        record = environment.memes.get(first.json()["results"][0]["meme_id"])
        assert record is not None
        physical_image = test_client.app.state.metadata.blob_store.resolve(record.storage_key)
    physical_image.write_bytes(png_bytes("blue"))
    response = test_client.post("/images/upload", files=[("files", ("reconcile.png", content, "image/png"))])
    assert response.status_code == 200
    assert response.json()["results"][0]["error"] == "upload_reconciliation_required"


def test_multi_upload_seals_visual_batch(client):
    """多图上传必须把视觉任务登记到批次后再封口，不能返回 batch_not_found。"""
    test_client, _ = client
    response = test_client.post(
        "/images/upload",
        files=[
            ("files", ("batch-a.png", png_bytes("red"), "image/png")),
            ("files", ("batch-b.png", png_bytes("blue"), "image/png")),
        ],
    )
    assert response.status_code == 200
    payload = response.json()
    assert payload["batch_id"]
    assert all(item["ok"] and item["visual_task_id"] for item in payload["results"])


def test_image_metadata_returns_complete_database_record(client):
    """图片库详情接口只返回受控 Meme 对应的完整数据库元数据。"""
    test_client, _ = client
    response = test_client.post("/images/upload", files=[("files", ("detail.png", png_bytes(), "image/png"))])
    assert response.status_code == 200
    meme_id = response.json()["results"][0]["meme_id"]
    with test_client.app.state.database.environment("local") as environment:
        record = environment.memes.get(meme_id)
        assert record is not None
        image = test_client.app.state.metadata.blob_store.resolve(record.storage_key)
    test_client.app.state.metadata.update_context(image, {"summary": "详情摘要", "keywords": ["测试"]}, producer="human", status="ready")

    metadata = test_client.get("/images/metadata", params={"meme_id": meme_id})
    assert metadata.status_code == 200
    payload = metadata.json()
    assert payload["image"]["relative_path"] == "detail.png"
    assert payload["context_status"] == "ready"
    assert payload["meme_context"]["summary"] == "详情摘要"
    assert payload["meme_context"]["keywords"] == ["测试"]


def test_image_metadata_requires_stable_id_and_reports_unknown_meme(client):
    """详情接口拒绝旧路径参数，并对未知 Meme 返回稳定错误。"""
    test_client, _ = client
    assert test_client.get("/images/metadata", params={"filename": "missing.png"}).json()["error"] == "meme_id_required"
    response = test_client.get("/images/metadata", params={"meme_id": str(uuid4())})
    assert response.status_code == 404
    assert response.json()["error"] == "meme_not_found"


def test_batch_upload_keeps_success_and_reports_failure(client):
    """批量上传允许部分成功且按内容身份幂等复用重复文件。"""
    test_client, _ = client
    response = test_client.post(
        "/images/upload",
        files=[
            ("files", ("good.png", png_bytes(), "image/png")),
            ("files", ("bad.png", b"not-an-image", "image/png")),
            ("files", ("note.txt", b"text", "text/plain")),
        ],
    )
    results = response.json()["results"]
    assert [item["ok"] for item in results] == [True, False, False]
    duplicate = test_client.post("/images/upload", files=[("files", ("good-copy.png", png_bytes(), "image/png"))])
    duplicate_item = duplicate.json()["results"][0]
    assert duplicate_item["ok"] is True
    assert duplicate_item["idempotent"] is True
    assert duplicate_item["meme_id"] == results[0]["meme_id"]


def test_upload_queues_context_job_and_removed_vlm_routes_are_not_found(client):
    """上传创建异步语境任务，原 VLM 路由不再暴露。"""
    test_client, _ = client
    response = test_client.post("/images/upload", files=[("files", ("pending.png", png_bytes(), "image/png"))])
    item = response.json()["results"][0]
    assert item["metadata_job_id"]
    assert test_client.get(f"/tasks/{item['metadata_job_id']}").status_code == 200
    assert test_client.post("/images/describe", json={"filename": "pending.png"}).status_code == 404
    assert test_client.post("/images/label-batch", json={"items": []}).status_code == 404


def test_collection_import_starts_with_visual_task(client):
    """合集导入后的新 Meme 必须先进入视觉任务，而不是直接进入 Agent。"""
    test_client, _ = client
    response = test_client.post("/collections/import", files={"file": ("import.zip", collection_zip_bytes(), "application/zip")})
    assert response.status_code == 200
    item = response.json()["results"][0]
    assert item["ok"] is True
    assert item["visual_task_id"] == item["metadata_job_id"]
    task = test_client.get(f"/tasks/{item['visual_task_id']}")
    assert task.status_code == 200
    assert task.json()["task_type"] == "visual_embedding_generation"


def test_removed_directory_contract_and_rename_conflicts(client):
    """目录接口永久删除，展示名可以重复且物理内容仍独立。"""
    test_client, _ = client
    assert test_client.get("/images/directories").status_code == 404
    assert test_client.post("/images/directories", json={"name": "work"}).status_code == 404
    one = test_client.post("/images/upload", files=[("files", ("one.png", png_bytes(), "image/png"))]).json()["results"][0]
    two = test_client.post("/images/upload", files=[("files", ("two.png", png_bytes("blue"), "image/png"))]).json()["results"][0]
    renamed_duplicate = test_client.post("/images/rename", json={"meme_id": one["meme_id"], "new_name": "two"})
    assert renamed_duplicate.status_code == 200
    assert renamed_duplicate.json()["filename"] == "two.png"
    renamed = test_client.post("/images/rename", json={"meme_id": one["meme_id"], "new_name": "first"})
    assert renamed.status_code == 200
    assert renamed.json()["meme_id"] == one["meme_id"]
    assert renamed.json()["filename"] == "first.png"
    assert test_client.get(f"/media/{one['meme_id']}").status_code == 200
    assert test_client.get(f"/media/{two['meme_id']}").status_code == 200
    with test_client.app.state.database.environment("local") as environment:
        first_record = environment.memes.get(one["meme_id"])
        second_record = environment.memes.get(two["meme_id"])
        assert first_record is not None and second_record is not None
        assert first_record.storage_key != second_record.storage_key


def test_delete_removes_image_and_database_record(client):
    """删除接口会移除数据库 Meme、文件和受控媒体访问。"""
    test_client, tmp_path = client
    uploaded = test_client.post("/images/upload", files=[("files", ("remove.png", png_bytes(), "image/png"))]).json()["results"][0]
    meme_id = uploaded["meme_id"]
    response = test_client.post("/images/delete", json={"meme_id": meme_id})
    assert response.status_code == 200
    assert not (tmp_path / "images" / "remove.png").exists()
    assert test_client.get(f"/media/{meme_id}").status_code == 404
    assert test_client.get("/images").json()["total"] == 0


def test_metadata_repair_is_pollable(client):
    """元数据修复任务报告数据库与文件完整性问题。"""
    test_client, tmp_path = client
    image = tmp_path / "images" / "legacy.png"
    image.write_bytes(png_bytes())
    response = test_client.post("/images/metadata/repair")
    assert response.status_code == 202
    task_id = response.json()["task_id"]
    for _ in range(50):
        status = test_client.get(f"/tasks/{task_id}").json()
        if status["status"] in {"succeeded", "failed"}:
            break
    assert status["status"] == "succeeded"
    assert status["result"]["processed"] == 0
    assert "legacy.png" in status["result"]["orphan_files"]
    second = test_client.post("/images/metadata/repair")
    assert second.status_code == 202


def test_path_traversal_and_symlink_escape_are_blocked(client):
    """目录穿越和指向根目录外的符号链接均不可读取。"""
    test_client, tmp_path = client
    outside = tmp_path / "outside"
    outside.mkdir()
    (outside / "secret.png").write_bytes(png_bytes())
    (tmp_path / "images" / "escape").symlink_to(outside, target_is_directory=True)
    traversal = test_client.get("/images", params={"directory": "../outside"})
    assert traversal.status_code == 400
    escaped = test_client.get("/media/escape/secret.png")
    assert escaped.status_code == 404


def test_unknown_task_has_stable_error(client):
    """未知或重启后丢失的任务返回 task_not_found。"""
    test_client, _ = client
    response = test_client.get("/tasks/unknown")
    assert response.status_code == 404
    assert response.json()["error"] == "task_not_found"


def test_task_list_filters_and_does_not_expose_payload(client):
    """持久任务列表支持类型筛选且不会返回内部 payload。"""
    test_client, _ = client
    submitted = test_client.post("/generate-cache")
    assert submitted.status_code == 202
    response = test_client.get("/tasks", params={"task_type": "cache_generation", "limit": 1})
    assert response.status_code == 200
    items = response.json()["items"]
    assert len(items) == 1
    assert items[0]["task_type"] == "cache_generation"
    assert "payload" not in items[0]


def test_search_maps_results_to_media_urls(client):
    """正常检索只返回受控媒体 URL 且不重复。"""
    test_client, _ = client
    uploaded = test_client.post("/images/upload", files=[("files", ("a.png", png_bytes(), "image/png"))]).json()["results"][0]
    meme_id = uploaded["meme_id"]

    class FakeSearch:
        def has_cache(self):
            return True

        def search(self, *args, **kwargs):
            return [meme_id, meme_id, str(uuid4())]

    test_client.app.state.search_engine = FakeSearch()
    response = test_client.post("/search", json={"query": "x", "n_results": 3})
    assert response.status_code == 200
    assert response.json() == {
        "results": [f"/media/{meme_id}"],
        "result_media": [
            {
                "meme_id": meme_id,
                "media_url": f"/media/{meme_id}",
                "thumbnail": {"status": "pending", "media_url": None},
            }
        ],
    }


def test_llm_failure_falls_back_to_original_query(client):
    """LLM 增强异常时仍执行一次普通检索。"""
    test_client, _ = client
    meme_id = test_client.post("/images/upload", files=[("files", ("a.png", png_bytes(), "image/png"))]).json()["results"][0]["meme_id"]
    calls = []

    class FakeSearch:
        def has_cache(self):
            return True

        def search(self, query, *args, **kwargs):
            calls.append((query, kwargs.get("use_llm")))
            if kwargs.get("use_llm"):
                raise RuntimeError("llm unavailable")
            return [meme_id]

    test_client.app.state.search_engine = FakeSearch()
    response = test_client.post("/search", json={"query": "原始查询", "llm_enhance": True})
    assert response.status_code == 200
    assert calls == [("原始查询", True), ("原始查询", False)]


def test_cache_generation_returns_pollable_task(client):
    """缓存生成通过 202 和任务状态接口返回结果。"""
    test_client, _ = client

    class FakeSearch:
        def generate_cache(self, progress):
            progress(1.0, "done")

        def has_cache(self):
            return False

    test_client.app.state.search_engine = FakeSearch()
    response = test_client.post("/generate-cache")
    assert response.status_code == 202
    task_id = response.json()["task_id"]
    for _ in range(50):
        status = test_client.get(f"/tasks/{task_id}").json()
        if status["status"] in {"succeeded", "failed"}:
            break
    assert status["status"] == "succeeded"


def test_parallel_context_batch_writes_independent_database_records_and_incremental_vectors(tmp_path, monkeypatch):
    """并发度为 2 时两张图片独立完成逐图文本向量，且不隐式创建全库缓存任务。"""
    monkeypatch.setenv("MEMEMEOW_DATABASE_URL", os.getenv("MEMEMEOW_TEST_DATABASE_URL", ""))
    monkeypatch.setenv("MEMEMEOW_DATA_ROOT", str(tmp_path / "data"))
    monkeypatch.setenv("MEMEMEOW_IMAGE_ROOT", str(tmp_path / "images"))
    monkeypatch.setenv("EMBEDDING_API_KEY", "")
    monkeypatch.setenv("EMBEDDING_BASE_URL", "")
    monkeypatch.setenv("MEMEMEOW_OPENCODE_MODEL", "")
    monkeypatch.setenv("MEMEMEOW_AGENT_CALLBACK_SECRET", "test-agent-callback-secret-1234")
    monkeypatch.setenv("MEMEMEOW_OPENCODE_CONCURRENCY", "2")
    monkeypatch.setenv("MEMEMEOW_AGENT_SCOPE_CONCURRENCY", "2")
    _clear_test_scope()
    with TestClient(app) as test_client:
        def hold_schedule(_worker, _job_id):
            """在测试替换视觉和 Agent 依赖前暂缓上传产生的 Job。"""
            return None

        original_schedule = ImageProcessingWorker.schedule
        monkeypatch.setattr(ImageProcessingWorker, "schedule", hold_schedule)
        first = test_client.post("/images/upload", files=[("files", ("parallel-a.png", png_bytes("red"), "image/png"))]).json()["results"][0]
        second = test_client.post("/images/upload", files=[("files", ("parallel-b.png", png_bytes("blue"), "image/png"))]).json()["results"][0]
        monkeypatch.setattr(ImageProcessingWorker, "schedule", original_schedule)
        started = threading.Barrier(2)
        sessions: list[str] = []

        def fake_embed(content, *, filename="image"):
            """模拟视觉服务，使并发断言只覆盖任务调度和 Agent 写回。"""
            del content, filename
            settings = test_client.app.state.settings
            return {
                "model": settings.visual_model,
                "dimensions": settings.visual_model_dimensions,
                "preprocess_version": settings.visual_preprocess_version,
                "embedding": [1.0] + [0.0] * (settings.visual_model_dimensions - 1),
            }

        monkeypatch.setattr(test_client.app.state.visual_inference, "embed", fake_embed)

        def fake_text_embedding(text):
            """模拟文本 embedding，使统一 job 可以完成最后阶段。"""
            del text
            return [1.0] + [0.0] * 1023

        monkeypatch.setattr(test_client.app.state.search_engine, "_embedding", fake_text_embedding)

        def fake_run(image, progress, *, task_id=None, **kwargs):
            """模拟两个独立 Agent session，并在 barrier 处确认真正并行。"""
            del progress, kwargs
            assert task_id
            sessions.append(image.name)
            started.wait(timeout=2)
            return {
                "title": image.stem.replace("-", " "),
                "summary": "并发测试",
                "subjects": [image.name],
                "visible_text": [],
                "references": [],
                "meaning": None,
                "keywords": ["测试"],
                "search_queries": [],
                "uncertainties": [],
                "source_urls": [],
            }, f"session-{image.stem}"

        monkeypatch.setattr(test_client.app.state.opencode, "run", fake_run)
        submitted = test_client.post("/images/context/batch", json={"items": [{"meme_id": first["meme_id"]}, {"meme_id": second["meme_id"]}], "include_unready": True})
        assert submitted.status_code == 200
        task_ids = [item["task_id"] for item in submitted.json()["results"]]
        assert len(task_ids) == 2
        with test_client.app.state.database.environment("local") as environment:
            expected_sessions = {
                environment.memes.get(first["meme_id"]).storage_key,
                environment.memes.get(second["meme_id"]).storage_key,
            }
        deadline = time.monotonic() + 5
        while len(sessions) < 2 and time.monotonic() < deadline:
            time.sleep(0.01)
        assert set(sessions) == expected_sessions
        for task_id in task_ids:
            deadline = time.monotonic() + 5
            while time.monotonic() < deadline:
                record = test_client.get(f"/tasks/{task_id}").json()
                if record["status"] in {"succeeded", "failed"}:
                    break
                time.sleep(0.01)
        assert record["status"] == "succeeded"
        assert test_client.get(f"/images/metadata?meme_id={first['meme_id']}").json()["context_status"] == "ready"
        assert test_client.get(f"/images/metadata?meme_id={second['meme_id']}").json()["context_status"] == "ready"
        text_items = test_client.get("/tasks", params={"task_type": "text_embedding_generation"}).json()["items"]
        assert len(text_items) == 2
        assert all(item["status"] == "succeeded" for item in text_items)
        cache_items = test_client.get("/tasks", params={"task_type": "cache_generation"}).json()["items"]
        assert cache_items == []
    _clear_test_scope()


def test_pending_agent_ready_sidecar_and_v4_embedding_pipeline_is_backend_owned(tmp_path, monkeypatch):
    """测试图片从 pending 经 Agent 到 ready/v4 索引，canonical 写回始终由后端掌控。"""
    monkeypatch.setenv("MEMEMEOW_DATABASE_URL", os.getenv("MEMEMEOW_TEST_DATABASE_URL", ""))
    monkeypatch.setenv("MEMEMEOW_DATA_ROOT", str(tmp_path / "data"))
    monkeypatch.setenv("MEMEMEOW_IMAGE_ROOT", str(tmp_path / "images"))
    monkeypatch.setenv("EMBEDDING_API_KEY", "")
    monkeypatch.setenv("EMBEDDING_BASE_URL", "")
    monkeypatch.setenv("MEMEMEOW_OPENCODE_MODEL", "mememeow/gpt-5.6-luna")
    monkeypatch.setenv("MEMEMEOW_AGENT_CALLBACK_SECRET", "test-agent-callback-secret-1234")
    monkeypatch.setenv("MEMEMEOW_OPENCODE_CONCURRENCY", "1")
    _clear_test_scope()


    application_file = Path(__file__).resolve().parent.parent / "api.py"
    application_before = application_file.read_bytes()
    with TestClient(app) as test_client:
        def fake_visual_embed(content, *, filename="image"):
            """模拟视觉服务，使测试只验证逐图任务衔接。"""
            del content, filename
            settings = test_client.app.state.settings
            return {
                "model": settings.visual_model,
                "dimensions": settings.visual_model_dimensions,
                "preprocess_version": settings.visual_preprocess_version,
                "embedding": [1.0] + [0.0] * (settings.visual_model_dimensions - 1),
            }

        def fake_text_embedding(text):
            """模拟 v4/逐图文本 embedding 服务并保留固定维度。"""
            assert text
            return [1.0] + [0.0] * 1023

        def fake_agent_run(image, progress, *, task_id=None, callback_token=None, **kwargs):
            """模拟 Agent 只交付任务 artifact，拒绝把伪造字段写入 canonical 数据。"""
            del progress, kwargs
            assert task_id and callback_token
            assert callback_token != "test-agent-callback-secret-1234"
            binding = test_client.app.state.callback_verifier.verify(callback_token)
            remaining = binding.expires_at - datetime.now(UTC)
            assert timedelta(hours=1, minutes=59) < remaining <= timedelta(hours=2)
            runner = test_client.app.state.opencode
            draft_path, result_path = runner.create_task_result_paths(task_id)
            candidate = {
                "title": "pending agent",
                "summary": "后端拥有 canonical 写回",
                "subjects": ["测试图片"],
                "visible_text": [],
                "references": [],
                "meaning": None,
                "keywords": ["测试"],
                "search_queries": [],
                "uncertainties": [],
                "source_urls": [],
                "canonical_sidecar": "agent cannot set this",
                "application_code": "agent cannot set this",
            }
            # 运行器只接受任务专属结果文件；canonical sidecar 和应用代码不在
            # Agent artifact 路径中，最终写回由 context_handler 的白名单完成。
            draft_path.write_text(json.dumps(candidate, ensure_ascii=False), encoding="utf-8")
            draft_path.replace(result_path)
            # 这里直接返回候选，专门覆盖后端 canonical 写回的字段白名单；结果
            # 文件读取器对未知顶层字段会严格拒绝，该边界由 OpenCode 单测覆盖。
            return candidate, "pending-session"

        monkeypatch.setattr(test_client.app.state.visual_inference, "embed", fake_visual_embed)
        monkeypatch.setattr(test_client.app.state.search_engine, "_embedding", fake_text_embedding)
        monkeypatch.setattr(test_client.app.state.opencode, "run", fake_agent_run)

        uploaded = test_client.post(
            "/images/upload",
            files=[("files", ("pending-agent.png", png_bytes("purple"), "image/png"))],
        )
        assert uploaded.status_code == 200
        result = uploaded.json()["results"][0]
        meme_id = result["meme_id"]
        assert result["metadata_status"] == "pending"
        assert application_file.read_bytes() == application_before

        processing_id = result["processing_job_id"]
        deadline = time.monotonic() + 10
        while time.monotonic() < deadline:
            processing = test_client.get(f"/images/processing/{processing_id}").json()
            if processing["status"] in {"succeeded", "failed", "blocked", "unknown_execution"}:
                break
            time.sleep(0.02)
        assert processing["status"] == "succeeded"
        assert [stage["stage"] for stage in processing["stages"]] == ["visual", "agent", "auto_rename", "text_embedding"]
        assert [stage["status"] for stage in processing["stages"]] == ["succeeded", "succeeded", "skipped", "succeeded"]

        metadata = test_client.get("/images/metadata", params={"meme_id": meme_id})
        assert metadata.status_code == 200
        metadata_payload = metadata.json()
        assert metadata_payload["context_status"] == "ready"
        assert metadata_payload["meme_context"]["title"] == "pending agent"
        assert "canonical_sidecar" not in metadata_payload["meme_context"]
        assert "application_code" not in metadata_payload["meme_context"]
        assert application_file.read_bytes() == application_before

        text_tasks = test_client.get("/tasks", params={"task_type": "text_embedding_generation"}).json()["items"]
        assert len(text_tasks) == 1
        assert text_tasks[0]["status"] == "succeeded"
        cache_submission = test_client.post("/generate-cache")
        assert cache_submission.status_code == 202
        cache_task_id = cache_submission.json()["task_id"]
        deadline = time.monotonic() + 10
        while time.monotonic() < deadline:
            cache_task = test_client.get(f"/tasks/{cache_task_id}").json()
            if cache_task["status"] in {"succeeded", "failed"}:
                break
            time.sleep(0.02)
        assert cache_task["status"] == "succeeded"
        assert cache_task["result"]["indexed_count"] == 1
        assert cache_task["result"]["model"] == test_client.app.state.settings.embedding_model
        assert application_file.read_bytes() == application_before
    _clear_test_scope()


def test_auto_name_pipeline_renames_safely_and_freezes_final_metadata_hash(tmp_path, monkeypatch):
    """自动命名四阶段必须按顺序完成，并让文本 Task 绑定重命名后的 metadata hash。"""
    monkeypatch.setenv("MEMEMEOW_DATABASE_URL", os.getenv("MEMEMEOW_TEST_DATABASE_URL", ""))
    monkeypatch.setenv("MEMEMEOW_DATA_ROOT", str(tmp_path / "data"))
    monkeypatch.setenv("MEMEMEOW_IMAGE_ROOT", str(tmp_path / "images"))
    monkeypatch.setenv("EMBEDDING_API_KEY", "")
    monkeypatch.setenv("EMBEDDING_BASE_URL", "")
    monkeypatch.setenv("MEMEMEOW_OPENCODE_MODEL", "mememeow/gpt-5.6-luna")
    monkeypatch.setenv("MEMEMEOW_AGENT_CALLBACK_SECRET", "test-agent-callback-secret-1234")
    monkeypatch.setenv("MEMEMEOW_OPENCODE_CONCURRENCY", "1")
    _clear_test_scope()
    with TestClient(app) as test_client:
        def fake_visual_embed(content, *, filename="image"):
            """模拟视觉服务，确保测试只验证图片处理控制面。"""
            del content, filename
            settings = test_client.app.state.settings
            return {
                "model": settings.visual_model,
                "dimensions": settings.visual_model_dimensions,
                "preprocess_version": settings.visual_preprocess_version,
                "embedding": [1.0] + [0.0] * (settings.visual_model_dimensions - 1),
            }

        def fake_text_embedding(text_value):
            """模拟固定维度文本向量，避免访问外部 embedding 服务。"""
            assert text_value
            return [1.0] + [0.0] * 1023

        def fake_agent_run(image, progress, *, task_id=None, callback_token=None, **kwargs):
            """交付带非法路径字符的标题，验证服务端安全派生文件名。"""
            del progress, callback_token, kwargs
            assert task_id
            runner = test_client.app.state.opencode
            draft_path, result_path = runner.create_task_result_paths(task_id)
            draft_path.write_text(
                json.dumps(
                    {
                        "title": "Project Launch v2",
                        "summary": "自动命名测试",
                        "subjects": ["测试图片"],
                        "visible_text": [],
                        "references": [],
                        "meaning": None,
                        "keywords": ["测试"],
                        "search_queries": [],
                        "uncertainties": [],
                        "source_urls": [],
                    },
                    ensure_ascii=False,
                ),
                encoding="utf-8",
            )
            draft_path.replace(result_path)
            return runner.read_result_file(result_path), f"session-{image.stem}"

        monkeypatch.setattr(test_client.app.state.visual_inference, "embed", fake_visual_embed)
        monkeypatch.setattr(test_client.app.state.search_engine, "_embedding", fake_text_embedding)
        monkeypatch.setattr(test_client.app.state.opencode, "run", fake_agent_run)

        uploaded = test_client.post(
            "/images/upload",
            data={"auto_name": "true", "reverse_image_policy": "forbid"},
            files=[("files", ("original-name.png", png_bytes("purple"), "image/png"))],
        )
        assert uploaded.status_code == 200
        result = uploaded.json()["results"][0]
        assert result["ok"] is True
        assert result["auto_name"] is True
        assert result["saved_filename"] == "original-name.png"
        job_id = result["processing_job_id"]

        deadline = time.monotonic() + 10
        while time.monotonic() < deadline:
            processing = test_client.get(f"/images/processing/{job_id}").json()
            if processing.get("status") in {"succeeded", "failed", "blocked", "unknown_execution"}:
                break
            time.sleep(0.02)
        assert processing["status"] == "succeeded"
        assert processing["auto_name"] is True
        assert [stage["stage"] for stage in processing["stages"]] == ["visual", "agent", "auto_rename", "text_embedding"]
        assert [stage["status"] for stage in processing["stages"]] == ["succeeded", "succeeded", "succeeded", "succeeded"]
        assert processing["has_warnings"] is False

        metadata = test_client.get("/images/metadata", params={"meme_id": result["meme_id"]}).json()
        assert metadata["image"]["relative_path"] == "Project Launch v2.png"
        assert not (tmp_path / "images" / "original-name.png").exists()

        with test_client.app.state.database.environment("local") as environment:
            meme = environment.memes.get(result["meme_id"])
            assert meme is not None
            assert meme.display_name == "Project Launch v2"
            physical_image = tmp_path / "images" / meme.storage_key
            assert physical_image.exists()
            assert not (tmp_path / "images" / "Project Launch v2.png").exists()
            final_hash = ImageProcessingWorker._metadata_hash(meme)
            assert final_hash is not None
            rename_task = environment.tasks.get(
                next(
                    stage["task_id"]
                    for stage in processing["stages"]
                    if stage["stage"] == "auto_rename"
                )
            )
            assert rename_task is not None
            assert rename_task.status == "succeeded"
            assert "auto_name" not in rename_task.payload
            text_task = environment.uow.session.scalar(
                select(Task).where(
                    Task.scope_id == "local",
                    Task.task_type == "text_embedding_generation",
                    Task.processing_job_id == job_id,
                )
            )
            assert text_task is not None
            assert text_task.payload["metadata_hash"] == final_hash
            old_hash_rows = list(
                environment.uow.session.scalars(
                    select(MemeTextEmbedding).where(
                        MemeTextEmbedding.scope_id == "local",
                        MemeTextEmbedding.meme_id == meme.id,
                        MemeTextEmbedding.metadata_hash == final_hash,
                        MemeTextEmbedding.status == "ready",
                    )
                )
            )
            assert len(old_hash_rows) == 1

    _clear_test_scope()


def test_auto_name_warning_keeps_leaf_failure_and_continues_text_stage(tmp_path, monkeypatch):
    """缺少标题时叶子 Task 必须失败为事实，父 Job 以 warning 继续文本阶段。"""
    monkeypatch.setenv("MEMEMEOW_DATABASE_URL", os.getenv("MEMEMEOW_TEST_DATABASE_URL", ""))
    monkeypatch.setenv("MEMEMEOW_DATA_ROOT", str(tmp_path / "data"))
    monkeypatch.setenv("MEMEMEOW_IMAGE_ROOT", str(tmp_path / "images"))
    monkeypatch.setenv("EMBEDDING_API_KEY", "")
    monkeypatch.setenv("EMBEDDING_BASE_URL", "")
    monkeypatch.setenv("MEMEMEOW_OPENCODE_MODEL", "mememeow/gpt-5.6-luna")
    monkeypatch.setenv("MEMEMEOW_AGENT_CALLBACK_SECRET", "test-agent-callback-secret-1234")
    _clear_test_scope()


    with TestClient(app) as test_client:
        def fake_visual_embed(content, *, filename="image"):
            """模拟视觉向量服务。"""
            del content, filename
            settings = test_client.app.state.settings
            return {
                "model": settings.visual_model,
                "dimensions": settings.visual_model_dimensions,
                "preprocess_version": settings.visual_preprocess_version,
                "embedding": [1.0] + [0.0] * (settings.visual_model_dimensions - 1),
            }

        def fake_text_embedding(text_value):
            """模拟文本向量服务。"""
            assert text_value
            return [1.0] + [0.0] * 1023

        def fake_agent_run(image, progress, *, task_id=None, callback_token=None, **kwargs):
            """交付没有标题但其余字段有效的 Agent 语境。"""
            del image, progress, callback_token, kwargs
            assert task_id
            runner = test_client.app.state.opencode
            draft_path, result_path = runner.create_task_result_paths(task_id)
            draft_path.write_text(
                json.dumps(
                    {
                        "title": None,
                        "summary": "保留原始文件名",
                        "subjects": ["测试图片"],
                        "visible_text": [],
                        "references": [],
                        "meaning": None,
                        "keywords": ["测试"],
                        "search_queries": [],
                        "uncertainties": [],
                        "source_urls": [],
                    },
                    ensure_ascii=False,
                ),
                encoding="utf-8",
            )
            draft_path.replace(result_path)
            return runner.read_result_file(result_path), "warning-session"

        monkeypatch.setattr(test_client.app.state.visual_inference, "embed", fake_visual_embed)
        monkeypatch.setattr(test_client.app.state.search_engine, "_embedding", fake_text_embedding)
        monkeypatch.setattr(test_client.app.state.opencode, "run", fake_agent_run)
        uploaded = test_client.post(
            "/images/upload",
            data={"auto_name": "true"},
            files=[("files", ("keep-original.png", png_bytes("blue"), "image/png"))],
        )
        assert uploaded.status_code == 200
        result = uploaded.json()["results"][0]
        job_id = result["processing_job_id"]

        deadline = time.monotonic() + 10
        while time.monotonic() < deadline:
            processing = test_client.get(f"/images/processing/{job_id}").json()
            if processing.get("status") in {"succeeded", "failed", "blocked", "unknown_execution"}:
                break
            time.sleep(0.02)
        assert processing["status"] == "succeeded"
        assert processing["has_warnings"] is True
        assert processing["warnings"] == [
            {
                "stage": "auto_rename",
                "error": "auto_rename_title_missing",
                "message": "自动重命名未完成",
                "recoverable": True,
            }
        ]
        assert [stage["status"] for stage in processing["stages"]] == ["succeeded", "succeeded", "warning", "succeeded"]
        assert test_client.get("/images/metadata", params={"meme_id": result["meme_id"]}).json()["image"]["relative_path"] == "keep-original.png"
        rename_tasks = test_client.get("/tasks", params={"task_type": "image_auto_rename"}).json()["items"]
        assert len(rename_tasks) == 1
        assert rename_tasks[0]["status"] == "failed"
        assert rename_tasks[0]["error"]["error"] == "auto_rename_title_missing"
        assert rename_tasks[0]["image_stage_recoverable"] is True
        text_tasks = test_client.get("/tasks", params={"task_type": "text_embedding_generation"}).json()["items"]
        assert len(text_tasks) == 1 and text_tasks[0]["status"] == "succeeded"
    _clear_test_scope()


def test_scope_unready_processing_is_option_only_and_fail_closed(client, monkeypatch):
    """scope 级未就绪入口不接受分页筛选，并在能力拒绝前不产生业务副作用。"""
    test_client, _ = client
    # 该测试必须稳定覆盖能力拒绝分支，不能受宿主环境中的 SerpAPI 凭据影响。
    monkeypatch.setattr(test_client.app.state.reverse_image, "_provider_factory", None)
    monkeypatch.setattr(test_client.app.state.reverse_image.settings, "serpapi_api_key", "")

    unavailable = test_client.post(
        "/images/processing/unready",
        json={"reverse_image_policy": "auto", "auto_name": False},
    )
    assert unavailable.status_code == 503
    assert unavailable.json()["error"] == "reverse_image_unavailable"
    with test_client.app.state.database.environment("local") as environment:
        assert environment.uow.session.scalar(select(Meme.id)) is None

    first = test_client.post("/images/upload", files=[("files", ("scope-first.png", png_bytes("red"), "image/png"))]).json()["results"][0]
    second = test_client.post("/images/upload", files=[("files", ("scope-second.png", png_bytes("blue"), "image/png"))]).json()["results"][0]
    invalid_bool = test_client.post(
        "/images/processing/unready",
        json={"reverse_image_policy": "forbid", "auto_name": "true"},
    )
    assert invalid_bool.status_code == 400
    assert invalid_bool.json()["error"] == "invalid_request"

    response = test_client.post(
        "/images/processing/unready",
        json={"reverse_image_policy": "forbid", "auto_name": False},
    )
    assert response.status_code == 202
    body = response.json()
    assert body["target_count"] == 2
    assert {item["meme_id"] for item in body["results"]} == {first["meme_id"], second["meme_id"]}
    assert all(item.get("processing_job_id") or item.get("error") for item in body["results"])

    filtered = test_client.post(
        "/images/processing/unready?page=2",
        json={"reverse_image_policy": "forbid", "auto_name": False},
    )
    assert filtered.status_code == 400
    assert filtered.json()["error"] == "invalid_request"


def test_selected_stage_batch_validates_core_mapping_and_submits_independent_tasks(client):
    """选中重试只接受三个核心阶段，并为每个图片阶段创建独立任务。"""
    test_client, _ = client
    uploaded = test_client.post(
        "/images/upload",
        files=[("files", ("selected-stage.png", png_bytes("green"), "image/png"))],
    ).json()["results"][0]
    meme_id = uploaded["meme_id"]

    invalid = test_client.post(
        "/images/stages/batch",
        json={"items": [{"meme_id": meme_id}], "stages": ["auto_rename"]},
    )
    assert invalid.status_code == 422
    assert invalid.json()["error"] == "invalid_image_stage"

    response = test_client.post(
        "/images/stages/batch",
        json={"items": [{"meme_id": meme_id}], "stages": ["visual", "text_embedding"]},
    )
    assert response.status_code == 202
    body = response.json()
    assert body["target_count"] == 2
    assert {item["stage"] for item in body["results"]} == {"visual", "text_embedding"}
    assert all(item.get("task_id") or item.get("error") for item in body["results"])


def test_standalone_auto_rename_preserves_current_text_embedding_without_creating_job(tmp_path, monkeypatch):
    """内容寻址图片仅更新展示名时保留语义向量，且不能创建父 Job 或文本任务。"""
    monkeypatch.setenv("MEMEMEOW_DATABASE_URL", os.getenv("MEMEMEOW_TEST_DATABASE_URL", ""))
    monkeypatch.setenv("MEMEMEOW_DATA_ROOT", str(tmp_path / "data"))
    monkeypatch.setenv("MEMEMEOW_IMAGE_ROOT", str(tmp_path / "images"))
    monkeypatch.setenv("EMBEDDING_API_KEY", "")
    monkeypatch.setenv("EMBEDDING_BASE_URL", "")
    monkeypatch.setenv("MEMEMEOW_OPENCODE_MODEL", "")
    _clear_test_scope()
    with TestClient(app) as test_client:
        metadata_service = test_client.app.state.metadata
        meme_id, image = metadata_service.upload_bytes(png_bytes("blue"), target_key="standalone-source.png")
        metadata_service.update_context(
            image,
            {"title": "Standalone renamed", "summary": "独立阶段测试"},
            producer="research",
            model="test-model",
            status="ready",
        )
        with test_client.app.state.database.environment("local") as environment:
            record = environment.memes.get(meme_id)
            assert record is not None
            old_metadata_hash = ImageProcessingWorker._metadata_hash(record)
            assert old_metadata_hash is not None
            old_sha256 = record.sha256
            environment.uow.session.add(
                MemeTextEmbedding(
                    scope_id="local",
                    meme_id=record.id,
                    image_sha256=old_sha256,
                    metadata_hash=old_metadata_hash,
                    embedding_model_version=test_client.app.state.settings.embedding_model,
                    dimensions=EMBEDDING_DIMENSIONS,
                    semantic_document="标题：Standalone renamed",
                    embedding=[1.0] + [0.0] * (EMBEDDING_DIMENSIONS - 1),
                    status="ready",
                )
            )
            environment.uow.session.commit()

        submitted = test_client.post(
            "/images/stages",
            json={"meme_id": str(meme_id), "stage": "auto_rename"},
        )
        assert submitted.status_code == 202
        task_id = submitted.json()["task_id"]
        deadline = time.monotonic() + 10
        while time.monotonic() < deadline:
            task = test_client.get(f"/tasks/{task_id}").json()
            if task["status"] in {"succeeded", "failed"}:
                break
            time.sleep(0.02)
        assert task["status"] == "succeeded"

        with test_client.app.state.database.environment("local") as environment:
            record = environment.memes.get(meme_id)
            assert record is not None
            assert record.display_name == "Standalone renamed"
            assert record.storage_key == Path(image).name
            assert Path(image).exists()
            row = environment.uow.session.scalar(
                select(MemeTextEmbedding).where(
                    MemeTextEmbedding.scope_id == "local",
                    MemeTextEmbedding.meme_id == record.id,
                    MemeTextEmbedding.image_sha256 == old_sha256,
                    MemeTextEmbedding.metadata_hash == old_metadata_hash,
                )
            )
            assert row is not None
            # storage_key 和七字段语义 hash 均未变化，展示名不是文本向量输入。
            assert row.status == "ready"
            assert environment.uow.session.scalar(select(Task).where(Task.task_type == "text_embedding_generation")) is None
            assert environment.uow.session.scalar(select(Task).where(Task.processing_job_id.is_not(None))) is None
    _clear_test_scope()


def test_context_target_deleted_during_agent_is_reported_as_target_changed(tmp_path, monkeypatch):
    """Agent 运行期间图片被删除时，任务必须返回稳定的目标变化错误。"""
    monkeypatch.setenv("MEMEMEOW_DATABASE_URL", os.getenv("MEMEMEOW_TEST_DATABASE_URL", ""))
    monkeypatch.setenv("MEMEMEOW_DATA_ROOT", str(tmp_path / "data"))
    monkeypatch.setenv("MEMEMEOW_IMAGE_ROOT", str(tmp_path / "images"))
    monkeypatch.setenv("EMBEDDING_API_KEY", "")
    monkeypatch.setenv("EMBEDDING_BASE_URL", "")
    monkeypatch.setenv("MEMEMEOW_OPENCODE_MODEL", "")
    monkeypatch.setenv("MEMEMEOW_AGENT_CALLBACK_SECRET", "test-agent-callback-secret-1234")
    _clear_test_scope()
    with TestClient(app) as test_client:
        uploaded = test_client.post("/images/upload", files=[("files", ("deleted.png", png_bytes(), "image/png"))]).json()["results"][0]

        def fake_embed(content, *, filename="image"):
            """模拟视觉服务，避免目标变化测试访问真实视觉模型。"""
            del content, filename
            settings = test_client.app.state.settings
            return {
                "model": settings.visual_model,
                "dimensions": settings.visual_model_dimensions,
                "preprocess_version": settings.visual_preprocess_version,
                "embedding": [1.0] + [0.0] * (settings.visual_model_dimensions - 1),
            }

        monkeypatch.setattr(test_client.app.state.visual_inference, "embed", fake_embed)

        def fake_run(path, progress, *, task_id=None, **kwargs):
            """模拟 Agent 完成前目标图片被删除。"""
            del progress, kwargs
            assert task_id
            path.unlink()
            return {}, "deleted-session"

        monkeypatch.setattr(test_client.app.state.opencode, "run", fake_run)
        response = test_client.post("/images/context", json={"meme_id": uploaded["meme_id"]})
        assert response.status_code == 202
        task_id = response.json()["task_id"]
        deadline = time.monotonic() + 5
        while time.monotonic() < deadline:
            status = test_client.get(f"/tasks/{task_id}").json()
            if status["status"] in {"succeeded", "failed"}:
                break
            time.sleep(0.01)
        assert status["status"] == "failed"
        assert status["error"]["error"] == "target_changed"
    _clear_test_scope()