"""Notes + timeline + media domain — behavior tests.

Three layers:

- pure-unit media/router tests (run anywhere, incl. host),
- read tests against the migrated v1 snapshot (db_session fixture — skip
  without DATABASE_URL, run in-container),
- write tests wrapped in the always-rollback db_session savepoint.

Migrated-data facts asserted below (frozen snapshot, 2026-07-02): 187
field_notes; 168 bodied → 168 derived notes; field note #176 body starts
"Planted the full spring 2026" with primary target planting #60; area #32
"Raspberry Patch" (parent chain via #22, station #18, multi-species subtree
including species #5 "Raspberry"); years table contains 2026.
"""
from __future__ import annotations

import asyncio
import io
import os

import pytest

GARDEN = 1


# ---------------------------------------------------------------------------
# media service (pure unit — tmp data dir)
# ---------------------------------------------------------------------------

@pytest.fixture()
def media_dir(tmp_path, monkeypatch):
    from app.config import get_settings

    monkeypatch.setenv("DATA_DIR", str(tmp_path))
    get_settings.cache_clear()
    yield tmp_path
    get_settings.cache_clear()


def _upload(data: bytes, filename: str, content_type: str):
    from fastapi import UploadFile
    from starlette.datastructures import Headers

    return UploadFile(
        file=io.BytesIO(data),
        filename=filename,
        headers=Headers({"content-type": content_type}),
    )


def test_safe_note_path_rejects_traversal(media_dir):
    from app.services import media

    # A real file to prove the happy path works.
    rel = media.save_upload(_upload(b"jpegdata", "p.jpg", "image/jpeg"), "photo")
    assert media.safe_note_path(rel)  # resolves
    assert rel.count("/") == 1 and rel.endswith(".jpg")  # YYYY-MM/<uuid>.jpg

    # Everything unsafe resolves to None.
    assert media.safe_note_path("") is None
    assert media.safe_note_path("/etc/passwd") is None
    assert media.safe_note_path("../secrets.txt") is None
    assert media.safe_note_path("2026-07/../../etc/passwd") is None
    assert media.safe_note_path("..") is None
    assert media.safe_note_path("2026-07/..") is None
    # Missing file is also None, not an exception.
    assert media.safe_note_path("2099-01/nope.jpg") is None


def test_save_upload_extension_and_size_cap(media_dir, monkeypatch):
    from app.errors import AppError
    from app.services import media

    # Allowed filename extension wins.
    rel = media.save_upload(_upload(b"m4adata", "memo.M4A", "audio/mp4"), "audio")
    assert rel.endswith(".m4a")

    # No usable filename ext: content-type guess if the platform's mime DB
    # yields an allowed ext, else the kind default (slim images have no
    # /etc/mime.types, so both outcomes are legitimate v1 behavior).
    import mimetypes

    guessed = (mimetypes.guess_extension("audio/ogg") or "").lower()
    expected = guessed if guessed in media.ALLOWED_AUDIO_EXT else ".webm"
    rel = media.save_upload(_upload(b"oggdata", "blob", "audio/ogg"), "audio")
    assert rel.endswith(expected)

    # Disallowed extension + unknown type falls back to the kind default.
    rel2 = media.save_upload(_upload(b"x", "evil.php", "application/x-php"), "photo")
    assert rel2.endswith(".jpg")

    # Over-cap upload is rejected AND the partial file is removed.
    monkeypatch.setattr(media, "MAX_IMAGE_BYTES", 4)
    with pytest.raises(AppError) as exc:
        media.save_upload(_upload(b"12345678", "big.png", "image/png"), "photo")
    assert exc.value.code == media.MEDIA_TOO_LARGE
    assert exc.value.http_status == 413
    month_dir = os.path.join(media.notes_dir(), sorted(os.listdir(media.notes_dir()))[0])
    assert not [f for f in os.listdir(month_dir) if f.endswith(".png")]


def test_delete_note_file_roundtrip(media_dir):
    from app.services import media

    rel = media.save_upload(_upload(b"data", "a.webm", "audio/webm"), "audio")
    abs_path = media.safe_note_path(rel)
    assert abs_path and os.path.isfile(abs_path)
    media.delete_note_file(rel)
    assert media.safe_note_path(rel) is None
    media.delete_note_file(rel)  # idempotent, silent
    media.delete_note_file("../../etc/passwd")  # unsafe path: silent no-op


# ---------------------------------------------------------------------------
# routers (pure unit — registration + async contract)
# ---------------------------------------------------------------------------

def test_router_routes_registered():
    from app.routers import media as media_r
    from app.routers import notes as notes_r
    from app.routers import timeline as timeline_r

    def pairs(router):
        return {(r.path, m) for r in router.routes for m in r.methods}

    notes = pairs(notes_r.router)
    for expected in [
        ("/api/notes", "POST"),
        ("/api/notes", "GET"),
        ("/api/notes/queue", "GET"),
        ("/api/notes/inferred-targets", "GET"),
        ("/api/notes/{note_id}", "PATCH"),
        ("/api/notes/{note_id}", "DELETE"),
        ("/api/notes/{note_id}/transcribe", "POST"),
        ("/api/notes/comments", "POST"),
        ("/api/notes/comments/{comment_id}", "PATCH"),
        ("/api/notes/comments/{comment_id}", "DELETE"),
        ("/api/notes/comments/{comment_id}/targets", "POST"),
        ("/api/notes/comments/{comment_id}/targets/{target_type}/{target_id}", "DELETE"),
    ]:
        assert expected in notes, expected

    tl = pairs(timeline_r.router)
    assert ("/api/timeline", "GET") in tl
    assert ("/api/timeline/{entity_type}/{entity_id}", "GET") in tl
    assert ("/api/timeline/overview-history/{entity_type}/{entity_id}", "GET") in tl

    assert ("/notes/{rel_path:path}", "GET") in pairs(media_r.router)

    # Literal paths must be registered before /{note_id} (route order).
    notes_paths = [r.path for r in notes_r.router.routes]
    assert notes_paths.index("/api/notes/queue") < notes_paths.index(
        "/api/notes/{note_id}"
    )
    assert notes_paths.index("/api/notes/comments") < notes_paths.index(
        "/api/notes/{note_id}"
    )


def test_transcribe_is_the_only_async_route():
    """Per standards: sync def for DB routes, async ONLY for the AI route."""
    from fastapi.routing import APIRoute

    from app.routers import media as media_r
    from app.routers import notes as notes_r
    from app.routers import timeline as timeline_r

    for router in (notes_r.router, timeline_r.router, media_r.router):
        for r in router.routes:
            if not isinstance(r, APIRoute):
                continue
            is_async = asyncio.iscoroutinefunction(r.endpoint)
            if r.path.endswith("/transcribe"):
                assert is_async, r.path
            else:
                assert not is_async, r.path


def test_parse_secondaries_skips_malformed():
    from app.routers.notes import _parse_secondaries

    assert _parse_secondaries(
        ["area:5", "bogus", "nope:1", "species:x", "plant:9"]
    ) == [("area", 5), ("plant", 9)]


# ---------------------------------------------------------------------------
# migrated-data reads (DB — skip on host)
# ---------------------------------------------------------------------------

def test_migrated_counts_and_note_176(db_session):
    from sqlalchemy import func, select

    from app.models.notes import FieldNote, FieldNoteTarget, Note

    # No count assertions: the DB is in live acceptance use (v2-native rows
    # accumulate, and migrated rows may legitimately be edited or deleted).
    # Exact-count verification against a frozen v1 export is the diff
    # harness's job (scripts/diff_harness.py) — this test only spot-checks
    # a stable migrated row end-to-end.
    assert db_session.scalar(select(func.count()).select_from(FieldNote)) > 0
    assert db_session.scalar(select(func.count()).select_from(Note)) > 0

    fn = db_session.get(FieldNote, 176)
    assert fn.body.startswith("Planted the full spring 2026")
    # The derived diary layer mirrors it 1:1 (migrate_v1: note.id == fn.id).
    note = db_session.get(Note, 176)
    assert note.author == "user"
    assert note.body_source == "user_raw"
    assert note.body_md == fn.body
    assert note.source_capture_id == 176
    prim = db_session.scalar(
        select(FieldNoteTarget).where(
            FieldNoteTarget.field_note_id == 176,
            FieldNoteTarget.is_primary.is_(True),
        )
    )
    assert (prim.target_type, prim.target_id) == ("planting", 60)


def test_load_comments_for_target_planting_60(db_session):
    from app.services import notes as svc

    comments = svc.load_comments_for_target(db_session, GARDEN, "planting", 60)
    by_id = {c["id"]: c for c in comments}
    assert 176 in by_id
    c = by_id[176]
    assert c["primary"]["type"] == "planting" and c["primary"]["id"] == 60
    assert c["is_primary_for_scope"] is True
    # Secondaries surface as labeled chips excluding the requested target.
    assert all(
        not (t["type"] == "planting" and t["id"] == 60) for t in c["other_targets"]
    )


def test_global_timeline_matches_bodied_rows(db_session):
    from sqlalchemy import func, select

    from app.models.notes import FieldNote
    from app.services import timeline as tsvc

    expected = db_session.scalar(
        select(func.count())
        .select_from(FieldNote)
        .where(
            FieldNote.garden_id == GARDEN,
            FieldNote.body.isnot(None),
            FieldNote.body != "",
            FieldNote.is_archived.isnot(True),
        )
    )
    feed = tsvc.global_feed(db_session, GARDEN)
    assert feed["view"] == "notes"
    assert len(feed["comments"]) == expected
    assert any(c["id"] == 176 for c in feed["comments"])


def test_entity_timeline_and_log_view(db_session):
    from app.services import timeline as tsvc

    feed = tsvc.entity_feed(db_session, GARDEN, "planting", 60)
    assert feed["label"]  # resolves to a real label, not the #id fallback
    assert any(c["id"] == 176 for c in feed["comments"])

    log = tsvc.global_feed(db_session, GARDEN, view="log")
    assert log["view"] == "log"
    assert isinstance(log["log_entries"], list)

    from app.errors import AppError

    with pytest.raises(AppError) as exc:
        tsvc.entity_feed(db_session, GARDEN, "gnome", 1)
    assert exc.value.http_status == 404


def test_overview_history_and_snapshot(db_session):
    from app.services import timeline as tsvc

    hist = tsvc.overview_history(db_session, GARDEN, "area", 32)
    assert hist["label"]
    before = len(hist["versions"])

    # snapshot_overview stores the CURRENT content before an edit lands…
    current = tsvc.get_current_overview(db_session, GARDEN, "area", 32)
    tsvc.snapshot_overview(db_session, GARDEN, "area", 32, current + " CHANGED")
    versions = tsvc.load_overview_versions(db_session, "area", 32)
    assert len(versions) == before + 1
    assert versions[0]["content"] == current
    # …and is a no-op when nothing changed.
    tsvc.snapshot_overview(db_session, GARDEN, "area", 32, current)
    assert len(tsvc.load_overview_versions(db_session, "area", 32)) == before + 1


# ---------------------------------------------------------------------------
# narrowed inference (DB)
# ---------------------------------------------------------------------------

def test_inferred_targets_narrowed_default(db_session):
    """Area comment: ONLY year + parent-area chain + serving station.

    This is the sanctioned v1 fix — v1 linked every plant/species/variety in
    the Raspberry Patch subtree (25 plants, 8 species); the narrowed default
    must link none of them."""
    from datetime import date

    from app.services import notes as svc

    pairs = svc.inferred_targets_for(
        db_session, GARDEN, "area", 32, date(2026, 7, 2),
        body_text="Sawfly larvae all over the place this morning.",
    )
    types = {t for t, _ in pairs}
    assert types <= {"year", "area", "station"}, pairs
    # Year 2026 (existing row), the serving station, and the parent chain.
    year_2026 = [i for t, i in pairs if t == "year"]
    assert len(year_2026) == 1
    assert ("station", 18) in pairs
    by_id = svc.load_area_maps(db_session, GARDEN, include_archived=False)
    parent_chain = svc.get_ancestor_ids(32, by_id)[:-1]  # ancestors, not self
    for aid in parent_chain:
        assert ("area", aid) in pairs
    assert ("area", 32) not in pairs  # self is always dropped


def test_inferred_targets_mention_gated(db_session):
    """species/variety/plant secondaries appear ONLY when named in the body."""
    from datetime import date

    from sqlalchemy import select

    from app.models.catalog import Species
    from app.models.planting import Plant
    from app.services import notes as svc

    raspberry_id = db_session.scalar(
        select(Species.id).where(Species.garden_id == GARDEN, Species.name == "Raspberry")
    )
    assert raspberry_id is not None

    pairs = svc.inferred_targets_for(
        db_session, GARDEN, "area", 32, date(2026, 7, 2),
        body_text="The raspberry canes are loaded with fruit.",
    )
    assert ("species", raspberry_id) in pairs
    # Plant-group links: only raspberry plants, only inside the subtree, ≤ cap.
    plant_ids = [i for t, i in pairs if t == "plant"]
    assert plant_ids, "mentioned species should link its in-scope plant groups"
    assert len(plant_ids) <= svc.PLANT_CAP
    for pid in plant_ids:
        assert db_session.scalar(
            select(Plant.species_id).where(Plant.id == pid)
        ) == raspberry_id
    # An UNmentioned species in the same subtree (e.g. Apple) is NOT linked.
    apple_id = db_session.scalar(
        select(Species.id).where(Species.garden_id == GARDEN, Species.name == "Apple")
    )
    assert apple_id is not None and ("species", apple_id) not in pairs


def test_inferred_targets_variety_keeps_parent_species(db_session):
    from sqlalchemy import select

    from app.models.catalog import Variety
    from app.services import notes as svc

    row = db_session.execute(
        select(Variety.id, Variety.species_id)
        .where(Variety.species_id.isnot(None), Variety.is_archived.isnot(True))
        .limit(1)
    ).first()
    pairs = svc.inferred_targets_for(db_session, GARDEN, "variety", row.id)
    assert ("species", row.species_id) in pairs
    # Narrowed: no plant links without a body mention, no planting links.
    assert not [p for p in pairs if p[0] in ("plant", "planting")]


def test_get_or_create_year_idempotent(db_session):
    from sqlalchemy import select

    from app.models.garden import Year
    from app.services import notes as svc

    existing = db_session.scalar(
        select(Year.id).where(Year.garden_id == GARDEN, Year.year == 2026)
    )
    assert svc.get_or_create_year(db_session, GARDEN, 2026) == existing
    new_id = svc.get_or_create_year(db_session, GARDEN, 2031)
    assert svc.get_or_create_year(db_session, GARDEN, 2031) == new_id


# ---------------------------------------------------------------------------
# comment writes: dual-write + mirror lifecycle (DB, rolled back)
# ---------------------------------------------------------------------------

def _create_test_comment(db, **kw):
    from datetime import date

    from app.services import notes as svc

    defaults = dict(
        target_type="area",
        target_id=32,
        comment_date=date(2026, 7, 2),
        body="Parity-suite test note — dual write check.",
        kind="observation",
    )
    defaults.update(kw)
    return svc.create_comment(db, GARDEN, **defaults)


def test_create_comment_dual_writes_field_note_and_mirror(db_session):
    from sqlalchemy import select

    from app.models.notes import FieldNote, FieldNoteTarget, Note, NoteTarget

    c = _create_test_comment(db_session)
    cid = c["id"]

    fn = db_session.get(FieldNote, cid)
    assert fn.status == "processed" and fn.processed_at is not None
    assert fn.body == "Parity-suite test note — dual write check."
    assert fn.kind == "observation"

    fnts = db_session.scalars(
        select(FieldNoteTarget).where(FieldNoteTarget.field_note_id == cid)
    ).all()
    prim = [t for t in fnts if t.is_primary]
    assert len(prim) == 1 and (prim[0].target_type, prim[0].target_id) == ("area", 32)
    # Narrowed inference ran: year + stations + parent areas only.
    assert {t.target_type for t in fnts} <= {"area", "station", "year"}

    # Mirror note: author='user', body VERBATIM, targets mirrored 'confirmed'.
    mirror = db_session.scalar(select(Note).where(Note.source_capture_id == cid))
    assert mirror is not None
    assert mirror.author == "user" and mirror.body_source == "user_raw"
    assert mirror.body_md == fn.body
    assert mirror.note_date == fn.comment_date
    nts = db_session.scalars(
        select(NoteTarget).where(NoteTarget.note_id == mirror.id)
    ).all()
    assert {(t.target_type, t.target_id, bool(t.is_primary)) for t in nts} == {
        (t.target_type, t.target_id, bool(t.is_primary)) for t in fnts
    }
    assert all(t.link_source == "confirmed" for t in nts)

    # Activity log row (audit on mutations).
    from app.models.history import ActivityLog

    log = db_session.scalars(
        select(ActivityLog).where(
            ActivityLog.entity_type == "comment", ActivityLog.entity_id == cid
        )
    ).all()
    assert log and log[0].category == "add"


def test_create_comment_explicit_secondaries_skip_inference(db_session):
    from sqlalchemy import select

    from app.models.notes import FieldNoteTarget

    c = _create_test_comment(db_session, explicit_secondaries=[("species", 5)])
    fnts = db_session.scalars(
        select(FieldNoteTarget).where(FieldNoteTarget.field_note_id == c["id"])
    ).all()
    assert {(t.target_type, t.target_id) for t in fnts} == {
        ("area", 32), ("species", 5)
    }


def test_media_only_comment_has_no_mirror(db_session, media_dir):
    from sqlalchemy import select

    from app.models.notes import Note

    c = _create_test_comment(
        db_session, body="", photos=[_upload(b"img", "leaf.jpg", "image/jpeg")]
    )
    assert c["photos"] and c["body"] == ""
    # Migration invariant: one notes row per BODIED field_note only.
    assert (
        db_session.scalar(select(Note).where(Note.source_capture_id == c["id"]))
        is None
    )


def test_update_comment_syncs_mirror(db_session):
    from sqlalchemy import select

    from app.models.notes import Note, NoteRevision
    from app.services import notes as svc

    c = _create_test_comment(db_session)
    cid = c["id"]
    old_body = c["body"]

    updated = svc.update_comment(db_session, GARDEN, cid, body="Rewritten body.")
    assert updated["body"] == "Rewritten body."
    mirror = db_session.scalar(select(Note).where(Note.source_capture_id == cid))
    assert mirror.body_md == "Rewritten body."
    revs = db_session.scalars(
        select(NoteRevision).where(NoteRevision.note_id == mirror.id)
    ).all()
    assert len(revs) == 1 and revs[0].body == old_body


def test_delete_comment_cleans_both_layers(db_session):
    from sqlalchemy import select

    from app.models.notes import FieldNote, FieldNoteTarget, Note, NoteTarget
    from app.services import notes as svc

    c = _create_test_comment(db_session)
    cid = c["id"]
    mirror_id = db_session.scalar(
        select(Note.id).where(Note.source_capture_id == cid)
    )
    svc.delete_field_note(db_session, GARDEN, cid)
    assert db_session.get(FieldNote, cid) is None
    assert db_session.get(Note, mirror_id) is None
    assert not db_session.scalars(
        select(FieldNoteTarget).where(FieldNoteTarget.field_note_id == cid)
    ).all()
    assert not db_session.scalars(
        select(NoteTarget).where(NoteTarget.note_id == mirror_id)
    ).all()


def test_scoped_timeline_expands_children_and_related(db_session):
    """v1 load_comments_expanded semantics: a comment on a child area shows
    up in the parent area's scoped feed when children are included (the
    default for areas), and NOT when children are excluded."""
    from app.services import notes as svc
    from app.services import timeline as tsvc

    # explicit_secondaries=[] keeps the comment target-pure (no inferred
    # parent-chain links), so visibility on the parent MUST come from the
    # children expansion, not from a cached secondary target.
    c = _create_test_comment(db_session, explicit_secondaries=[])  # area 32
    by_id = svc.load_area_maps(db_session, GARDEN)
    parent = by_id[32]["parent_id"]
    assert parent

    feed = tsvc.global_feed(db_session, GARDEN, scope_type="area", scope_id=parent)
    assert feed["include_children"] and feed["supports_children"]
    mine = [x for x in feed["comments"] if x["id"] == c["id"]]
    assert mine and mine[0]["is_primary_for_scope"] is False
    assert mine[0]["primary"]["id"] == 32

    feed_no_children = tsvc.global_feed(
        db_session, GARDEN, scope_type="area", scope_id=parent, children=False
    )
    assert not any(x["id"] == c["id"] for x in feed_no_children["comments"])

    # With DEFAULT inference the parent-chain areas become real secondary
    # targets, so the comment is visible on the parent even without
    # expansion — that is the point of the narrowed-inference chain.
    c_inferred = _create_test_comment(db_session)
    feed_no_children2 = tsvc.global_feed(
        db_session, GARDEN, scope_type="area", scope_id=parent, children=False
    )
    assert any(x["id"] == c_inferred["id"] for x in feed_no_children2["comments"])

    # Related expansion from the species side finds area comments that
    # mention the species (comment got a mention-gated species link).
    c2 = _create_test_comment(
        db_session, body="The raspberry canes doubled this week."
    )
    from sqlalchemy import select

    from app.models.catalog import Species

    rasp = db_session.scalar(select(Species.id).where(Species.name == "Raspberry"))
    feed_sp = tsvc.global_feed(
        db_session, GARDEN, scope_type="species", scope_id=rasp, children=False
    )
    assert any(x["id"] == c2["id"] for x in feed_sp["comments"])


def test_target_add_remove_and_last_target_guard(db_session):
    from app.errors import AppError
    from app.services import notes as svc

    c = _create_test_comment(db_session, explicit_secondaries=[])
    cid = c["id"]

    # Only the primary exists → removal refused with the structured code.
    with pytest.raises(AppError) as exc:
        svc.remove_comment_target(db_session, GARDEN, cid, "area", 32)
    assert exc.value.code == svc.NOTE_LAST_TARGET

    targets = svc.add_comment_target(db_session, GARDEN, cid, "species", 5)
    assert {(t["type"], t["id"]) for t in targets} == {("area", 32), ("species", 5)}
    # Idempotent (v1 INSERT OR IGNORE).
    assert len(svc.add_comment_target(db_session, GARDEN, cid, "species", 5)) == 2

    # Removing a target that isn't attached is a silent no-op (v1 parity;
    # the count guard passes because two targets exist).
    assert len(svc.remove_comment_target(db_session, GARDEN, cid, "area", 99999)) == 2

    # Removing the PRIMARY promotes the first remaining target.
    targets = svc.remove_comment_target(db_session, GARDEN, cid, "area", 32)
    assert [(t["type"], t["id"], t["is_primary"]) for t in targets] == [
        ("species", 5, True)
    ]


def test_invalid_target_type_rejected(db_session):
    from app.errors import AppError
    from app.services import notes as svc

    with pytest.raises(AppError) as exc:
        _create_test_comment(db_session, target_type="gnome")
    assert exc.value.code == svc.NOTE_INVALID_TARGET_TYPE


# ---------------------------------------------------------------------------
# captures (DB, rolled back)
# ---------------------------------------------------------------------------

def test_capture_lifecycle(db_session):
    from app.errors import AppError
    from app.services import notes as svc

    cap = svc.create_capture(db_session, GARDEN, text="check the peas tomorrow")
    assert cap["status"] == "new" and cap["processed_at"] is None

    listed = svc.list_captures(db_session, GARDEN, status="new")
    assert any(n["id"] == cap["id"] for n in listed)
    assert svc.queue_counts(db_session, GARDEN)["new"] >= 1

    # Bad status in the list API falls back to 'new' (v1 parity).
    assert svc.list_captures(db_session, GARDEN, status="bogus") == listed

    # PATCH: finalize.
    upd = svc.update_capture(
        db_session, GARDEN, cap["id"],
        {"status": "processed", "processing_notes": "handled in parity test"},
    )
    assert upd["status"] == "processed" and upd["processed_at"] is not None
    assert upd["processing_notes"] == "handled in parity test"

    with pytest.raises(AppError) as exc:
        svc.update_capture(db_session, GARDEN, cap["id"], {})
    assert exc.value.code == svc.NOTE_NO_UPDATABLE_FIELDS
    with pytest.raises(AppError) as exc:
        svc.update_capture(db_session, GARDEN, cap["id"], {"status": "nope"})
    assert exc.value.code == svc.NOTE_INVALID_STATUS

    svc.delete_field_note(db_session, GARDEN, cap["id"])
    with pytest.raises(AppError):
        svc.update_capture(db_session, GARDEN, cap["id"], {"text": "gone"})


def test_empty_capture_rejected(db_session):
    from app.errors import AppError
    from app.services import notes as svc

    with pytest.raises(AppError) as exc:
        svc.create_capture(db_session, GARDEN, text="   ")
    assert exc.value.code == svc.NOTE_EMPTY


def test_transcribe_guards(db_session):
    """No audio → structured 400 (the Gemini call itself is not exercised
    here — it would need a live key; the guard path is the DB behavior)."""
    from app.errors import AppError
    from app.services import notes as svc

    cap = svc.create_capture(db_session, GARDEN, text="text-only capture")
    with pytest.raises(AppError) as exc:
        svc.get_capture_audio_abs_path(db_session, GARDEN, cap["id"])
    assert exc.value.code == svc.NOTE_NO_AUDIO


# ---------------------------------------------------------------------------
# App feedback (v2-only: kind='feedback' + page_context)
# ---------------------------------------------------------------------------

def test_feedback_capture_roundtrip(db_session):
    from app.services import notes as svc

    out = svc.create_capture(
        db_session, GARDEN,
        text="the layout tray hides the compass on small screens",
        kind="feedback",
        page_context="/garden/areas/12/layout",
    )
    assert out["kind"] == "feedback"
    assert out["page_context"] == "/garden/areas/12/layout"
    assert out["status"] == "new"

    # kind filter returns it; plain-new listing also includes it (status new)
    ids = [c["id"] for c in svc.list_captures(db_session, GARDEN, status="all", kind="feedback")]
    assert out["id"] in ids


def test_capture_invalid_kind_coerced_and_page_context_capped(db_session):
    from app.services import notes as svc

    out = svc.create_capture(
        db_session, GARDEN,
        text="x",
        kind="not-a-kind",
        page_context="/p" + "a" * 500,
    )
    assert out["kind"] == ""
    assert len(out["page_context"]) <= 300


def test_feedback_excluded_from_diary_surfaces(db_session):
    from datetime import date

    from app.services import notes as svc
    from app.services import timeline as tl

    # A PROCESSED feedback comment (worst case: bodied + processed — the
    # status filter alone would NOT hide it) must still stay out of entity
    # threads and the global timeline.
    out = svc.create_comment(
        db_session, GARDEN,
        target_type="area", target_id=32,
        comment_date=date(2026, 7, 3),
        body="feedback left from an entity page",
        kind="feedback",
        page_context="/garden/areas/32",
    )
    fid = out["id"]

    thread_ids = [c["id"] for c in svc.load_comments_for_target(db_session, GARDEN, "area", 32)]
    assert fid not in thread_ids

    by_id = svc.load_area_maps(db_session, GARDEN)
    global_ids = [c["id"] for c in tl._global_comments(db_session, GARDEN, by_id, True)]
    assert fid not in global_ids

    # ...but it IS retrievable through the captures listing by kind.
    fb_ids = [c["id"] for c in svc.list_captures(db_session, GARDEN, status="all", kind="feedback")]
    assert fid in fb_ids

    # And the notes-mirror dual-write skipped it (no diary-layer row).
    from sqlalchemy import select as _select

    from app.models.notes import Note as _Note

    assert (
        db_session.scalar(_select(_Note).where(_Note.source_capture_id == fid))
        is None
    )

    # AI surfaces exclude it too (review finding, dev.13): the chat starter
    # context and both note-reading chat tools must never see feedback.
    from app.services.ai import snapshot as snap
    from app.services.ai.tools import garden_tools as gt

    ctx = snap.build_starter_context(db_session, GARDEN)
    assert "feedback left from an entity page" not in ctx

    recent = gt.get_recent_notes(
        db_session, GARDEN, target_type="area", target_id=32
    )
    assert fid not in [n["id"] for n in recent.get("notes", [])]

    hits = gt.search_notes(db_session, GARDEN, query="feedback left from an entity")
    assert fid not in [n["id"] for n in hits.get("matches", [])]