"""Tests for vendor-neutral memory portability — service + admin endpoints.

Covers the contracts that don't depend on a running database:

  * route registration for all six endpoints
  * `list_starter_packs` reads on-disk manifests and returns the expected
    set of bundled packs (statewave-support + 5 demo agents)
  * subject-id validation (length, charset, reserved prefixes)
  * payload validation (format, version, unknown fields, size limit)
  * support reseed targets the configured shared subject and never a
    per-visitor `demo_web_*__statewave-support` id
  * clone refuses an empty / non-existent source
  * export rejects an empty `subject_ids` list and oversize requests
  * import rejects malformed payloads with clear error messages

Tests that exercise the actual database write path live in
`tests/integration/` because they need a real Postgres + asyncpg session
factory; these unit tests focus on the validation/dispatch contract.
"""

from __future__ import annotations

from pathlib import Path
from unittest.mock import AsyncMock, MagicMock

import pytest

from server.app import create_app
from server.core.config import settings
from server.services import memory_packs as mp
from tests.conftest import iter_routes


# ─── Route registration ─────────────────────────────────────────────────────


@pytest.mark.parametrize(
    "path, methods",
    [
        ("/admin/memory/starter-packs", {"GET"}),
        ("/admin/memory/starter-packs/import", {"POST"}),
        ("/admin/memory/support/reseed", {"POST"}),
        ("/admin/memory/clone", {"POST"}),
        ("/admin/memory/export", {"POST"}),
        ("/admin/memory/import", {"POST"}),
    ],
)
def test_memory_endpoints_registered(path, methods):
    app = create_app()
    for route in iter_routes(app):
        if route.path == path:
            assert methods.issubset(route.methods), (
                f"{path} expected {methods}, got {route.methods}"
            )
            return
    pytest.fail(f"{path} not registered on the app")


def test_docs_pack_reseed_alias_is_present_for_back_compat():
    """`/admin/docs-pack/reseed` is preserved as a deprecated alias.

    The old vendor-locked implementation (which dispatched a GitHub Actions
    workflow) is gone. The path is kept as a backward-compatible alias that
    delegates to the new vendor-neutral support reseed service so legacy
    operator scripts keep working without GitHub tokens.
    """
    app = create_app()
    aliases = [r for r in iter_routes(app) if r.path == "/admin/docs-pack/reseed"]
    assert aliases, "/admin/docs-pack/reseed alias must be registered"
    route = aliases[0]
    assert "POST" in route.methods
    # FastAPI surfaces deprecated routes through the OpenAPI schema — assert
    # the alias is flagged so generated clients / docs warn callers.
    schema = app.openapi()
    op = schema["paths"]["/admin/docs-pack/reseed"]["post"]
    assert op.get("deprecated") is True


# ─── Starter pack registry ──────────────────────────────────────────────────


def test_list_starter_packs_returns_bundled_set():
    packs = mp.list_starter_packs()
    pack_ids = {p["pack_id"] for p in packs}
    # Demo pack ids carry the `demo-` prefix so an import lands on the same
    # subject id the marketing-widget demo flow already uses on
    # statewave.ai (e.g. `demo-support-agent`). Keeping them aligned means
    # an operator-imported pack is immediately picked up by the live demo
    # without an extra rename step.
    expected = {
        "statewave-support-agent",
        "demo-support-agent",
        "demo-coding-assistant",
        "demo-sales-copilot",
        "demo-devops-agent",
        "demo-research-assistant",
    }
    assert expected.issubset(pack_ids), (
        f"missing packs: {expected - pack_ids}"
    )


def test_starter_pack_metadata_shape():
    """Every bundled pack must carry the spec-required manifest fields."""
    packs = mp.list_starter_packs()
    required = {
        "pack_id",
        "display_name",
        "description",
        "version",
        "subject_id_suggestion",
        "episode_count",
        "memory_count",
    }
    for pack in packs:
        if "__error" in pack:
            pytest.fail(f"pack {pack['pack_id']} failed to load: {pack['__error']}")
        missing = required - set(pack.keys())
        assert not missing, f"{pack['pack_id']} missing fields: {missing}"
        assert pack["episode_count"] >= 0
        assert pack["memory_count"] >= 0


def test_starter_pack_directory_layout_on_disk():
    """Each pack's directory contains manifest.json + episodes.jsonl + memories.jsonl."""
    root = Path(__file__).resolve().parent.parent / "server" / "starter_packs"
    for pack in mp.list_starter_packs():
        d = root / pack["pack_id"]
        assert (d / "manifest.json").exists(), f"missing manifest for {pack['pack_id']}"
        # JSONL files may be empty for stub packs but must exist
        assert (d / "episodes.jsonl").exists(), f"missing episodes.jsonl for {pack['pack_id']}"
        assert (d / "memories.jsonl").exists(), f"missing memories.jsonl for {pack['pack_id']}"


# ─── __error sanitization (CodeQL alert #1, py/stack-trace-exposure) ────────
#
# `list_starter_packs` returns `__error: <code>` for packs whose manifest
# can't be read. The pre-fix code did `__error: str(exc)`, leaking
# whatever the exception message contained (filesystem paths, JSON parser
# positions, future error-message text). These tests pin the contract
# that the exposed value is a stable, sanitized code regardless of which
# exception path fired.


def _stub_index_with_one_pack(monkeypatch, tmp_path):
    """Replace _read_index() and _PACKS_ROOT so a synthetic pack dir
    drives list_starter_packs without touching the bundled corpus."""
    monkeypatch.setattr(mp, "_PACKS_ROOT", tmp_path)
    monkeypatch.setattr(
        mp,
        "_read_index",
        lambda: [{"pack_id": "synthetic-pack", "kind": "demo_agent", "directory": "synthetic-pack"}],
    )
    pack_dir = tmp_path / "synthetic-pack"
    pack_dir.mkdir()
    return pack_dir


def test_list_starter_packs_returns_sanitized_code_when_manifest_missing(
    monkeypatch, tmp_path
):
    """Missing manifest → `__error: 'manifest_unreadable'`, no path leak."""
    _stub_index_with_one_pack(monkeypatch, tmp_path)
    # No manifest.json written — _read_starter_pack_manifest raises
    # StarterPackError with the directory name in the message.

    packs = mp.list_starter_packs()
    assert len(packs) == 1
    p = packs[0]
    assert p["__error"] == "manifest_unreadable", (
        f"expected sanitized code, got {p['__error']!r} — likely leaking exception message"
    )
    # Belt-and-suspenders: the directory name (which the StarterPackError
    # message DID contain) must not appear anywhere in the payload.
    assert "synthetic-pack" not in p["__error"]


def test_list_starter_packs_handles_corrupt_json_without_leaking(
    monkeypatch, tmp_path
):
    """Pre-fix bug: json.JSONDecodeError bubbled past the narrow
    `except StarterPackError` and broke the soft-fail contract; if it
    HAD been caught, the message would have included the parser line/
    column. Now: caught by the widened handler AND sanitized."""
    pack_dir = _stub_index_with_one_pack(monkeypatch, tmp_path)
    (pack_dir / "manifest.json").write_text("{ this is not valid json", encoding="utf-8")

    packs = mp.list_starter_packs()
    assert len(packs) == 1
    p = packs[0]
    assert p["__error"] == "manifest_unreadable"
    assert p["pack_id"] == "synthetic-pack"


def test_list_starter_packs_handles_oserror_without_leaking(
    monkeypatch, tmp_path
):
    """Pre-fix bug: an unreadable manifest (e.g. permissions denied)
    raised OSError, which slipped past the narrow except clause.
    Now caught and sanitized."""
    pack_dir = _stub_index_with_one_pack(monkeypatch, tmp_path)
    # Create manifest.json as a directory so .read_text() raises
    # IsADirectoryError (an OSError subclass) — portable across OS,
    # avoids relying on chmod which behaves differently when running
    # as root in a container.
    (pack_dir / "manifest.json").mkdir()

    packs = mp.list_starter_packs()
    assert len(packs) == 1
    p = packs[0]
    assert p["__error"] == "manifest_unreadable"


def test_list_starter_packs_logs_full_detail_server_side(
    monkeypatch, tmp_path, caplog
):
    """Operator-facing diagnostic info must still be available — just
    via the server log, not the API response."""
    pack_dir = _stub_index_with_one_pack(monkeypatch, tmp_path)
    (pack_dir / "manifest.json").write_text("{ broken", encoding="utf-8")

    import logging

    with caplog.at_level(logging.WARNING):
        mp.list_starter_packs()

    # Note: structlog routes through stdlib logging; caplog captures the
    # event name. The full exception chain is attached via exc_info.
    assert any(
        "starter_pack_manifest_unreadable" in record.getMessage()
        or "starter_pack_manifest_unreadable" in str(record.__dict__)
        for record in caplog.records
    ), (
        f"expected 'starter_pack_manifest_unreadable' log event; "
        f"saw: {[r.getMessage() for r in caplog.records]}"
    )


# ─── Subject-id validation ──────────────────────────────────────────────────


def test_validate_subject_id_accepts_valid_ids():
    for ok in ("user-1", "tenant-a:agent-7", "default-support-agent", "support_v2.1"):
        assert mp._validate_subject_id(ok) == ok


@pytest.mark.parametrize(
    "bad",
    [
        "",  # empty
        "x" * 257,  # too long (>256 — unified subject-id contract, #121)
        "has spaces",  # invalid char
        "weird/path",  # path separator
        "demo_web_abc",  # reserved prefix used by visitor subjects
    ],
)
def test_validate_subject_id_rejects_invalid_ids(bad):
    with pytest.raises(mp.StarterPackError):
        mp._validate_subject_id(bad)


# ─── Support reseed targets only the shared subject ─────────────────────────


def test_support_subject_target_is_shared_not_per_visitor():
    """Sanity check on the configured target id used by reseed."""
    sid = settings.support_subject_id
    # Must NOT collide with the per-visitor namespace used by the marketing
    # widget's hybrid Statewave Support flow.
    assert not sid.startswith("demo_web_")
    assert "__statewave-support" not in sid
    assert sid == "statewave-support-docs"


# ─── Payload validation (no DB needed) ──────────────────────────────────────


@pytest.mark.asyncio
async def test_import_rejects_non_dict_payload():
    with pytest.raises(mp.StarterPackError) as ei:
        await mp.import_memory_payload(payload=[], target_tenant_id=None)
    assert "JSON object" in str(ei.value)


@pytest.mark.asyncio
async def test_import_rejects_wrong_format():
    with pytest.raises(mp.StarterPackError) as ei:
        await mp.import_memory_payload(
            payload={"format": "something-else", "format_version": 1},
            target_tenant_id=None,
        )
    assert "format" in str(ei.value).lower()


@pytest.mark.asyncio
async def test_import_rejects_wrong_format_version():
    with pytest.raises(mp.StarterPackError) as ei:
        await mp.import_memory_payload(
            payload={
                "format": mp.PAYLOAD_FORMAT,
                "format_version": 99,
            },
            target_tenant_id=None,
        )
    assert "format_version" in str(ei.value)


@pytest.mark.asyncio
async def test_import_rejects_unknown_top_level_fields():
    with pytest.raises(mp.StarterPackError) as ei:
        await mp.import_memory_payload(
            payload={
                "format": mp.PAYLOAD_FORMAT,
                "format_version": mp.PAYLOAD_FORMAT_VERSION,
                "subjects": [],
                "episodes": [],
                "memories": [],
                "evil": "drop_table",
            },
            target_tenant_id=None,
        )
    assert "unknown fields" in str(ei.value)


@pytest.mark.asyncio
async def test_import_rejects_oversize_payload(monkeypatch):
    """Hard byte cap protects the worker from pathological payloads."""
    monkeypatch.setattr(settings, "memory_import_max_bytes", 1024)  # tiny cap
    big_blob = "x" * 4096
    payload = {
        "format": mp.PAYLOAD_FORMAT,
        "format_version": mp.PAYLOAD_FORMAT_VERSION,
        "subjects": [{"original_subject_id": "s", "metadata": {"blob": big_blob}}],
        "episodes": [],
        "memories": [],
    }
    with pytest.raises(mp.StarterPackError) as ei:
        await mp.import_memory_payload(payload=payload, target_tenant_id=None)
    assert ei.value.status_code == 413


# ─── Export validation ──────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_export_rejects_empty_subject_ids():
    with pytest.raises(mp.StarterPackError) as ei:
        await mp.export_memory_payload(subject_ids=[], tenant_id=None)
    assert "required" in str(ei.value)


@pytest.mark.asyncio
async def test_export_rejects_too_many_subjects(monkeypatch):
    monkeypatch.setattr(settings, "memory_import_max_subjects", 2)
    with pytest.raises(mp.StarterPackError) as ei:
        await mp.export_memory_payload(
            subject_ids=["a", "b", "c"], tenant_id=None
        )
    assert ei.value.status_code == 413


@pytest.mark.asyncio
async def test_export_rejects_invalid_subject_id():
    # `demo_web_*` is reserved as a TARGET prefix but allowed as a SOURCE
    # (export is read-only, operators legitimately need to back up visitor
    # subjects). Use a genuinely malformed id here instead.
    with pytest.raises(mp.StarterPackError):
        await mp.export_memory_payload(subject_ids=["has spaces"], tenant_id=None)


# ─── Clone validation ───────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_clone_rejects_invalid_source_id():
    with pytest.raises(mp.StarterPackError):
        await mp.clone_subject(
            source_subject_id="has spaces",
            target_subject_id=None,
            target_display_name=None,
            target_tenant_id=None,
        )


@pytest.mark.asyncio
async def test_clone_rejects_reserved_prefix_TARGET_id():
    """The reserved-prefix guard still fires on the target side."""
    with pytest.raises(mp.StarterPackError) as ei:
        await mp.clone_subject(
            source_subject_id="user-1",
            target_subject_id="demo_web_abc",
            target_display_name=None,
            target_tenant_id=None,
        )
    assert "reserved prefix" in str(ei.value)


def test_clone_allows_reserved_prefix_SOURCE_id_in_validation():
    """Source-side validation accepts the demo_web_ prefix so operators can
    clone visitor memory subjects for inspection. We pin the validator
    directly because hitting the full clone path requires a real DB."""
    sid = "demo_web_abc__support-agent"
    # default — refuses
    with pytest.raises(mp.StarterPackError):
        mp._validate_subject_id(sid, field="source_subject_id")
    # allow_reserved=True — accepts
    assert mp._validate_subject_id(sid, field="source_subject_id", allow_reserved=True) == sid


def test_suggest_target_id_rewrites_reserved_prefix():
    """Importing a .swmem captured from a visitor subject must NOT land the
    new subject back in the demo_web_ namespace. The rewriter swaps the
    reserved prefix for `imported-` so the result is namespace-safe.
    """
    new_id = mp._suggest_target_subject_id("demo_web_abc__support-agent")
    assert new_id.startswith("imported-abc__support-agent-")
    assert not new_id.startswith("demo_web_")


@pytest.mark.asyncio
async def test_import_accepts_reserved_prefix_in_original_subject_id_during_validation():
    """The payload validator must not reject an archive whose subjects'
    `original_subject_id` carry the reserved prefix — that's the common
    case for archives generated from the marketing widget. We exercise the
    validator path explicitly (not the full ingest, which needs a DB)."""
    # An ill-formed id should still raise (this part of validation is
    # unaffected by the carve-out).
    with pytest.raises(mp.StarterPackError):
        await mp.import_memory_payload(
            payload={
                "format": mp.PAYLOAD_FORMAT,
                "format_version": mp.PAYLOAD_FORMAT_VERSION,
                "subjects": [{"original_subject_id": "has spaces"}],
                "episodes": [],
                "memories": [],
            },
            target_tenant_id=None,
        )


# ─── Starter pack import error: unknown pack ────────────────────────────────


@pytest.mark.asyncio
async def test_starter_pack_import_unknown_pack_404():
    with pytest.raises(mp.StarterPackError) as ei:
        await mp.import_starter_pack(
            pack_id="not-a-real-pack",
            target_subject_id=None,
            target_display_name=None,
            target_tenant_id=None,
        )
    assert ei.value.status_code == 404


# ─── HTTP-level checks via the test client ──────────────────────────────────


@pytest.mark.asyncio
async def test_http_starter_packs_list_endpoint(client):
    resp = await client.get("/admin/memory/starter-packs")
    assert resp.status_code == 200
    body = resp.json()
    assert "packs" in body
    pack_ids = {p["pack_id"] for p in body["packs"]}
    assert "statewave-support-agent" in pack_ids
    assert "demo-coding-assistant" in pack_ids


@pytest.mark.asyncio
async def test_http_import_rejects_unknown_format(client):
    resp = await client.post(
        "/admin/memory/import",
        json={"payload": {"format": "wrong", "format_version": 1}},
    )
    assert resp.status_code == 400
    assert "format" in resp.json()["error"]["message"].lower()


@pytest.mark.asyncio
async def test_http_starter_pack_import_unknown_pack_404(client):
    resp = await client.post(
        "/admin/memory/starter-packs/import",
        json={"pack_id": "not-a-real-pack"},
    )
    assert resp.status_code == 404


@pytest.mark.asyncio
async def test_http_clone_rejects_invalid_source_subject(client):
    resp = await client.post(
        "/admin/memory/clone",
        json={"source_subject_id": "has spaces"},
    )
    assert resp.status_code == 400


@pytest.mark.asyncio
async def test_http_export_rejects_empty_subject_list(client):
    resp = await client.post(
        "/admin/memory/export",
        json={"subject_ids": []},
    )
    assert resp.status_code == 400


# ─── Backward-compatible /admin/docs-pack/reseed alias ──────────────────────


@pytest.mark.asyncio
async def test_http_docs_pack_reseed_alias_delegates_to_support_reseed(
    client, monkeypatch
):
    """The legacy alias must call the new vendor-neutral service.

    Pins three properties old operator scripts depend on:
      * `/admin/docs-pack/reseed` is reachable (200, not 404)
      * it dispatches to `reseed_support_subject` — the same code path the
        new `/admin/memory/support/reseed` endpoint uses
      * the legacy `reason` body field is forwarded so callers that already
        emit it keep working
    """
    captured: dict[str, object] = {}

    async def fake_reseed(*, reason=None, force=False):
        captured["reason"] = reason
        return {
            "subject_id": settings.support_subject_id,
            "pack_id": settings.support_starter_pack_id,
            "pack_version": "1.0.0",
            "imported_episodes": 4,
            "imported_memories": 1,
            "reseeded_at": "2026-05-03T10:00:00+00:00",
            "reason": reason,
        }

    monkeypatch.setattr(mp, "reseed_support_subject", fake_reseed)

    resp = await client.post(
        "/admin/docs-pack/reseed",
        json={"reason": "legacy operator script"},
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()

    # Service was actually invoked with the legacy `reason` field.
    assert captured["reason"] == "legacy operator script"

    # Response shape covers the fields old callers checked.
    assert body["subject_id"] == settings.support_subject_id
    assert body["imported_episodes"] == 4
    assert body["imported_memories"] == 1
    assert "reseeded_at" in body


@pytest.mark.asyncio
async def test_http_docs_pack_reseed_alias_works_without_body(client, monkeypatch):
    """Old scripts sometimes POSTed with no body — must still 200."""
    async def fake_reseed(*, reason=None, force=False):
        return {
            "subject_id": settings.support_subject_id,
            "pack_id": settings.support_starter_pack_id,
            "pack_version": "1.0.0",
            "imported_episodes": 0,
            "imported_memories": 0,
            "reseeded_at": "2026-05-03T10:00:00+00:00",
            "reason": None,
        }

    monkeypatch.setattr(mp, "reseed_support_subject", fake_reseed)

    resp = await client.post("/admin/docs-pack/reseed")
    assert resp.status_code == 200


@pytest.mark.asyncio
async def test_http_docs_pack_reseed_alias_does_not_require_github_config(
    client, monkeypatch
):
    """Vendor-neutral guarantee: the alias must not touch any GitHub plumbing.

    The pre-rewrite implementation dispatched a workflow via the GitHub API
    and required `GITHUB_TOKEN` / `DOCS_REFRESH_*` settings. The new alias
    must work with none of those configured. We assert two things:
      * the request returns 200 even when no GitHub-related env exists
      * the underlying service is the new vendor-neutral one (we observe
        this by patching it and confirming the patched function ran)
    """
    # Strip every GitHub-shaped env var that could possibly be probed.
    for var in [
        "GITHUB_TOKEN",
        "DOCS_REFRESH_GITHUB_TOKEN",
        "DOCS_REFRESH_GITHUB_REPO",
        "DOCS_REFRESH_GITHUB_WORKFLOW",
    ]:
        monkeypatch.delenv(var, raising=False)

    called = {"n": 0}

    async def fake_reseed(*, reason=None, force=False):
        called["n"] += 1
        return {
            "subject_id": settings.support_subject_id,
            "pack_id": settings.support_starter_pack_id,
            "pack_version": "1.0.0",
            "imported_episodes": 1,
            "imported_memories": 0,
            "reseeded_at": "2026-05-03T10:00:00+00:00",
            "reason": reason,
        }

    monkeypatch.setattr(mp, "reseed_support_subject", fake_reseed)

    resp = await client.post("/admin/docs-pack/reseed", json={})
    assert resp.status_code == 200
    assert called["n"] == 1


@pytest.mark.asyncio
async def test_http_docs_pack_reseed_alias_requires_admin_auth(monkeypatch):
    """Alias must sit behind the same `X-API-Key` gate as every other admin route."""
    from httpx import ASGITransport, AsyncClient

    monkeypatch.setattr(settings, "api_key", "test-admin-secret")

    # Rebuild the app so the middleware picks up the configured key.
    app = create_app()
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        # No header → 401.
        resp = await ac.post("/admin/docs-pack/reseed", json={})
        assert resp.status_code == 401
        # Wrong key → 403.
        resp = await ac.post(
            "/admin/docs-pack/reseed",
            json={},
            headers={"X-API-Key": "nope"},
        )
        assert resp.status_code == 403


@pytest.mark.asyncio
async def test_http_docs_pack_reseed_alias_targets_only_shared_subject(
    client, monkeypatch
):
    """The alias targets `support_subject_id` only — never a visitor subject."""
    seen: dict[str, str] = {}

    async def fake_reseed(*, reason=None, force=False):
        # Pin: the service does not accept a target arg from the request,
        # so the alias cannot be pointed at a per-visitor subject by a
        # client. Returning the configured shared id is the only path.
        seen["target"] = settings.support_subject_id
        return {
            "subject_id": settings.support_subject_id,
            "pack_id": settings.support_starter_pack_id,
            "pack_version": "1.0.0",
            "imported_episodes": 0,
            "imported_memories": 0,
            "reseeded_at": "2026-05-03T10:00:00+00:00",
            "reason": reason,
        }

    monkeypatch.setattr(mp, "reseed_support_subject", fake_reseed)

    # Client tries to redirect the reseed at a visitor subject — body ignored.
    resp = await client.post(
        "/admin/docs-pack/reseed",
        json={
            "subject_id": "demo_web_attacker__statewave-support",
            "target_subject_id": "demo_web_attacker__statewave-support",
        },
    )
    assert resp.status_code == 200
    body = resp.json()
    assert seen["target"] == settings.support_subject_id
    assert body["subject_id"] == settings.support_subject_id
    assert not body["subject_id"].startswith("demo_web_")


# ─── Versioned payload format constants ─────────────────────────────────────


def test_payload_format_constants_pinned():
    """Format identifiers are part of the on-the-wire contract — locked."""
    assert mp.PAYLOAD_FORMAT == "statewave-memory-payload"
    assert mp.PAYLOAD_FORMAT_VERSION == 1
    assert mp.STARTER_PACK_FORMAT == "statewave-starter-pack"


# ─── Embedding scheduler in _ingest_records_async ─────────────────────────
#
# Regression guards for the import path's embedding behaviour. Before the
# patch that added a post-commit `schedule_embedding_backfill` call, a
# fresh starter-pack import or operator `.swmem` import would land all
# memories with `embedding=NULL` (the bundled JSONL ships without
# embeddings) and silently break vector retrieval. These tests pin the
# fix without needing a real Postgres — the DB session is fully mocked.


def _make_fake_session_factory():
    """Build a session_factory() callable that hands out an awaitable-async-ctx
    session whose .add / .execute / .commit are no-ops. The mock is shaped to
    satisfy `async with engine_module.get_session_factory()() as session:`.
    """
    fake_session = AsyncMock()
    fake_session.add = MagicMock()
    fake_session.execute = AsyncMock()
    fake_session.commit = AsyncMock()
    fake_session_ctx = MagicMock()
    fake_session_ctx.__aenter__ = AsyncMock(return_value=fake_session)
    fake_session_ctx.__aexit__ = AsyncMock(return_value=None)
    return MagicMock(return_value=fake_session_ctx)


@pytest.mark.asyncio
async def test_ingest_schedules_embedding_for_memories_without_embeddings(monkeypatch):
    """Bundled starter packs ship without embeddings — they must be
    auto-scheduled for embedding generation after insert. This is the
    behaviour that was missing before the core patch and that broke
    vector retrieval on every clean reseed."""
    monkeypatch.setattr(
        "server.db.engine.get_session_factory", _make_fake_session_factory
    )
    captured: list[tuple[list, list]] = []

    def fake_scheduler(ids, texts):
        captured.append((list(ids), list(texts)))
        return None

    monkeypatch.setattr(
        "server.services.embeddings.backfill.schedule_embedding_backfill",
        fake_scheduler,
    )

    await mp._ingest_records_async(
        target_subject_id="test-subject",
        target_tenant_id=None,
        episodes_data=[],
        memories_data=[
            {"kind": "fact", "content": "Statewave uses PostgreSQL with pgvector."},
            {"kind": "procedure", "content": "Run docker compose up -d to start."},
        ],
        extra_provenance={},
        extra_metadata={},
    )

    assert len(captured) == 1, "scheduler should fire exactly once after commit"
    ids, texts = captured[0]
    assert len(ids) == 2
    assert sorted(texts) == [
        "Run docker compose up -d to start.",
        "Statewave uses PostgreSQL with pgvector.",
    ]


@pytest.mark.asyncio
async def test_ingest_skips_scheduling_when_embedding_already_present(monkeypatch):
    """`.swmem` archives can carry pre-computed embeddings — re-embedding
    them would burn provider tokens for no benefit. Only memories
    arriving with `embedding=None` should be scheduled."""
    monkeypatch.setattr(
        "server.db.engine.get_session_factory", _make_fake_session_factory
    )
    captured_texts: list[str] = []

    def fake_scheduler(ids, texts):
        captured_texts.extend(texts)
        return None

    monkeypatch.setattr(
        "server.services.embeddings.backfill.schedule_embedding_backfill",
        fake_scheduler,
    )

    pre_embedded = [0.1] * 4  # length doesn't matter for this test
    await mp._ingest_records_async(
        target_subject_id="test-subject",
        target_tenant_id=None,
        episodes_data=[],
        memories_data=[
            {"kind": "fact", "content": "Already embedded.", "embedding": pre_embedded},
            {"kind": "fact", "content": "Needs an embedding."},
        ],
        extra_provenance={},
        extra_metadata={},
    )

    assert captured_texts == ["Needs an embedding."]


@pytest.mark.asyncio
async def test_ingest_does_not_schedule_when_no_memories(monkeypatch):
    """Episode-only imports must not schedule embedding work."""
    monkeypatch.setattr(
        "server.db.engine.get_session_factory", _make_fake_session_factory
    )
    called = False

    def fake_scheduler(ids, texts):
        nonlocal called
        called = True
        return None

    monkeypatch.setattr(
        "server.services.embeddings.backfill.schedule_embedding_backfill",
        fake_scheduler,
    )

    await mp._ingest_records_async(
        target_subject_id="test-subject",
        target_tenant_id=None,
        episodes_data=[
            {"id": "00000000-0000-4000-8000-000000000001", "source": "test", "type": "imported", "payload": {"text": "x"}}
        ],
        memories_data=[],
        extra_provenance={},
        extra_metadata={},
    )

    assert called is False
