feat(ml): presentation-chrome auto-hide sweep + hard-skip + conflict flagging (#141 step 4)
presentation_auto_apply_sweep fires banner/editor-screenshot heads at the FLAT presentation threshold (source=presentation_auto). Two guards: (1) hard-skip any image already carrying a human/confirmed content tag — you valued it, so the model can't bury it; (2) if an auto-hide ALSO scores >= presentation_conflict_threshold on a content head, hide it but record a PresentationReview row (conflict tag + score) for the Hidden view. _auto_apply_heads now excludes system tags, so a graduated wip/banner can't fire via the content path (and wip never auto-applies at all). presentation_auto added to _AUTO_SOURCES so auto-hidden chrome never self-trains. Tests: applies, hard-skip valued, conflict-flag, disabled no-op, ignores wip, content-path excludes system. Settings UI + scheduling land next. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM
This commit is contained in:
@@ -0,0 +1,152 @@
|
||||
"""Presentation-chrome auto-hide sweep (#141). numpy-only (no sklearn), so the
|
||||
apply + guard logic is tested directly via the sync session."""
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
|
||||
from backend.app.models import (
|
||||
HeadAutoApplyRun,
|
||||
ImageRecord,
|
||||
MLSettings,
|
||||
PresentationReview,
|
||||
Tag,
|
||||
TagHead,
|
||||
TagKind,
|
||||
)
|
||||
from backend.app.models.tag import image_tag
|
||||
from backend.app.services.ml.heads import (
|
||||
auto_apply_sweep,
|
||||
presentation_auto_apply_sweep,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
def _emb(slot: int) -> list[float]:
|
||||
v = [0.0] * 1152
|
||||
v[slot] = 3.0
|
||||
return v
|
||||
|
||||
|
||||
def _img(db, sha: str, emb) -> 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 _head(db, tag_id: int, slot: int, *, weight=1.0):
|
||||
# weight 3.0 → score sigmoid(3)=0.95 clears the 0.90 presentation floor;
|
||||
# weight 1.0 → sigmoid(1)=0.73 clears the 0.50 conflict floor.
|
||||
s = db.execute(select(MLSettings).where(MLSettings.id == 1)).scalar_one()
|
||||
w = [0.0] * 1152
|
||||
w[slot] = weight
|
||||
db.add(TagHead(
|
||||
tag_id=tag_id, embedding_version=s.embedder_model_version,
|
||||
weights=w, bias=0.0, suggest_threshold=0.5, auto_apply_threshold=0.5,
|
||||
n_pos=60, n_neg=90, ap=0.9, precision_cv=0.98, recall=0.7,
|
||||
))
|
||||
|
||||
|
||||
def _system_tag(db, name):
|
||||
return db.execute(
|
||||
select(Tag).where(Tag.is_system.is_(True), Tag.name == name)
|
||||
).scalar_one()
|
||||
|
||||
|
||||
def _source(db, image_id, tag_id):
|
||||
return db.execute(
|
||||
select(image_tag.c.source)
|
||||
.where(image_tag.c.image_record_id == image_id)
|
||||
.where(image_tag.c.tag_id == tag_id)
|
||||
).scalar_one_or_none()
|
||||
|
||||
|
||||
def test_presentation_sweep_hides_chrome(db_sync):
|
||||
banner = _system_tag(db_sync, "banner")
|
||||
_head(db_sync, banner.id, 0, weight=3.0)
|
||||
img = _img(db_sync, "a" * 64, _emb(0))
|
||||
db_sync.commit()
|
||||
res = presentation_auto_apply_sweep(db_sync)
|
||||
assert res["n_applied"] == 1
|
||||
assert _source(db_sync, img.id, banner.id) == "presentation_auto"
|
||||
|
||||
|
||||
def test_presentation_sweep_hard_skips_valued_image(db_sync):
|
||||
# An image the operator already tagged with a content tag is never auto-hidden.
|
||||
banner = _system_tag(db_sync, "banner")
|
||||
_head(db_sync, banner.id, 0, weight=3.0)
|
||||
content = Tag(name="mychar", kind=TagKind.character)
|
||||
db_sync.add(content)
|
||||
db_sync.flush()
|
||||
img = _img(db_sync, "b" * 64, _emb(0))
|
||||
db_sync.execute(image_tag.insert().values(
|
||||
image_record_id=img.id, tag_id=content.id, source="manual"))
|
||||
db_sync.commit()
|
||||
res = presentation_auto_apply_sweep(db_sync)
|
||||
assert res["n_applied"] == 0
|
||||
assert _source(db_sync, img.id, banner.id) is None
|
||||
|
||||
|
||||
def test_presentation_sweep_flags_conflict(db_sync):
|
||||
# Matches banner AND scores high on a content head → hidden but flagged with
|
||||
# that content tag as the conflict.
|
||||
banner = _system_tag(db_sync, "banner")
|
||||
_head(db_sync, banner.id, 0, weight=3.0)
|
||||
content = Tag(name="looksreal", kind=TagKind.general)
|
||||
db_sync.add(content)
|
||||
db_sync.flush()
|
||||
_head(db_sync, content.id, 0, weight=1.0) # content head also fires
|
||||
img = _img(db_sync, "c" * 64, _emb(0))
|
||||
db_sync.commit()
|
||||
res = presentation_auto_apply_sweep(db_sync)
|
||||
assert res["n_applied"] == 1
|
||||
assert res["n_flagged"] == 1
|
||||
assert _source(db_sync, img.id, banner.id) == "presentation_auto"
|
||||
flag = db_sync.execute(
|
||||
select(PresentationReview).where(
|
||||
PresentationReview.image_record_id == img.id,
|
||||
PresentationReview.tag_id == banner.id,
|
||||
)
|
||||
).scalar_one()
|
||||
assert flag.conflict_tag_id == content.id
|
||||
assert flag.conflict_score >= 0.5
|
||||
|
||||
|
||||
def test_presentation_sweep_disabled_is_noop(db_sync):
|
||||
s = db_sync.execute(select(MLSettings).where(MLSettings.id == 1)).scalar_one()
|
||||
s.presentation_auto_apply_enabled = False
|
||||
banner = _system_tag(db_sync, "banner")
|
||||
_head(db_sync, banner.id, 0, weight=3.0)
|
||||
img = _img(db_sync, "d" * 64, _emb(0))
|
||||
db_sync.commit()
|
||||
res = presentation_auto_apply_sweep(db_sync)
|
||||
assert res["n_applied"] == 0
|
||||
assert _source(db_sync, img.id, banner.id) is None
|
||||
|
||||
|
||||
def test_presentation_sweep_ignores_wip(db_sync):
|
||||
# wip is a system tag but NOT presentation chrome → never auto-applied.
|
||||
wip = _system_tag(db_sync, "wip")
|
||||
_head(db_sync, wip.id, 0, weight=3.0)
|
||||
img = _img(db_sync, "e" * 64, _emb(0))
|
||||
db_sync.commit()
|
||||
res = presentation_auto_apply_sweep(db_sync)
|
||||
assert res["n_applied"] == 0
|
||||
assert _source(db_sync, img.id, wip.id) is None
|
||||
|
||||
|
||||
def test_content_sweep_never_fires_system_tags(db_sync):
|
||||
# A graduated banner (system) head must NOT auto-apply via the content path.
|
||||
banner = _system_tag(db_sync, "banner")
|
||||
_head(db_sync, banner.id, 0, weight=3.0)
|
||||
img = _img(db_sync, "f" * 64, _emb(0))
|
||||
run = HeadAutoApplyRun(dry_run=False, params={}, status="running")
|
||||
db_sync.add(run)
|
||||
db_sync.flush()
|
||||
db_sync.commit()
|
||||
auto_apply_sweep(db_sync, run, dry_run=False)
|
||||
assert _source(db_sync, img.id, banner.id) is None
|
||||
Reference in New Issue
Block a user