FabledCurator/tests/test_retract_auto_apply.py

"""Soft auto-apply (milestone 139): the retraction sweeps drop standing
head_auto/ccip_auto tags now below their threshold, keep the ones still above,
and never touch manual or operator-confirmed tags. Sync + sklearn-free (they
score with STORED weights/vectors), so tested directly via db_sync."""
import pytest
from sqlalchemy import select

from backend.app.models import (
    CharacterPrototype,
    ImageRecord,
    ImageRegion,
    MLSettings,
    Tag,
    TagHead,
    TagKind,
    TagPositiveConfirmation,
)
from backend.app.models.tag import image_tag
from backend.app.services.ml.character_prototypes import retract_auto_applied_ccip
from backend.app.services.ml.heads import retract_auto_applied_heads

pytestmark = pytest.mark.integration


def _emb(slot: int) -> list[float]:
    v = [0.0] * 1152
    v[slot] = 3.0
    return v


def _ccip(slot: int) -> list[float]:
    v = [0.0] * 768
    v[slot] = 1.0
    return v


def _img(db, sha: str, emb=None) -> ImageRecord:
    img = ImageRecord(
        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
        width=1, height=1, origin="imported_filesystem",
        integrity_status="unknown", siglip_embedding=emb,
    )
    db.add(img)
    db.flush()
    return img


def _figure(db, image_id: int, ccip) -> None:
    db.add(ImageRegion(
        image_record_id=image_id, kind="figure",
        rx=0.0, ry=0.0, rw=1.0, rh=1.0,
        ccip_embedding=ccip, embedding_version="ccip-test",
    ))
    db.flush()


def _tag(db, name: str, kind: TagKind) -> Tag:
    t = Tag(name=name, kind=kind)
    db.add(t)
    db.flush()
    return t


def _apply(db, image_id: int, tag_id: int, source: str) -> None:
    db.execute(image_tag.insert().values(
        image_record_id=image_id, tag_id=tag_id, source=source,
    ))


def _version(db) -> str:
    return db.execute(
        select(MLSettings.embedder_model_version).where(MLSettings.id == 1)
    ).scalar_one()


def _head(db, tag_id: int, slot: int, threshold: float, version: str) -> None:
    w = [0.0] * 1152
    w[slot] = 1.0
    db.add(TagHead(
        tag_id=tag_id, embedding_version=version, weights=w, bias=0.0,
        suggest_threshold=0.5, auto_apply_threshold=threshold,
        n_pos=60, n_neg=180, ap=0.9, precision_cv=0.98, recall=0.7,
    ))
    db.flush()


def _has_tag(db, image_id: int, tag_id: int) -> bool:
    return db.execute(
        select(image_tag.c.tag_id)
        .where(image_tag.c.image_record_id == image_id)
        .where(image_tag.c.tag_id == tag_id)
    ).first() is not None


def test_retract_head_auto(db_sync):
    ver = _version(db_sync)
    tag = _tag(db_sync, "glasses", TagKind.general)
    _head(db_sync, tag.id, slot=0, threshold=0.7, version=ver)
    hi = _img(db_sync, "a" * 64, _emb(0))     # aligned → ~0.73 ≥ 0.7 → keep
    lo = _img(db_sync, "b" * 64, _emb(5))     # orthogonal → 0.5 < 0.7 → retract
    man = _img(db_sync, "c" * 64, _emb(5))    # low score but manual → keep
    conf = _img(db_sync, "d" * 64, _emb(5))   # low score, head_auto, CONFIRMED → keep
    _apply(db_sync, hi.id, tag.id, "head_auto")
    _apply(db_sync, lo.id, tag.id, "head_auto")
    _apply(db_sync, man.id, tag.id, "manual")
    _apply(db_sync, conf.id, tag.id, "head_auto")
    db_sync.add(TagPositiveConfirmation(image_record_id=conf.id, tag_id=tag.id))
    db_sync.commit()

    assert retract_auto_applied_heads(db_sync) == 1
    assert not _has_tag(db_sync, lo.id, tag.id)     # retracted (below threshold)
    assert _has_tag(db_sync, hi.id, tag.id)         # kept (still above)
    assert _has_tag(db_sync, man.id, tag.id)        # kept (manual, not auto)
    assert _has_tag(db_sync, conf.id, tag.id)       # kept (operator-confirmed)


def test_retract_head_auto_noop_when_disabled(db_sync):
    s = db_sync.execute(select(MLSettings).where(MLSettings.id == 1)).scalar_one()
    s.head_auto_apply_enabled = False
    ver = _version(db_sync)
    tag = _tag(db_sync, "glasses", TagKind.general)
    _head(db_sync, tag.id, slot=0, threshold=0.7, version=ver)
    lo = _img(db_sync, "e" * 64, _emb(5))     # would be below threshold
    _apply(db_sync, lo.id, tag.id, "head_auto")
    db_sync.commit()

    assert retract_auto_applied_heads(db_sync) == 0
    assert _has_tag(db_sync, lo.id, tag.id)         # switch off → nothing retracted


def test_retract_ccip_auto(db_sync):
    char = _tag(db_sync, "Raven", TagKind.character)
    db_sync.add(CharacterPrototype(tag_id=char.id, ccip_embedding=_ccip(0)))
    hi = _img(db_sync, "f" * 64)              # figure matches prototype → keep
    lo = _img(db_sync, "g" * 64)             # figure orthogonal → retract
    conf = _img(db_sync, "h" * 64)           # orthogonal, CONFIRMED → keep
    man = _img(db_sync, "i" * 64)            # orthogonal, manual → keep
    _figure(db_sync, hi.id, _ccip(0))
    _figure(db_sync, lo.id, _ccip(5))
    _figure(db_sync, conf.id, _ccip(5))
    _figure(db_sync, man.id, _ccip(5))
    _apply(db_sync, hi.id, char.id, "ccip_auto")
    _apply(db_sync, lo.id, char.id, "ccip_auto")
    _apply(db_sync, conf.id, char.id, "ccip_auto")
    _apply(db_sync, man.id, char.id, "manual")
    db_sync.add(TagPositiveConfirmation(image_record_id=conf.id, tag_id=char.id))
    db_sync.commit()

    assert retract_auto_applied_ccip(db_sync) == 1
    assert not _has_tag(db_sync, lo.id, char.id)    # retracted (below threshold)
    assert _has_tag(db_sync, hi.id, char.id)        # kept (match ≥ threshold)
    assert _has_tag(db_sync, conf.id, char.id)      # kept (operator-confirmed)
    assert _has_tag(db_sync, man.id, char.id)       # kept (manual, not auto)