Activity search + RecoverySweep fix + tagger_predictions shrink (#762, #764) + backup polish #91

Merged
bvandeusen merged 6 commits from dev into main 2026-06-10 14:33:35 -04:00
25 changed files with 616 additions and 50 deletions
@@ -0,0 +1,37 @@
"""ml_settings.tagger_store_floor
The ingest confidence floor below which tagger predictions are not stored,
promoted from the TAGGER_STORE_FLOOR env var to a DB-backed, UI-tunable
setting. Default 0.70 (was an env default of 0.05): the suggestion path
already filters at 0.70 and the centroid/learned path covers low-confidence
preferred tags, so the sub-0.70 tail was redundant weight — it had grown
image_record's TOAST to ~100 GB. See plan-task #764.
Revision ID: 0044
Revises: 0043
Create Date: 2026-06-10
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
revision: str = "0044"
down_revision: Union[str, None] = "0043"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.add_column(
"ml_settings",
sa.Column(
"tagger_store_floor", sa.Float(),
nullable=False, server_default="0.7",
),
)
def downgrade() -> None:
op.drop_column("ml_settings", "tagger_store_floor")
+12
View File
@@ -348,3 +348,15 @@ async def trigger_reextract_archives():
async_result = reextract_archive_attachments_task.delay() async_result = reextract_archive_attachments_task.delay()
return jsonify({"task_id": async_result.id, "status": "queued"}), 202 return jsonify({"task_id": async_result.id, "status": "queued"}), 202
@admin_bp.route("/maintenance/prune-predictions", methods=["POST"])
async def trigger_prune_predictions():
"""Operator-triggered #764 backfill: drop stored tagger predictions below
the current ml_settings.tagger_store_floor and clamp allowlist thresholds
up to it. Shrinks image_record's TOAST (~100 GB of sub-0.70 scores).
Idempotent + self-resuming; runs on the maintenance_long lane."""
from ..tasks.admin import prune_low_confidence_predictions_task
async_result = prune_low_confidence_predictions_task.delay()
return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+35 -1
View File
@@ -13,6 +13,7 @@ _EDITABLE = (
"suggestion_threshold_general", "suggestion_threshold_general",
"centroid_similarity_threshold", "centroid_similarity_threshold",
"min_reference_images", "min_reference_images",
"tagger_store_floor",
) )
@@ -30,6 +31,7 @@ async def get_settings():
"suggestion_threshold_general": s.suggestion_threshold_general, "suggestion_threshold_general": s.suggestion_threshold_general,
"centroid_similarity_threshold": s.centroid_similarity_threshold, "centroid_similarity_threshold": s.centroid_similarity_threshold,
"min_reference_images": s.min_reference_images, "min_reference_images": s.min_reference_images,
"tagger_store_floor": s.tagger_store_floor,
"tagger_model_version": s.tagger_model_version, "tagger_model_version": s.tagger_model_version,
"embedder_model_version": s.embedder_model_version, "embedder_model_version": s.embedder_model_version,
} }
@@ -47,13 +49,45 @@ async def patch_settings():
s = ( s = (
await session.execute(select(MLSettings).where(MLSettings.id == 1)) await session.execute(select(MLSettings).where(MLSettings.id == 1))
).scalar_one() ).scalar_one()
# Merge the patch over current values, then validate the result as a
# whole — the store-floor invariant couples three fields, so they
# can't be checked one at a time.
proposed = {f: getattr(s, f) for f in _EDITABLE}
for field in _EDITABLE: for field in _EDITABLE:
if field in body: if field in body:
setattr(s, field, body[field]) proposed[field] = body[field]
err = _validate(proposed)
if err is not None:
return jsonify({"error": err}), 400
for field in _EDITABLE:
setattr(s, field, proposed[field])
await session.commit() await session.commit()
return await get_settings() return await get_settings()
def _validate(p: dict) -> str | None:
"""Returns an error string if the proposed settings are invalid, else None.
Invariant (plan-task #764): the per-category suggestion thresholds can't
drop below tagger_store_floor — nothing below the floor is stored, so a
lower threshold would silently surface nothing in that gap. The UI clamps
the sliders to the floor; this is the server-side backstop.
"""
floor = p["tagger_store_floor"]
if not (0.0 <= floor <= 1.0):
return "tagger_store_floor must be between 0 and 1"
for cat in ("character", "general"):
if p[f"suggestion_threshold_{cat}"] < floor:
return (
f"suggestion_threshold_{cat} cannot be below tagger_store_floor "
f"({floor}) — predictions below the floor are not stored"
)
return None
@ml_admin_bp.route("/backfill", methods=["POST"]) @ml_admin_bp.route("/backfill", methods=["POST"])
async def trigger_backfill(): async def trigger_backfill():
from ..tasks.ml import backfill from ..tasks.ml import backfill
+13
View File
@@ -147,6 +147,7 @@ async def list_runs():
"""Paginated task_run history. Query params: """Paginated task_run history. Query params:
queue=<name> filter to one queue queue=<name> filter to one queue
status=<status> filter to one status (running/ok/error/timeout/retry) status=<status> filter to one status (running/ok/error/timeout/retry)
task=<substr> case-insensitive substring match on task_name
limit=<int> default 50, max 200 limit=<int> default 50, max 200
before_id=<int> cursor for keyset pagination before_id=<int> cursor for keyset pagination
@@ -161,6 +162,7 @@ async def list_runs():
queue = request.args.get("queue") queue = request.args.get("queue")
status = request.args.get("status") status = request.args.get("status")
task = request.args.get("task")
before_id_raw = request.args.get("before_id") before_id_raw = request.args.get("before_id")
before_id = int(before_id_raw) if before_id_raw else None before_id = int(before_id_raw) if before_id_raw else None
@@ -170,6 +172,11 @@ async def list_runs():
stmt = stmt.where(TaskRun.queue == queue) stmt = stmt.where(TaskRun.queue == queue)
if status: if status:
stmt = stmt.where(TaskRun.status == status) stmt = stmt.where(TaskRun.status == status)
if task:
# Task names contain literal underscores (download_source,
# vacuum_analyze) — escape LIKE wildcards so a search for
# "vacuum_analyze" doesn't treat "_" as a single-char match.
stmt = stmt.where(TaskRun.task_name.ilike(f"%{_escape_like(task)}%", escape="\\"))
if before_id is not None: if before_id is not None:
stmt = stmt.where(TaskRun.id < before_id) stmt = stmt.where(TaskRun.id < before_id)
stmt = stmt.limit(limit + 1) stmt = stmt.limit(limit + 1)
@@ -225,6 +232,12 @@ async def list_failures():
}) })
def _escape_like(value: str) -> str:
"""Escape SQL LIKE/ILIKE metacharacters so user search text is matched
literally. Pairs with `escape="\\"` on the .ilike() call."""
return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
def _row_to_dict(r: TaskRun) -> dict: def _row_to_dict(r: TaskRun) -> dict:
return { return {
"id": r.id, "id": r.id,
+9
View File
@@ -28,6 +28,15 @@ class MLSettings(Base):
centroid_similarity_threshold: Mapped[float] = mapped_column( centroid_similarity_threshold: Mapped[float] = mapped_column(
Float, nullable=False, default=0.55 Float, nullable=False, default=0.55
) )
# Ingest floor: tagger predictions below this confidence are not stored
# (tagger.Tagger.infer). Default 0.70 — the suggestion path already
# filters at 0.70 and the centroid/learned path covers low-confidence
# preferred tags, so the sub-0.70 tail is redundant weight (it had
# bloated image_record's TOAST to ~100 GB; plan-task #764). Operator-
# tunable via Settings → ML; must stay ≤ the suggestion thresholds.
tagger_store_floor: Mapped[float] = mapped_column(
Float, nullable=False, default=0.70
)
min_reference_images: Mapped[int] = mapped_column( min_reference_images: Mapped[int] = mapped_column(
Integer, nullable=False, default=5 Integer, nullable=False, default=5
) )
+21 -11
View File
@@ -24,11 +24,17 @@ from pathlib import Path
_BACKUPS_DIRNAME = "_backups" _BACKUPS_DIRNAME = "_backups"
# Subprocess-level guardrails BEYOND the Celery soft_time_limit. The # Subprocess-level guardrails BEYOND the Celery soft_time_limit. The Celery
# Celery soft limit signals the Python process; subprocess.Popen in a # soft limit signals the Python process; subprocess.Popen in a blocking syscall
# blocking syscall ignores that signal. These bound the worst case. # ignores that signal, so these bound the worst case directly. Each sits just
_DB_SUBPROCESS_TIMEOUT_S = 12 * 60 # 12 min (Celery soft is 10 min) # UNDER its task's Celery soft_time_limit so the bounded-kill (_run_bounded) is
_IMAGES_SUBPROCESS_TIMEOUT_S = 7 * 60 * 60 # 7 hr (Celery soft is 6 hr) # the primary guard and fires cleanly before Celery's soft/hard limits — which
# matters because an NFS D-state hang defeats even Celery's SIGKILL (the failure
# that wedged the maintenance lane for hours, #739).
# backup_db_task: soft=1800s / hard=2100s → 1700s
# backup_images_task: soft=21600s / hard=23400s → 21000s
_DB_SUBPROCESS_TIMEOUT_S = 1700 # ~28 min, under the 30-min DB soft limit
_IMAGES_SUBPROCESS_TIMEOUT_S = 21000 # ~5.8 hr, under the 6-hr images soft limit
# Grace after SIGKILL to reap the child. If it can't be reaped in this window # Grace after SIGKILL to reap the child. If it can't be reaped in this window
# (an uninterruptible NFS D-state — the failure mode that wedged the # (an uninterruptible NFS D-state — the failure mode that wedged the
# concurrency-1 maintenance lane for hours, operator-flagged 2026-06-07), we # concurrency-1 maintenance lane for hours, operator-flagged 2026-06-07), we
@@ -115,18 +121,21 @@ def backup_db(
to persist into BackupRun. Raises on subprocess failure.""" to persist into BackupRun. Raises on subprocess failure."""
ts = _now_ts() ts = _now_ts()
out_dir = _backups_dir(images_root) out_dir = _backups_dir(images_root)
sql_path = out_dir / f"fc_db_{ts}.sql" # Custom format (-Fc): compressed (much smaller on NFS) and restored with
# pg_restore. The .dump extension marks it as non-SQL. The BackupRun field
# is still named sql_path — it's just "the db artifact path".
sql_path = out_dir / f"fc_db_{ts}.dump"
# Dump to LOCAL disk first, then move the finished file to the (NFS) backups # Dump to LOCAL disk first, then move the finished file to the (NFS) backups
# dir. pg_dump's long phase is then a DB-socket wait + local writes — both # dir. pg_dump's long phase is then a DB-socket wait + local writes — both
# killable — instead of an NFS write that can hang uninterruptibly. Only the # killable — instead of an NFS write that can hang uninterruptibly. Only the
# final move touches NFS, and it's a bounded single-file step. # final move touches NFS, and it's a bounded single-file step.
fd, tmp_name = tempfile.mkstemp(prefix="fc_db_", suffix=".sql") fd, tmp_name = tempfile.mkstemp(prefix="fc_db_", suffix=".dump")
os.close(fd) os.close(fd)
tmp_path = Path(tmp_name) tmp_path = Path(tmp_name)
try: try:
_run_bounded( _run_bounded(
[ [
"pg_dump", "--no-owner", "--no-acl", "pg_dump", "--no-owner", "--no-acl", "-Fc",
"-f", str(tmp_path), _libpq_url(db_url), "-f", str(tmp_path), _libpq_url(db_url),
], ],
_DB_SUBPROCESS_TIMEOUT_S, _DB_SUBPROCESS_TIMEOUT_S,
@@ -184,8 +193,8 @@ def backup_images(
def restore_db(*, db_url: str, sql_path: Path) -> None: def restore_db(*, db_url: str, sql_path: Path) -> None:
"""Wipe public schema, then load from .sql. Raises on subprocess """Wipe public schema, then load from the custom-format dump. Raises on
failure; partial-restore state is the caller's concern.""" subprocess failure; partial-restore state is the caller's concern."""
libpq = _libpq_url(db_url) libpq = _libpq_url(db_url)
subprocess.run( subprocess.run(
[ [
@@ -194,8 +203,9 @@ def restore_db(*, db_url: str, sql_path: Path) -> None:
], ],
capture_output=True, check=True, timeout=120, capture_output=True, check=True, timeout=120,
) )
# Custom-format (-Fc) dumps are restored with pg_restore, not psql.
subprocess.run( subprocess.run(
["psql", libpq, "-f", str(sql_path)], ["pg_restore", "--no-owner", "--no-acl", "-d", libpq, str(sql_path)],
capture_output=True, check=True, capture_output=True, check=True,
timeout=_DB_SUBPROCESS_TIMEOUT_S, timeout=_DB_SUBPROCESS_TIMEOUT_S,
) )
+15 -2
View File
@@ -9,7 +9,7 @@ from sqlalchemy import delete, select
from sqlalchemy.dialects.postgresql import insert from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from ...models import Tag, TagAllowlist, TagSuggestionRejection from ...models import MLSettings, Tag, TagAllowlist, TagSuggestionRejection
from ...models.tag import image_tag from ...models.tag import image_tag
from .aliases import AliasService from .aliases import AliasService
@@ -91,12 +91,25 @@ class AllowlistService:
) )
await self.dismiss(image_id, tag_id) await self.dismiss(image_id, tag_id)
async def _store_floor(self) -> float:
return (
await self.session.execute(
select(MLSettings.tagger_store_floor).where(MLSettings.id == 1)
)
).scalar_one()
async def update_threshold( async def update_threshold(
self, tag_id: int, min_confidence: float self, tag_id: int, min_confidence: float
) -> None: ) -> None:
row = await self.session.get(TagAllowlist, tag_id) row = await self.session.get(TagAllowlist, tag_id)
if row is not None: if row is not None:
row.min_confidence = min_confidence # An allowlist tag can't auto-apply more permissively than the
# ingest store floor — predictions below tagger_store_floor aren't
# stored, so a lower min_confidence would behave identically to the
# floor. Clamp so the stored threshold matches actual behavior
# (#764).
floor = await self._store_floor()
row.min_confidence = max(min_confidence, floor)
async def remove(self, tag_id: int) -> None: async def remove(self, tag_id: int) -> None:
await self.session.execute( await self.session.execute(
+14 -6
View File
@@ -33,8 +33,13 @@ _MODEL_DIR = Path(os.environ.get("ML_MODEL_DIR", "/models")) / "camie"
_MODEL_FILE = f"{MODEL_NAME}.onnx" _MODEL_FILE = f"{MODEL_NAME}.onnx"
_METADATA_FILE = f"{MODEL_NAME}-metadata.json" _METADATA_FILE = f"{MODEL_NAME}-metadata.json"
# Below this confidence, predictions aren't stored (keeps the JSON compact). # Ingest floor below which predictions aren't stored (keeps the JSON compact).
STORE_FLOOR = float(os.environ.get("TAGGER_STORE_FLOOR", "0.05")) # DEFAULT/fallback only — the live value is DB-backed
# (ml_settings.tagger_store_floor) and passed into infer() per call by the ml
# task. 0.70: the suggestion path already filters there and the centroid path
# covers lower-confidence preferred tags, so the sub-0.70 tail is redundant
# (it had bloated image_record's TOAST to ~100 GB; plan-task #764).
DEFAULT_STORE_FLOOR = 0.70
# The categories FC-2b surfaces in the UI. Others (meta/rating/year) are # The categories FC-2b surfaces in the UI. Others (meta/rating/year) are
# still stored but the suggestion service filters them out. # still stored but the suggestion service filters them out.
@@ -145,10 +150,13 @@ class Tagger:
arr = arr.transpose(2, 0, 1) # HWC -> CHW arr = arr.transpose(2, 0, 1) # HWC -> CHW
return arr[np.newaxis, :, :, :] # NCHW return arr[np.newaxis, :, :, :] # NCHW
def infer(self, image_path: Path) -> dict[str, TagPrediction]: def infer(
self, image_path: Path, *, store_floor: float = DEFAULT_STORE_FLOOR,
) -> dict[str, TagPrediction]:
"""Run Camie v2 on one image. Returns {name: TagPrediction} with """Run Camie v2 on one image. Returns {name: TagPrediction} with
confidence >= STORE_FLOOR (across all categories — the suggestion confidence >= store_floor (across all categories — the suggestion
service does category filtering later). service does category filtering later). store_floor is the DB-backed
ml_settings.tagger_store_floor, passed in by the ml task.
v2 emits multiple outputs; we use the refined predictions v2 emits multiple outputs; we use the refined predictions
(output[1] per onnx_inference.py). Sigmoid is applied to raw (output[1] per onnx_inference.py). Sigmoid is applied to raw
@@ -167,7 +175,7 @@ class Tagger:
cats = self._tag_categories cats = self._tag_categories
for idx, score in enumerate(probs): for idx, score in enumerate(probs):
conf = float(score) conf = float(score)
if conf < STORE_FLOOR: if conf < store_floor:
continue continue
if idx >= len(names): if idx >= len(names):
# Output longer than metadata declared — shouldn't happen but # Output longer than metadata declared — shouldn't happen but
+96
View File
@@ -207,3 +207,99 @@ def rescan_series_suggestions_task(self, after_post_id: int = 0) -> dict:
) )
rescan_series_suggestions_task.delay(summary["resume_after_id"]) rescan_series_suggestions_task.delay(summary["resume_after_id"])
return summary return summary
@celery.task(
name="backend.app.tasks.admin.prune_low_confidence_predictions_task",
bind=True,
autoretry_for=(OperationalError, DBAPIError),
retry_backoff=15, retry_backoff_max=180, max_retries=1,
soft_time_limit=3600, time_limit=4200, # 60 min / 70 min
)
def prune_low_confidence_predictions_task(self, after_id: int = 0) -> dict:
"""One-time #764 backfill: drop tagger_predictions entries below the DB
store floor (ml_settings.tagger_store_floor) from existing image_record
rows, and clamp any allowlist min_confidence below the floor up to it.
The Camie tagger emits ~10k tags; the old 0.05 floor stored the entire
near-zero tail, bloating image_record's TOAST to ~100 GB. This rewrites
each row to the new floor. Keyset by id ASC (restart-safe via after_id);
idempotent — already-pruned rows rewrite to themselves and are skipped.
Rewriting rows generates bloat, so run VACUUM FULL / pg_repack on
image_record afterward to return the disk to the OS.
The keep predicate (confidence >= floor) mirrors Tagger.infer's store
gate so backfilled rows match what new imports store. Self-resumes on the
soft time limit (re-enqueues from the last committed id)."""
from celery.exceptions import SoftTimeLimitExceeded
from sqlalchemy import select, update
from ..models import ImageRecord, MLSettings, TagAllowlist
SessionLocal = _sync_session_factory()
scanned = 0
pruned = 0
clamped = 0
last_id = after_id
try:
with SessionLocal() as session:
floor = session.execute(
select(MLSettings.tagger_store_floor).where(MLSettings.id == 1)
).scalar_one()
# Clamp allowlist thresholds below the new floor once, on the
# first pass (#764 consumer #4) — a sub-floor min_confidence can't
# apply more permissively now that nothing below it is stored.
if after_id == 0:
clamped = session.execute(
update(TagAllowlist)
.where(TagAllowlist.min_confidence < floor)
.values(min_confidence=floor)
).rowcount or 0
session.commit()
while True:
rows = session.execute(
select(ImageRecord.id, ImageRecord.tagger_predictions)
.where(ImageRecord.id > last_id)
.where(ImageRecord.tagger_predictions.is_not(None))
.order_by(ImageRecord.id.asc())
.limit(500)
).all()
if not rows:
break
for image_id, preds in rows:
scanned += 1
if not preds:
continue
kept = {
name: p for name, p in preds.items()
if float(p.get("confidence", 0.0)) >= floor
}
if len(kept) != len(preds):
session.execute(
update(ImageRecord)
.where(ImageRecord.id == image_id)
.values(tagger_predictions=kept)
)
pruned += 1
session.commit()
last_id = rows[-1].id # advance only after commit, for resume
except SoftTimeLimitExceeded:
log.warning(
"prune_low_confidence_predictions soft-limited at id=%s "
"(scanned=%d pruned=%d) — re-enqueuing", last_id, scanned, pruned,
)
prune_low_confidence_predictions_task.delay(last_id)
return {
"partial": True, "last_id": last_id,
"scanned": scanned, "pruned": pruned,
}
log.info(
"prune_low_confidence_predictions complete: floor=%s scanned=%d "
"pruned=%d allowlist_clamped=%d", floor, scanned, pruned, clamped,
)
return {
"floor": floor, "scanned": scanned, "pruned": pruned,
"allowlist_clamped": clamped, "last_id": last_id,
}
+9
View File
@@ -120,6 +120,15 @@ IMPORT_BATCH_KEEP_DAYS = 30
# files); time_limit=2100. # files); time_limit=2100.
QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = { QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
"ml": 25, "ml": 25,
# download_source legitimately walks 5-25 min (Patreon/gallery-dl
# deep creators); its hard time_limit is DOWNLOAD_HARD_TIME_LIMIT
# (1500s = 25m). The 5-min default flagged healthy in-flight walks as
# phantom 'RecoverySweep' failures (System Activity showed errors the
# Subscriptions view correctly didn't — the download finished ok and
# reset the source). 30 clears the 25-min limit with buffer and lines
# up with DOWNLOAD_STALL_THRESHOLD_MINUTES (30) so a genuine hard kill
# is swept by the task-run AND event sweeps together. Audit 2026-06-10.
"download": 30,
# Audit 2026-06-02 — maintenance/scan queues run tasks that # Audit 2026-06-02 — maintenance/scan queues run tasks that
# legitimately exceed the 5-min default (verify_integrity at 70m # legitimately exceed the 5-min default (verify_integrity at 70m
# hard, scan_directory at 70m hard, apply_allowlist_tags / # hard, scan_directory at 70m hard, apply_allowlist_tags /
+5 -2
View File
@@ -127,7 +127,10 @@ def tag_and_embed(self, image_id: int) -> dict:
phase = "video_infer" phase = "video_infer"
import numpy as np import numpy as np
preds = _maxpool_predictions([tagger.infer(f) for f in frames]) preds = _maxpool_predictions(
[tagger.infer(f, store_floor=settings.tagger_store_floor)
for f in frames]
)
embedding = np.mean( embedding = np.mean(
[embedder.infer(f) for f in frames], axis=0 [embedder.infer(f) for f in frames], axis=0
).astype("float32") ).astype("float32")
@@ -136,7 +139,7 @@ def tag_and_embed(self, image_id: int) -> dict:
else: else:
phase = "tag" phase = "tag"
t0 = time.monotonic() t0 = time.monotonic()
raw = tagger.infer(src) raw = tagger.infer(src, store_floor=settings.tagger_store_floor)
log.info( log.info(
"tag_and_embed tagged in %.1fs (%d tags): %s", "tag_and_embed tagged in %.1fs (%d tags): %s",
time.monotonic() - t0, len(raw), ctx, time.monotonic() - t0, len(raw), ctx,
@@ -10,7 +10,7 @@
<v-text-field <v-text-field
:model-value="item.min_confidence" type="number" :model-value="item.min_confidence" type="number"
density="compact" hide-details style="max-width: 100px;" density="compact" hide-details style="max-width: 100px;"
min="0.05" max="1" step="0.05" :min="floor" max="1" step="0.05"
@update:model-value="(v) => onThreshold(item.tag_id, v)" @update:model-value="(v) => onThreshold(item.tag_id, v)"
/> />
</template> </template>
@@ -25,23 +25,32 @@
</template> </template>
<script setup> <script setup>
import { onMounted } from 'vue' import { computed, onMounted } from 'vue'
import { useAllowlistStore } from '../../stores/allowlist.js' import { useAllowlistStore } from '../../stores/allowlist.js'
import { useMLStore } from '../../stores/ml.js'
const store = useAllowlistStore() const store = useAllowlistStore()
const ml = useMLStore()
// min_confidence can't be set below the tagger store floor — predictions
// below it aren't stored, so a lower threshold would behave identically to
// the floor. The backend clamps too (#764).
const floor = computed(() => ml.settings?.tagger_store_floor ?? 0.70)
const headers = [ const headers = [
{ title: 'Tag', key: 'tag_name', sortable: true }, { title: 'Tag', key: 'tag_name', sortable: true },
{ title: 'Kind', key: 'tag_kind', sortable: true, width: 110 }, { title: 'Kind', key: 'tag_kind', sortable: true, width: 110 },
{ title: 'Min confidence', key: 'min_confidence', sortable: false, width: 140 }, { title: 'Min confidence', key: 'min_confidence', sortable: false, width: 140 },
{ title: '', key: 'actions', sortable: false, width: 60 } { title: '', key: 'actions', sortable: false, width: 60 }
] ]
onMounted(() => store.load()) onMounted(() => {
store.load()
if (!ml.settings) ml.loadSettings()
})
let debounce = null let debounce = null
function onThreshold(tagId, value) { function onThreshold(tagId, value) {
if (debounce) clearTimeout(debounce) if (debounce) clearTimeout(debounce)
debounce = setTimeout(() => { debounce = setTimeout(() => {
const v = parseFloat(value) const v = Math.max(parseFloat(value), floor.value)
if (v > 0 && v <= 1) store.updateThreshold(tagId, v) if (v > 0 && v <= 1) store.updateThreshold(tagId, v)
}, 500) }, 500)
} }
@@ -6,9 +6,28 @@
<v-col cols="12"> <v-col cols="12">
<v-slider <v-slider
v-model="local[f.key]" :label="f.label" v-model="local[f.key]" :label="f.label"
:min="f.floorMin ? local.tagger_store_floor : 0" max="1" step="0.05"
thumb-label hide-details
color="accent" @end="save"
/>
</v-col>
</v-row>
<v-divider class="my-4" />
<v-row>
<v-col cols="12">
<v-slider
v-model="local.tagger_store_floor" label="Tagger store floor"
min="0" max="1" step="0.05" thumb-label hide-details min="0" max="1" step="0.05" thumb-label hide-details
color="accent" @end="save" color="accent" @end="save"
/> />
<div class="text-caption fc-muted mt-1">
Tagger predictions below this confidence aren't stored — raising it
keeps the image library lean. Suggestions can't be shown below the
floor; lower-confidence tags you actually want still surface through
the learned centroid path.
</div>
</v-col> </v-col>
</v-row> </v-row>
</v-card-text> </v-card-text>
@@ -24,17 +43,25 @@ import { useMLStore } from '../../stores/ml.js'
const store = useMLStore() const store = useMLStore()
// 'artist' (FC-2d-vii-c) and 'copyright' (2026-06-01) retired as // 'artist' (FC-2d-vii-c) and 'copyright' (2026-06-01) retired as
// suggestion categories; their threshold rows are gone. // suggestion categories; their threshold rows are gone.
// floorMin: the per-category suggestion thresholds can't drop below the
// tagger store floor (nothing below the floor is stored to surface).
const fields = [ const fields = [
{ key: 'suggestion_threshold_character', label: 'Character' }, { key: 'suggestion_threshold_character', label: 'Character', floorMin: true },
{ key: 'suggestion_threshold_general', label: 'General' }, { key: 'suggestion_threshold_general', label: 'General', floorMin: true },
{ key: 'centroid_similarity_threshold', label: 'Centroid similarity' } { key: 'centroid_similarity_threshold', label: 'Centroid similarity' }
] ]
const local = reactive({}) const local = reactive({})
watch(() => store.settings, (s) => { if (s) Object.assign(local, s) }, { immediate: true }) watch(() => store.settings, (s) => { if (s) Object.assign(local, s) }, { immediate: true })
async function save() { async function save() {
// Mirror the server invariant: keep the category thresholds at or above the
// store floor so a raised floor doesn't leave a threshold stranded below it.
const floor = local.tagger_store_floor
local.suggestion_threshold_character = Math.max(local.suggestion_threshold_character, floor)
local.suggestion_threshold_general = Math.max(local.suggestion_threshold_general, floor)
const patch = {} const patch = {}
for (const f of fields) patch[f.key] = local[f.key] for (const f of fields) patch[f.key] = local[f.key]
patch.tagger_store_floor = local.tagger_store_floor
try { await store.patchSettings(patch) } try { await store.patchSettings(patch) }
catch (e) { toast({ text: e.message, type: 'error' }) } catch (e) { toast({ text: e.message, type: 'error' }) }
} }
@@ -12,6 +12,7 @@
<ThumbnailBackfillCard /> <ThumbnailBackfillCard />
</div> </div>
<MLThresholdSliders class="mt-4" /> <MLThresholdSliders class="mt-4" />
<PrunePredictionsCard class="mt-4" />
<AllowlistTable class="mt-4" /> <AllowlistTable class="mt-4" />
<AliasTable class="mt-4" /> <AliasTable class="mt-4" />
<DbMaintenanceCard class="mt-6" /> <DbMaintenanceCard class="mt-6" />
@@ -31,6 +32,7 @@ import MLBackfillCard from './MLBackfillCard.vue'
import CentroidRecomputeCard from './CentroidRecomputeCard.vue' import CentroidRecomputeCard from './CentroidRecomputeCard.vue'
import ThumbnailBackfillCard from './ThumbnailBackfillCard.vue' import ThumbnailBackfillCard from './ThumbnailBackfillCard.vue'
import MLThresholdSliders from './MLThresholdSliders.vue' import MLThresholdSliders from './MLThresholdSliders.vue'
import PrunePredictionsCard from './PrunePredictionsCard.vue'
import AllowlistTable from './AllowlistTable.vue' import AllowlistTable from './AllowlistTable.vue'
import AliasTable from './AliasTable.vue' import AliasTable from './AliasTable.vue'
import DbMaintenanceCard from './DbMaintenanceCard.vue' import DbMaintenanceCard from './DbMaintenanceCard.vue'
@@ -0,0 +1,58 @@
<template>
<!-- #764: drop stored tagger predictions below the store floor to shrink
image_record's TOAST (the sub-0.70 score tail had grown it to ~100 GB). -->
<v-card>
<v-card-title>Prune low-confidence predictions</v-card-title>
<v-card-text>
<p class="text-body-2 mb-3">
Removes stored tagger predictions below the current store floor
(<strong>{{ floorPct }}</strong>) from every image, and clamps any
allowlist threshold below the floor up to it. This is what shrinks the
database — the low-confidence tail was the bulk of its size. Idempotent
and resumable; safe to run more than once. Afterward, reclaim the freed
space with <code>VACUUM FULL</code> / <code>pg_repack</code> on
<code>image_record</code>.
</p>
<v-btn color="primary" rounded="pill" :loading="busy" @click="run">
<v-icon start>mdi-database-minus-outline</v-icon> Prune predictions now
</v-btn>
<span v-if="queued" class="ml-3 text-caption text-success">Queued ✓</span>
<QueueStatusBar queue="maintenance_long" queue-label="Maintenance (long)" />
</v-card-text>
</v-card>
</template>
<script setup>
import { computed, onMounted, ref } from 'vue'
import { useApi } from '../../composables/useApi.js'
import { toast } from '../../utils/toast.js'
import { useMLStore } from '../../stores/ml.js'
import QueueStatusBar from './QueueStatusBar.vue'
const api = useApi()
const ml = useMLStore()
const busy = ref(false)
const queued = ref(false)
const floorPct = computed(() => {
const f = ml.settings?.tagger_store_floor
return f == null ? '' : `${Math.round(f * 100)}%`
})
onMounted(() => { if (!ml.settings) ml.loadSettings() })
async function run () {
busy.value = true
queued.value = false
try {
await api.post('/api/admin/maintenance/prune-predictions')
queued.value = true
toast({ text: 'Prediction prune queued', type: 'success' })
} catch (e) {
toast({ text: e?.body?.detail || e?.message || 'Failed to queue', type: 'error' })
} finally {
busy.value = false
}
}
</script>
@@ -21,8 +21,15 @@
<v-card class="mb-4"> <v-card class="mb-4">
<CardHeading icon="mdi-alert-circle-outline" title="Recent failures (last 24h)"> <CardHeading icon="mdi-alert-circle-outline" title="Recent failures (last 24h)">
<v-spacer /> <v-spacer />
<span class="text-caption fc-muted"> <v-text-field
{{ failureCount }} failures v-model="failureSearch"
placeholder="Search task, queue, target, error…"
density="compact" hide-details clearable
prepend-inner-icon="mdi-magnify"
style="max-width: 280px;"
/>
<span class="text-caption fc-muted ml-3">
{{ filteredFailures.length }} / {{ failureCount }}
</span> </span>
</CardHeading> </CardHeading>
<v-card-text> <v-card-text>
@@ -77,6 +84,14 @@
<v-card> <v-card>
<CardHeading icon="mdi-format-list-bulleted" title="All recent activity"> <CardHeading icon="mdi-format-list-bulleted" title="All recent activity">
<v-spacer /> <v-spacer />
<v-text-field
v-model="filterTask"
placeholder="Search task…"
density="compact" hide-details clearable
prepend-inner-icon="mdi-magnify"
style="max-width: 220px;"
@update:model-value="onTaskSearch"
/>
<v-select <v-select
v-model="filterQueue" v-model="filterQueue"
:items="queueOptions" density="compact" hide-details :items="queueOptions" density="compact" hide-details
@@ -170,6 +185,8 @@ const store = useSystemActivityStore()
const filterQueue = ref(null) const filterQueue = ref(null)
const filterStatus = ref(null) const filterStatus = ref(null)
const filterErrorType = ref(null) const filterErrorType = ref(null)
const filterTask = ref(null) // server-side task-name search (All activity)
const failureSearch = ref('') // client-side search over loaded failures
const queueOptions = [ const queueOptions = [
{ title: 'All queues', value: null }, { title: 'All queues', value: null },
@@ -225,9 +242,18 @@ const errorTypes = computed(() => {
const failureCount = computed(() => (store.failures?.recent || []).length) const failureCount = computed(() => (store.failures?.recent || []).length)
const filteredFailures = computed(() => { const filteredFailures = computed(() => {
const all = store.failures?.recent || [] let all = store.failures?.recent || []
if (!filterErrorType.value) return all if (filterErrorType.value) {
return all.filter(r => r.error_type === filterErrorType.value) all = all.filter(r => r.error_type === filterErrorType.value)
}
const q = (failureSearch.value || '').trim().toLowerCase()
if (q) {
all = all.filter(r =>
[r.task_name, r.queue, r.target_id, r.error_type]
.some(v => String(v ?? '').toLowerCase().includes(q)),
)
}
return all
}) })
function toggleErrorTypeFilter(name) { function toggleErrorTypeFilter(name) {
@@ -240,9 +266,23 @@ function toggleErrorTypeFilter(name) {
} }
function onFilterChange() { function onFilterChange() {
store.setFilter({ queue: filterQueue.value, status: filterStatus.value }) store.setFilter({
queue: filterQueue.value,
status: filterStatus.value,
task: (filterTask.value || '').trim() || null,
})
store.loadRuns({ reset: true }) store.loadRuns({ reset: true })
} }
// Server-side search hits the full task_run history, not just the loaded
// page — debounce so we don't fire a query per keystroke (matches the
// 300ms idiom in TagsView/AllowlistTable).
let taskSearchDebounce = null
function onTaskSearch() {
if (taskSearchDebounce) clearTimeout(taskSearchDebounce)
taskSearchDebounce = setTimeout(onFilterChange, 300)
}
function onLoadMore() { store.loadRuns({ reset: false }) } function onLoadMore() { store.loadRuns({ reset: false }) }
function onRefresh() { store.loadRuns({ reset: true }) } function onRefresh() { store.loadRuns({ reset: true }) }
+2 -1
View File
@@ -16,7 +16,7 @@ export const useSystemActivityStore = defineStore('systemActivity', () => {
const runs = ref([]) const runs = ref([])
const runsCursor = ref(null) const runsCursor = ref(null)
const runsHasMore = ref(false) const runsHasMore = ref(false)
const runsFilter = ref({ queue: null, status: null, limit: 50 }) const runsFilter = ref({ queue: null, status: null, task: null, limit: 50 })
const loading = ref({ queues: false, workers: false, runs: false, failures: false }) const loading = ref({ queues: false, workers: false, runs: false, failures: false })
const lastError = ref(null) const lastError = ref(null)
@@ -65,6 +65,7 @@ export const useSystemActivityStore = defineStore('systemActivity', () => {
const params = { limit: runsFilter.value.limit } const params = { limit: runsFilter.value.limit }
if (runsFilter.value.queue) params.queue = runsFilter.value.queue if (runsFilter.value.queue) params.queue = runsFilter.value.queue
if (runsFilter.value.status) params.status = runsFilter.value.status if (runsFilter.value.status) params.status = runsFilter.value.status
if (runsFilter.value.task) params.task = runsFilter.value.task
if (!reset && runsCursor.value) params.before_id = runsCursor.value if (!reset && runsCursor.value) params.before_id = runsCursor.value
const body = await api.get('/api/system/activity/runs', { params }) const body = await api.get('/api/system/activity/runs', { params })
runs.value = reset ? body.runs : [...runs.value, ...body.runs] runs.value = reset ? body.runs : [...runs.value, ...body.runs]
+22
View File
@@ -34,6 +34,28 @@ async def test_get_and_patch_settings(client):
assert (await resp.get_json())["suggestion_threshold_general"] == pytest.approx(0.90) assert (await resp.get_json())["suggestion_threshold_general"] == pytest.approx(0.90)
@pytest.mark.asyncio
async def test_tagger_store_floor_default_and_patch(client):
body = await (await client.get("/api/ml/settings")).get_json()
assert body["tagger_store_floor"] == pytest.approx(0.70)
resp = await client.patch("/api/ml/settings", json={"tagger_store_floor": 0.6})
assert resp.status_code == 200
assert (await resp.get_json())["tagger_store_floor"] == pytest.approx(0.6)
@pytest.mark.asyncio
async def test_suggestion_threshold_below_store_floor_rejected(client):
# Invariant (#764): a category threshold can't sit below the store floor —
# nothing below the floor is stored, so the gap would surface nothing.
# Floor defaults to 0.70; pushing general down to 0.50 must 400.
resp = await client.patch(
"/api/ml/settings", json={"suggestion_threshold_general": 0.50}
)
assert resp.status_code == 400
assert "tagger_store_floor" in (await resp.get_json())["error"]
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_backfill_and_recompute_trigger(client): async def test_backfill_and_recompute_trigger(client):
r1 = await client.post("/api/ml/backfill") r1 = await client.post("/api/ml/backfill")
+19
View File
@@ -182,6 +182,25 @@ async def test_runs_filter_by_status(client, _seed_runs):
assert len(body["runs"]) == 2 # i=3, i=4 assert len(body["runs"]) == 2 # i=3, i=4
@pytest.mark.asyncio
async def test_runs_filter_by_task_substring(client, _seed_runs):
# Case-insensitive substring across the full task_name.
resp = await client.get("/api/system/activity/runs?task=FAKE")
body = await resp.get_json()
assert len(body["runs"]) == 5
assert all("fake" in r["task_name"] for r in body["runs"])
@pytest.mark.asyncio
async def test_runs_filter_by_task_escapes_underscore(client, _seed_runs):
# The literal "_" in "task_3" must match one row, not act as a
# single-char wildcard matching every task_N.
resp = await client.get("/api/system/activity/runs?task=task_3")
body = await resp.get_json()
assert len(body["runs"]) == 1
assert body["runs"][0]["task_name"].endswith("task_3")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_runs_keyset_cursor(client, _seed_runs): async def test_runs_keyset_cursor(client, _seed_runs):
page1 = await (await client.get("/api/system/activity/runs?limit=2")).get_json() page1 = await (await client.get("/api/system/activity/runs?limit=2")).get_json()
+15 -7
View File
@@ -69,7 +69,7 @@ def test_backup_db_writes_sql_and_manifest(tmp_path, fake_subprocess):
) )
sql = Path(result["sql_path"]) sql = Path(result["sql_path"])
manifest = Path(result["manifest_path"]) manifest = Path(result["manifest_path"])
assert sql.is_file() and sql.suffix == ".sql" assert sql.is_file() and sql.suffix == ".dump"
assert manifest.is_file() and manifest.suffix == ".json" assert manifest.is_file() and manifest.suffix == ".json"
assert result["kind"] == "db" assert result["kind"] == "db"
assert result["tar_path"] is None assert result["tar_path"] is None
@@ -92,6 +92,12 @@ def test_backup_db_strips_sqlalchemy_psycopg_driver(tmp_path, fake_subprocess):
assert "+psycopg" not in cmd[-1] assert "+psycopg" not in cmd[-1]
def test_backup_db_uses_compressed_custom_format(tmp_path, fake_subprocess):
# -Fc → pg_restore-loadable, compressed dump (#739 backup polish).
backup_service.backup_db(db_url="postgresql://u@h/d", images_root=tmp_path)
assert "-Fc" in fake_subprocess[0]
def test_backup_db_strips_asyncpg_driver(tmp_path, fake_subprocess): def test_backup_db_strips_asyncpg_driver(tmp_path, fake_subprocess):
backup_service.backup_db( backup_service.backup_db(
db_url="postgresql+asyncpg://u:p@h/d", images_root=tmp_path, db_url="postgresql+asyncpg://u:p@h/d", images_root=tmp_path,
@@ -132,17 +138,19 @@ def test_backup_images_excludes_backups_and_quarantine(tmp_path, fake_subprocess
def test_restore_db_drops_schema_then_loads(tmp_path, fake_subprocess): def test_restore_db_drops_schema_then_loads(tmp_path, fake_subprocess):
sql_path = tmp_path / "fake.sql" dump_path = tmp_path / "fake.dump"
sql_path.write_text("SELECT 1;") dump_path.write_bytes(b"\x00fake custom dump")
backup_service.restore_db( backup_service.restore_db(
db_url="postgresql://u@h/d", sql_path=sql_path, db_url="postgresql://u@h/d", sql_path=dump_path,
) )
# Two psql calls: one with -c (DROP SCHEMA), one with -f (load). # Two calls: psql -c (DROP SCHEMA), then pg_restore -d (load the dump).
assert len(fake_subprocess) == 2 assert len(fake_subprocess) == 2
assert fake_subprocess[0][0] == "psql"
assert "-c" in fake_subprocess[0] assert "-c" in fake_subprocess[0]
assert "DROP SCHEMA IF EXISTS public CASCADE" in fake_subprocess[0][-1] assert "DROP SCHEMA IF EXISTS public CASCADE" in fake_subprocess[0][-1]
assert "-f" in fake_subprocess[1] assert fake_subprocess[1][0] == "pg_restore"
assert str(sql_path) in fake_subprocess[1] assert "-d" in fake_subprocess[1]
assert str(dump_path) in fake_subprocess[1]
# --- restore_images -------------------------------------------------- # --- restore_images --------------------------------------------------
+56
View File
@@ -350,6 +350,62 @@ def test_recover_stalled_task_runs_ml_queue_uses_longer_threshold(db_sync):
assert ml_stale_status == "error" assert ml_stale_status == "error"
def test_recover_stalled_task_runs_download_queue_uses_longer_threshold(db_sync):
"""download_source legitimately walks 5-25 min (Patreon/gallery-dl).
The 5-min default flagged healthy in-flight walks as phantom
'RecoverySweep' failures — visible in System Activity but absent from
the Subscriptions view because the download actually finished ok.
The 30-min download override (QUEUE_STUCK_THRESHOLD_MINUTES) must
protect a 10-min-old download row while still flagging a 35-min one.
Audit 2026-06-10."""
from sqlalchemy import select
from backend.app.models import TaskRun
from backend.app.tasks.maintenance import recover_stalled_task_runs
now = datetime.now(UTC)
# 10-min-old download row: stale by the default 5-min rule but fresh
# by the 30-min download override. Must survive the sweep.
dl_fresh_id = _make_task_run(
db_sync, status="running", queue="download",
task_name="backend.app.tasks.download.download_source",
started_at=now - timedelta(minutes=10),
)
# 35-min-old download row: past even the 30-min override (a genuine
# hard kill). Must be flagged.
dl_stale_id = _make_task_run(
db_sync, status="running", queue="download",
task_name="backend.app.tasks.download.download_source",
started_at=now - timedelta(minutes=35),
)
db_sync.commit()
recovered = recover_stalled_task_runs.apply().get()
assert recovered == 1
db_sync.expire_all()
dl_fresh_status = db_sync.execute(
select(TaskRun.status).where(TaskRun.id == dl_fresh_id)
).scalar_one()
dl_stale_status = db_sync.execute(
select(TaskRun.status).where(TaskRun.id == dl_stale_id)
).scalar_one()
assert dl_fresh_status == "running"
assert dl_stale_status == "error"
def test_download_stuck_threshold_exceeds_hard_time_limit():
"""Invariant guard (maintenance.py:112): every queue override MUST be
≥ the relevant task's hard time_limit, else the sweep flags in-flight
work. download_source is the one that regressed — pin it so a future
DOWNLOAD_HARD_TIME_LIMIT bump can't silently re-break it."""
from backend.app.tasks.download import DOWNLOAD_HARD_TIME_LIMIT
from backend.app.tasks.maintenance import QUEUE_STUCK_THRESHOLD_MINUTES
hard_minutes = DOWNLOAD_HARD_TIME_LIMIT / 60
assert QUEUE_STUCK_THRESHOLD_MINUTES["download"] >= hard_minutes
def test_recover_stalled_task_runs_archive_task_uses_longer_threshold(db_sync): def test_recover_stalled_task_runs_archive_task_uses_longer_threshold(db_sync):
"""import_archive_file shares the 'import' queue with fast """import_archive_file shares the 'import' queue with fast
single-file import_media_file, so it gets a per-task-name override single-file import_media_file, so it gets a per-task-name override
+14
View File
@@ -104,3 +104,17 @@ async def test_update_threshold_and_remove(db):
assert abs(row.min_confidence - 0.80) < 1e-6 assert abs(row.min_confidence - 0.80) < 1e-6
await svc.remove(tag.id) await svc.remove(tag.id)
assert await db.get(TagAllowlist, tag.id) is None assert await db.get(TagAllowlist, tag.id) is None
@pytest.mark.asyncio
async def test_update_threshold_clamped_to_store_floor(db):
# A min_confidence below the store floor (default 0.70) is clamped up —
# predictions below the floor aren't stored, so a lower threshold can't
# apply more permissively than the floor (#764).
tag = await TagService(db).find_or_create("Lowthr", TagKind.general)
svc = AllowlistService(db)
img = await _make_image(db)
await svc.accept(img.id, tag.id)
await svc.update_threshold(tag.id, 0.30)
row = await db.get(TagAllowlist, tag.id)
assert abs(row.min_confidence - 0.70) < 1e-6
+11 -7
View File
@@ -1,14 +1,14 @@
"""Tagger unit tests. The ONNX model isn't available in CI (it's a 1GB """Tagger unit tests. The ONNX model isn't available in CI (it's a 1GB
download into /models), so these test the pure-logic surface: STORE_FLOOR download into /models), so these test the pure-logic surface:
constant, SURFACED_CATEGORIES set, TagPrediction dataclass, and the DEFAULT_STORE_FLOOR constant, SURFACED_CATEGORIES set, TagPrediction
load()-missing-file error path. Full inference is exercised by the local dataclass, and the load()-missing-file error path. Full inference is
integration suite against a real /models volume. exercised by the local integration suite against a real /models volume.
""" """
import pytest import pytest
from backend.app.services.ml.tagger import ( from backend.app.services.ml.tagger import (
STORE_FLOOR, DEFAULT_STORE_FLOOR,
SURFACED_CATEGORIES, SURFACED_CATEGORIES,
Tagger, Tagger,
TagPrediction, TagPrediction,
@@ -26,8 +26,12 @@ def test_surfaced_categories():
assert "copyright" not in SURFACED_CATEGORIES assert "copyright" not in SURFACED_CATEGORIES
def test_store_floor_is_low(): def test_default_store_floor():
assert 0 < STORE_FLOOR < 0.2 # Raised 0.05 → 0.70 (plan-task #764): the suggestion path filters at
# 0.70 and the centroid path covers lower-confidence preferred tags, so
# storing the sub-0.70 tail was redundant (100 GB of TOAST). The live
# value is DB-backed (ml_settings.tagger_store_floor); this is the default.
assert DEFAULT_STORE_FLOOR == 0.70
def test_tag_prediction_dataclass(): def test_tag_prediction_dataclass():
+62
View File
@@ -42,6 +42,68 @@ def test_bulk_delete_images_task_registered():
) )
def test_prune_low_confidence_predictions_task_registered():
assert (
"backend.app.tasks.admin.prune_low_confidence_predictions_task"
in celery.tasks
)
@pytest.mark.asyncio
async def test_prune_low_confidence_predictions(db_sync, tmp_path):
# #764: drop stored tagger predictions below the store floor (default
# 0.70) and clamp allowlist thresholds up to it.
from backend.app.models import Tag, TagAllowlist, TagKind
from backend.app.tasks.admin import prune_low_confidence_predictions_task
f0 = tmp_path / "p0.jpg"
f0.write_bytes(b"x")
img0 = ImageRecord(
path=str(f0), sha256=f"{0:064x}", size_bytes=10, mime="image/jpeg",
origin="imported_filesystem",
tagger_predictions={
"keep_high": {"category": "general", "confidence": 0.92},
"keep_edge": {"category": "general", "confidence": 0.70},
"drop_mid": {"category": "general", "confidence": 0.40},
"drop_tiny": {"category": "general", "confidence": 0.06},
},
)
db_sync.add(img0)
f1 = tmp_path / "p1.jpg"
f1.write_bytes(b"x")
img1 = ImageRecord(
path=str(f1), sha256=f"{1:064x}", size_bytes=10, mime="image/jpeg",
origin="imported_filesystem",
tagger_predictions={"only": {"category": "general", "confidence": 0.99}},
)
db_sync.add(img1)
tag = Tag(name="lowthr-tag", kind=TagKind.general)
db_sync.add(tag)
db_sync.flush()
db_sync.add(TagAllowlist(tag_id=tag.id, min_confidence=0.30))
db_sync.commit()
img0_id, img1_id, tag_id = img0.id, img1.id, tag.id
result = prune_low_confidence_predictions_task.delay().get()
assert result["floor"] == pytest.approx(0.70)
assert result["pruned"] == 1 # only img0 had sub-floor entries
assert result["allowlist_clamped"] == 1
db_sync.expire_all()
p0 = db_sync.execute(
select(ImageRecord.tagger_predictions).where(ImageRecord.id == img0_id)
).scalar_one()
assert set(p0) == {"keep_high", "keep_edge"} # >=0.70 kept, <0.70 dropped
p1 = db_sync.execute(
select(ImageRecord.tagger_predictions).where(ImageRecord.id == img1_id)
).scalar_one()
assert set(p1) == {"only"} # already clean — untouched
clamped = db_sync.execute(
select(TagAllowlist.min_confidence).where(TagAllowlist.tag_id == tag_id)
).scalar_one()
assert clamped == pytest.approx(0.70)
# --- delete_artist_cascade_task ------------------------------------- # --- delete_artist_cascade_task -------------------------------------
+1 -1
View File
@@ -93,7 +93,7 @@ async def test_backup_db_task_creates_backup_run_row_status_ok(db_sync):
).one() ).one()
assert row.kind == "db" assert row.kind == "db"
assert row.status == "ok" assert row.status == "ok"
assert row.sql_path and row.sql_path.endswith(".sql") assert row.sql_path and row.sql_path.endswith(".dump")
assert row.size_bytes is not None and row.size_bytes > 0 assert row.size_bytes is not None and row.size_bytes > 0
assert row.finished_at is not None assert row.finished_at is not None
assert row.error is None assert row.error is None