From eedf8d109a19d959df14268187f299a43377caad Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 6 Jul 2026 23:11:26 -0400 Subject: [PATCH] feat(ml): presentation-chrome auto-hide sweep + hard-skip + conflict flagging (#141 step 4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM --- backend/app/services/ml/heads.py | 167 ++++++++++++++++++++++- backend/app/services/ml/training_data.py | 2 +- tests/test_presentation_auto_apply.py | 152 +++++++++++++++++++++ 3 files changed, 317 insertions(+), 4 deletions(-) create mode 100644 tests/test_presentation_auto_apply.py diff --git a/backend/app/services/ml/heads.py b/backend/app/services/ml/heads.py index ae85b46..47e4073 100644 --- a/backend/app/services/ml/heads.py +++ b/backend/app/services/ml/heads.py @@ -32,13 +32,14 @@ from ...models import ( ImageRecord, ImageRegion, MLSettings, + PresentationReview, Tag, TagHead, TagKind, TagPositiveConfirmation, TagSuggestionRejection, ) -from ...models.tag import image_tag +from ...models.tag import PRESENTATION_SYSTEM_TAGS, image_tag from .training_data import ( _AUTO_SOURCES, _auto_apply_point, @@ -649,8 +650,11 @@ def start_head_auto_apply_run(session: Session, params: dict[str, Any]) -> int: def _auto_apply_heads(session: Session, embedding_version: str, min_pos: int): - """Eligible heads to fire: graduated (auto_apply_threshold set), enough - support, current embedding. Returns the row list (tag_id/name/weights/...).""" + """Eligible CONTENT heads to fire: graduated (auto_apply_threshold set), + enough support, current embedding, NON-system. System tags never auto-apply + via this path — `wip` never auto-applies at all, and banner/editor screenshot + go through the presentation path at their own flat threshold (#141). Returns + the row list (tag_id/name/weights/...).""" return session.execute( select( TagHead.tag_id, Tag.name, TagHead.weights, TagHead.bias, @@ -660,6 +664,7 @@ def _auto_apply_heads(session: Session, embedding_version: str, min_pos: int): .where(TagHead.embedding_version == embedding_version) .where(TagHead.auto_apply_threshold.is_not(None)) .where(TagHead.n_pos >= min_pos) + .where(~Tag.is_system) ).all() @@ -743,6 +748,162 @@ def auto_apply_sweep( return {"n_applied": sum(applied), "concepts": concepts} +_PRESENTATION_SOURCE = "presentation_auto" + + +def _presentation_heads(session: Session, embedding_version: str): + """Trained heads for the presentation chrome tags (banner / editor screenshot). + They fire at the FLAT presentation threshold regardless of graduation — a head + exists once the operator has labelled enough chrome (head_min_positives).""" + return session.execute( + select(TagHead.tag_id, Tag.name, TagHead.weights, TagHead.bias) + .join(Tag, Tag.id == TagHead.tag_id) + .where(TagHead.embedding_version == embedding_version) + .where(Tag.is_system.is_(True)) + .where(Tag.name.in_(PRESENTATION_SYSTEM_TAGS)) + ).all() + + +def _conflict_heads(session: Session, embedding_version: str): + """ALL content (non-system) heads — the "does this ALSO look like real + content" signal for the presentation conflict guard (#141).""" + return session.execute( + select(TagHead.tag_id, TagHead.weights, TagHead.bias) + .join(Tag, Tag.id == TagHead.tag_id) + .where(TagHead.embedding_version == embedding_version) + .where(~Tag.is_system) + ).all() + + +def _valued_image_ids(session: Session) -> set[int]: + """Images the operator has shown they value: carrying a HUMAN or CONFIRMED + content (non-system) tag. The presentation sweep never auto-hides these + (guard 1) — you tagged it, so the model doesn't get to bury it (#141).""" + confirmed = exists().where( + TagPositiveConfirmation.image_record_id == image_tag.c.image_record_id, + TagPositiveConfirmation.tag_id == image_tag.c.tag_id, + ) + rows = session.execute( + select(image_tag.c.image_record_id) + .join(Tag, Tag.id == image_tag.c.tag_id) + .where(~Tag.is_system) + .where(image_tag.c.source.not_in(_AUTO_SOURCES) | confirmed) + ).all() + return {r[0] for r in rows} + + +def presentation_auto_apply_sweep(session: Session, dry_run: bool = False) -> dict: + """Auto-hide presentation chrome (banner / editor screenshot) at the FLAT + presentation threshold (#141) — NOT the per-head graduated threshold. Two + guards keep it safe: (1) never hide an image carrying a human/confirmed content + tag; (2) if an image about to be hidden ALSO scores >= the conflict threshold + on a content head, still hide it but flag it (PresentationReview) so the Hidden + view surfaces "also looks like " for review. No-op unless + presentation_auto_apply_enabled. numpy-only (no sklearn). Returns + {n_applied, n_flagged, concepts}.""" + import numpy as np + from sqlalchemy.dialects.postgresql import insert as pg_insert + + settings = _settings(session) + if not dry_run and not settings.presentation_auto_apply_enabled: + return {"n_applied": 0, "n_flagged": 0, "concepts": []} + ver = settings.embedder_model_version + pres = _presentation_heads(session, ver) + if not pres: + return {"n_applied": 0, "n_flagged": 0, "concepts": []} + thr = float(settings.presentation_auto_apply_threshold) + conflict_thr = float(settings.presentation_conflict_threshold) + + Wp = np.vstack([np.asarray(r.weights, dtype=np.float32) for r in pres]) + bp = np.asarray([r.bias for r in pres], dtype=np.float32) + pres_tag_ids = [r.tag_id for r in pres] + pres_names = [r.name for r in pres] + + conf = _conflict_heads(session, ver) + Wc = bc = conf_tag_ids = None + if conf: + Wc = np.vstack([np.asarray(r.weights, dtype=np.float32) for r in conf]) + bc = np.asarray([r.bias for r in conf], dtype=np.float32) + conf_tag_ids = [r.tag_id for r in conf] + + valued = _valued_image_ids(session) + + # Skip images that already carry, or have rejected, each presentation tag. + skip = {tid: set() for tid in pres_tag_ids} + for tid in pres_tag_ids: + for (iid,) in session.execute( + select(image_tag.c.image_record_id).where(image_tag.c.tag_id == tid) + ): + skip[tid].add(iid) + for (iid,) in session.execute( + select(TagSuggestionRejection.image_record_id).where( + TagSuggestionRejection.tag_id == tid + ) + ): + skip[tid].add(iid) + + applied = [0] * len(pres) + n_flagged = 0 + scanned = 0 + all_ids = list(session.execute( + select(ImageRecord.id).where(ImageRecord.siglip_embedding.is_not(None)) + ).scalars()) + for start in range(0, len(all_ids), _AUTO_APPLY_CHUNK): + chunk = all_ids[start:start + _AUTO_APPLY_CHUNK] + emb = _load_embeddings(session, chunk) + cids = [i for i in chunk if i in emb] + if not cids: + continue + Xn = _l2norm(np.vstack([emb[i] for i in cids]).astype(np.float32), np) + probs = 1.0 / (1.0 + np.exp(-(Xn @ Wp.T + bp))) # (N, P) + if Wc is not None: + cprobs = 1.0 / (1.0 + np.exp(-(Xn @ Wc.T + bc))) # (N, C) + max_c = cprobs.max(axis=1) + arg_c = cprobs.argmax(axis=1) + scanned += len(cids) + for p in range(len(pres)): + tid = pres_tag_ids[p] + for idx in np.where(probs[:, p] >= thr)[0]: + iid = cids[int(idx)] + if iid in skip[tid] or iid in valued: + continue + skip[tid].add(iid) + applied[p] += 1 + if not dry_run: + session.execute( + pg_insert(image_tag) + .values( + image_record_id=iid, tag_id=tid, + source=_PRESENTATION_SOURCE, + ) + .on_conflict_do_nothing() + ) + # Guard 2: also looks like content → hide but flag for review. + if Wc is not None and float(max_c[idx]) >= conflict_thr: + n_flagged += 1 + if not dry_run: + session.execute( + pg_insert(PresentationReview) + .values( + image_record_id=iid, tag_id=tid, + conflict_tag_id=conf_tag_ids[int(arg_c[idx])], + conflict_score=float(max_c[idx]), + ) + .on_conflict_do_nothing() + ) + if not dry_run: + session.commit() + + concepts = [ + {"tag_id": pres_tag_ids[p], "name": pres_names[p], + "applied": applied[p], "scanned": scanned, "threshold": thr} + for p in range(len(pres)) + ] + return { + "n_applied": sum(applied), "n_flagged": n_flagged, "concepts": concepts, + } + + def retract_auto_applied_heads(session: Session) -> int: """Soft auto-apply (milestone 139): re-score every standing source='head_auto' tag against its CURRENT head and REMOVE the ones now BELOW the head's diff --git a/backend/app/services/ml/training_data.py b/backend/app/services/ml/training_data.py index 6c9802a..8fcf8cc 100644 --- a/backend/app/services/ml/training_data.py +++ b/backend/app/services/ml/training_data.py @@ -29,7 +29,7 @@ from ...models.tag import image_tag # a CCIP reference) unless the operator confirms them (milestone 139). Keeping # auto-applied predictions out of training is what makes them "soft" — a misfire # can't reinforce itself, so the retraction sweep can actually drop it. -_AUTO_SOURCES = ("head_auto", "ccip_auto", "ml_auto") +_AUTO_SOURCES = ("head_auto", "ccip_auto", "ml_auto", "presentation_auto") def _hygiene_excluded_ids(session: Session) -> set[int]: diff --git a/tests/test_presentation_auto_apply.py b/tests/test_presentation_auto_apply.py new file mode 100644 index 0000000..ac7a4d7 --- /dev/null +++ b/tests/test_presentation_auto_apply.py @@ -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