From 015760609a099ac9dc89f49980023754f881013a Mon Sep 17 00:00:00 2001 From: davidamacey Date: Thu, 15 Oct 2026 18:27:22 -0400 Subject: [PATCH 01/10] feat(audit): serve the audit state on the item wire (#166) audit_sample, audit_batch_id, audit_sampled_at, audit_label_name, audit_label_source, audit_outcome and audit_human_name go through serialize_item, ItemDoc and the contracts; the audit queue shows the machine label a crop was drawn with. --- CHANGELOG.md | 1 + contracts/json/item_wire.json | 97 +++++ contracts/openapi/curation.json | 332 ++++++++++++++++++ contracts/ts/itemWire.ts | 14 + docs/design/curation_api_contract.md | 1 + frontend/CHANGELOG.md | 1 + frontend/src/lib/api_cropMap.ts | 17 + .../components/audit/AuditQueueList.svelte | 8 + .../src/lib/components/audit/audit.test.ts | 20 ++ frontend/src/lib/test/makeItem.ts | 5 + frontend/src/lib/types.ts | 6 + src/routers/curation/_item_models.py | 10 + src/services/curation/wire.py | 3 + tests/curation/test_wire_audit_fields.py | 33 ++ 14 files changed, 548 insertions(+) create mode 100644 tests/curation/test_wire_audit_fields.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 09eecdddd..a441645c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ history of this codebase and was never published. This release is `[0.5.0]`. ## [Unreleased] ### Added +- The item wire serves the accuracy audit state (#166): `audit_sample`, `audit_batch_id`, `audit_sampled_at`, `audit_label_name`, `audit_label_source`, `audit_outcome` and `audit_human_name` (null for an item never drawn), so `GET /audit/queue` items can show the machine label they were drawn with. - Cropwright **Assist…** (cluster page, one crop, one box, Matching items): a dialog to pick what the VLM sees (selected, all, a representative sample), the actions (classify, attributes, verify, segment) and the VLM and prompt pack, with the server's dry run; the cluster and sub-cluster label proposal card with the agreement, an explicit dry-run-first **Apply** and the `scope_changed` refusal; dashed "VLM suggests" attribute chips; and a **Region verification** switch (`region_verify`) on the VLM scope panel. See the guide [Selective VLM and segmenter assist](docs-site/docs/guides/selective-assist.mdx). - Pack attributes in whole-project VLM runs: `vlm_attributes=true` on `POST /pipeline/auto_label` and `/start` makes the VLM stage use the pack's combined `assist` call and store the item attributes in each item's `assist` slot. The region worker's combined call asks for the pack's item attributes on the items it classifies and writes them to `assist.attributes` too. diff --git a/contracts/json/item_wire.json b/contracts/json/item_wire.json index 07f7ef2fa..f39657791 100644 --- a/contracts/json/item_wire.json +++ b/contracts/json/item_wire.json @@ -58,6 +58,13 @@ "excluded_reason", "excluded_at", "review_dismissed_at", + "audit_sample", + "audit_batch_id", + "audit_sampled_at", + "audit_label_name", + "audit_label_source", + "audit_outcome", + "audit_human_name", "source", "test_holdout", "crop_rank_in_image", @@ -364,6 +371,96 @@ ], "default": null }, + "audit_batch_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Audit Batch Id" + }, + "audit_human_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Audit Human Name" + }, + "audit_label_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Audit Label Name" + }, + "audit_label_source": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Audit Label Source" + }, + "audit_outcome": { + "anyOf": [ + { + "enum": [ + "agree", + "detector_wrong", + "vlm_wrong", + "both_wrong" + ], + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Audit Outcome" + }, + "audit_sample": { + "anyOf": [ + { + "type": "boolean" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Audit Sample" + }, + "audit_sampled_at": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Audit Sampled At" + }, "bbox_norm": { "items": { "type": "number" diff --git a/contracts/openapi/curation.json b/contracts/openapi/curation.json index 730076185..b30afab91 100644 --- a/contracts/openapi/curation.json +++ b/contracts/openapi/curation.json @@ -11333,6 +11333,89 @@ } ] }, + "audit_batch_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Batch Id" + }, + "audit_human_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Human Name" + }, + "audit_label_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Label Name" + }, + "audit_label_source": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Label Source" + }, + "audit_outcome": { + "anyOf": [ + { + "enum": [ + "agree", + "detector_wrong", + "vlm_wrong", + "both_wrong" + ], + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Outcome" + }, + "audit_sample": { + "anyOf": [ + { + "type": "boolean" + }, + { + "type": "null" + } + ], + "title": "Audit Sample" + }, + "audit_sampled_at": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Sampled At" + }, "bbox_norm": { "items": { "type": "number" @@ -20539,6 +20622,89 @@ } ] }, + "audit_batch_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Batch Id" + }, + "audit_human_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Human Name" + }, + "audit_label_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Label Name" + }, + "audit_label_source": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Label Source" + }, + "audit_outcome": { + "anyOf": [ + { + "enum": [ + "agree", + "detector_wrong", + "vlm_wrong", + "both_wrong" + ], + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Outcome" + }, + "audit_sample": { + "anyOf": [ + { + "type": "boolean" + }, + { + "type": "null" + } + ], + "title": "Audit Sample" + }, + "audit_sampled_at": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Sampled At" + }, "bbox_norm": { "items": { "type": "number" @@ -24175,6 +24341,89 @@ } ] }, + "audit_batch_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Batch Id" + }, + "audit_human_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Human Name" + }, + "audit_label_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Label Name" + }, + "audit_label_source": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Label Source" + }, + "audit_outcome": { + "anyOf": [ + { + "enum": [ + "agree", + "detector_wrong", + "vlm_wrong", + "both_wrong" + ], + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Outcome" + }, + "audit_sample": { + "anyOf": [ + { + "type": "boolean" + }, + { + "type": "null" + } + ], + "title": "Audit Sample" + }, + "audit_sampled_at": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Sampled At" + }, "bbox_norm": { "items": { "type": "number" @@ -25724,6 +25973,89 @@ } ] }, + "audit_batch_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Batch Id" + }, + "audit_human_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Human Name" + }, + "audit_label_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Label Name" + }, + "audit_label_source": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Label Source" + }, + "audit_outcome": { + "anyOf": [ + { + "enum": [ + "agree", + "detector_wrong", + "vlm_wrong", + "both_wrong" + ], + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Outcome" + }, + "audit_sample": { + "anyOf": [ + { + "type": "boolean" + }, + { + "type": "null" + } + ], + "title": "Audit Sample" + }, + "audit_sampled_at": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Audit Sampled At" + }, "bbox_norm": { "items": { "type": "number" diff --git a/contracts/ts/itemWire.ts b/contracts/ts/itemWire.ts index 61095cc69..d354bb0ed 100644 --- a/contracts/ts/itemWire.ts +++ b/contracts/ts/itemWire.ts @@ -83,6 +83,13 @@ export interface ItemWire { excluded_reason: string | null; excluded_at: string | null; review_dismissed_at: string | null; + audit_sample: boolean | null; + audit_batch_id: string | null; + audit_sampled_at: string | null; + audit_label_name: string | null; + audit_label_source: string | null; + audit_outcome: string | null; + audit_human_name: string | null; source: string; test_holdout: boolean; crop_rank_in_image: number | null; @@ -204,6 +211,13 @@ export const ITEM_WIRE_KEYS = [ 'excluded_reason', 'excluded_at', 'review_dismissed_at', + 'audit_sample', + 'audit_batch_id', + 'audit_sampled_at', + 'audit_label_name', + 'audit_label_source', + 'audit_outcome', + 'audit_human_name', 'source', 'test_holdout', 'crop_rank_in_image', diff --git a/docs/design/curation_api_contract.md b/docs/design/curation_api_contract.md index 7159b29d6..21e2dda6d 100644 --- a/docs/design/curation_api_contract.md +++ b/docs/design/curation_api_contract.md @@ -672,6 +672,7 @@ the TypeScript form. `tests/curation/test_wire_contract.py` pins the key set. | Region (item level) | `region_status`, `region_reason`, `region_rejection_reason`, `region_validated`, `region_auto_confirmed`, `region_verified`, `region_verified_at`, `region_verifier`, `region_verifier_version`, `region_visible`, `region_detector_chain`, `region_detected_at`, `region_profile`, `region_profile_revision`, `region_class_id`, `region_label_source`, `region_pairing`, `region_skip_verify` | | Region boxes | `region_boxes`, `region_count`, `region_rejected_count`, `region_max_score`, `region_set_complete`, `region_revision` | | Import and combine | `import_ids`, `dataset_split`, `imported_at`, `proposed_by_import`, `on_negative_frame`, `import_standalone_region`, `origin_project`, `origin_item_id`, `origin_image_id`, `origin_split`, `combine_conflict`, `combine_conflict_origins`, `combine_merged_origins` | +| Accuracy audit | `audit_sample`, `audit_batch_id`, `audit_sampled_at`, `audit_label_name`, `audit_label_source`, `audit_outcome`, `audit_human_name` (all null on an item never drawn into an audit) | | Text | `item_text_lines` | Endpoint-specific extra keys on top of the item: diff --git a/frontend/CHANGELOG.md b/frontend/CHANGELOG.md index 72f7bf3d9..8d5c61295 100644 --- a/frontend/CHANGELOG.md +++ b/frontend/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- The audit queue shows, on each drawn crop, the machine label and its source the crop carried when it was drawn (`audit_label_name`, `audit_label_source`, now on the item wire, #166). - **Assist…** on the cluster page (the cluster or the open sub-cluster, or the selected crops), on one crop's details panel, on one box in `/review`, and on Matching items (a filter): a dialog that posts one assist run (`POST {prefix}/assist/runs`). It offers what the VLM sees (selected, all members, a seeded representative sample with diff --git a/frontend/src/lib/api_cropMap.ts b/frontend/src/lib/api_cropMap.ts index 05addd771..28b1a3bba 100644 --- a/frontend/src/lib/api_cropMap.ts +++ b/frontend/src/lib/api_cropMap.ts @@ -76,6 +76,13 @@ export type RawCrop = { classifier_raw_confidence?: number | null; proposal_name?: string | null; detector_class_name?: string | null; + /** Accuracy audit state: drawn into a sample, the machine label it carried + * when drawn, and the verdict once a human labeled it. */ + audit_sample?: boolean | null; + audit_label_name?: string | null; + audit_label_source?: string | null; + audit_outcome?: string | null; + audit_human_name?: string | null; detector_class_id?: number | null; detector_confidence?: number | null; vlm_confidence?: string | null; @@ -178,6 +185,11 @@ export const RAW_CROP_KEYS = [ 'classifier_raw_confidence', 'proposal_name', 'detector_class_name', + 'audit_sample', + 'audit_label_name', + 'audit_label_source', + 'audit_outcome', + 'audit_human_name', 'detector_class_id', 'detector_confidence', 'vlm_confidence', @@ -269,6 +281,11 @@ export function mapRawCrop(c: RawCrop): Crop { class_id: c.class_id ?? null, class_name: c.class_name ?? null, detector_class_name: c.detector_class_name ?? null, + audit_sample: c.audit_sample ?? null, + audit_label_name: c.audit_label_name ?? null, + audit_label_source: c.audit_label_source ?? null, + audit_outcome: c.audit_outcome ?? null, + audit_human_name: c.audit_human_name ?? null, detector_class_id: c.detector_class_id ?? null, detector_confidence: c.detector_confidence ?? null, class_source: c.class_source ?? null, diff --git a/frontend/src/lib/components/audit/AuditQueueList.svelte b/frontend/src/lib/components/audit/AuditQueueList.svelte index f2e8bd4e1..2b457d01b 100644 --- a/frontend/src/lib/components/audit/AuditQueueList.svelte +++ b/frontend/src/lib/components/audit/AuditQueueList.svelte @@ -62,6 +62,14 @@ > {/if} + {#if crop.audit_sample && crop.audit_label_name} + Drawn with: {crop.audit_label_name}{#if crop.audit_label_source} + ({classSourcesStore.labelFor(crop.audit_label_source) || + crop.audit_label_source}){/if} + {/if} { class_source: 'vlm', detector_class_name: 'widget_b', detector_confidence: 0.61, + audit_sample: false, }), ), ), @@ -237,6 +238,25 @@ describe('AuditQueueList', () => { expect(href).toContain('/review?tab=all&crop_id=q0'); }); + it('shows the machine label the crop carried when it was drawn', () => { + const q = queue(1); + q.items[0] = { + ...q.items[0]!, + audit_sample: true, + audit_label_name: 'widget_c', + audit_label_source: 'vlm', + audit_outcome: null, + }; + const el = render(AuditQueueList, { queue: q, ongoto: () => {} }); + const badge = el.querySelector('[data-testid="audit-queue-drawn"]'); + expect(badge?.textContent).toContain('widget_c'); + }); + + it('shows no audit badge on a crop that was never drawn', () => { + const el = render(AuditQueueList, { queue: queue(1), ongoto: () => {} }); + expect(el.querySelector('[data-testid="audit-queue-drawn"]')).toBeNull(); + }); + it('pages by the served total and page size', () => { const ongoto = vi.fn(); const el = render(AuditQueueList, { queue: queue(30, 65), ongoto }); diff --git a/frontend/src/lib/test/makeItem.ts b/frontend/src/lib/test/makeItem.ts index 2d8f1c69d..dd286d9d2 100644 --- a/frontend/src/lib/test/makeItem.ts +++ b/frontend/src/lib/test/makeItem.ts @@ -52,6 +52,11 @@ const DEFAULT_ITEM: Required = { classifier_raw_confidence: 0.61, proposal_name: 'widget_g', detector_class_name: 'widget_h', + audit_sample: true, + audit_label_name: 'widget_i', + audit_label_source: 'vlm', + audit_outcome: 'vlm_wrong', + audit_human_name: 'widget_j', detector_class_id: 8, detector_confidence: 0.72, vlm_confidence: 'high', diff --git a/frontend/src/lib/types.ts b/frontend/src/lib/types.ts index 4db021fdb..2d294b2b4 100644 --- a/frontend/src/lib/types.ts +++ b/frontend/src/lib/types.ts @@ -50,6 +50,12 @@ export interface Crop { * (which a VLM or a human may overwrite). Null when the detector gave * none, and on items stored before it was kept. */ detector_class_name?: string | null; + /** Accuracy audit state (#119): see `RawCrop`. */ + audit_sample?: boolean | null; + audit_label_name?: string | null; + audit_label_source?: string | null; + audit_outcome?: string | null; + audit_human_name?: string | null; detector_class_id?: number | null; detector_confidence?: number | null; /** Where the class assignment came from — `human*`, the fixed VLM diff --git a/src/routers/curation/_item_models.py b/src/routers/curation/_item_models.py index 731f2f2da..92a362894 100644 --- a/src/routers/curation/_item_models.py +++ b/src/routers/curation/_item_models.py @@ -138,6 +138,16 @@ class ItemDoc(BaseModel): excluded_reason: str | None = None excluded_at: str | None = None review_dismissed_at: str | None = None + # Accuracy audit (#119): drawn into a sample (null = never drawn), the batch + # and time, the machine label it carried when drawn (name and class_source), + # and, once a human labeled it, the verdict and the human's class. + audit_sample: bool | None = None + audit_batch_id: str | None = None + audit_sampled_at: str | None = None + audit_label_name: str | None = None + audit_label_source: str | None = None + audit_outcome: Literal['agree', 'detector_wrong', 'vlm_wrong', 'both_wrong'] | None = None + audit_human_name: str | None = None # Ingest source tag. source: str = '' test_holdout: bool = False diff --git a/src/services/curation/wire.py b/src/services/curation/wire.py index f61c81529..370de0129 100644 --- a/src/services/curation/wire.py +++ b/src/services/curation/wire.py @@ -22,6 +22,7 @@ from src.clients.occ_locks import is_locked_box, is_locked_item from src.config.curation import BACKBONE_EMBEDDING_FIELD, ITEM_EMBEDDING_FIELD from src.config.region_fields import RegionFields, get_region_fields +from src.domain.audit_math import AUDIT_FIELDS from src.domain.item_text import ITEM_TEXT_LINES_FIELD, item_text_lines_to_wire from src.services.curation.assist.item_wire import assist_to_wire from src.services.curation.class_sources import ( @@ -404,6 +405,8 @@ def serialize_item( # Every OCR line on the item crop; [] when none / not yet read. 'item_text_lines': item_text_lines_to_wire(src.get(ITEM_TEXT_LINES_FIELD)), } + # The accuracy audit's state (drawn or not, the machine label it carried, the verdict). + item.update({key: src.get(key) for key in AUDIT_FIELDS}) item.update(region_to_wire(src, f)) item.update(region_boxes_to_wire(src, f, crop_id=crop_id, prefix=prefix)) # The VLM's suggestions (class answer, pack attributes), apart from the labels. diff --git a/tests/curation/test_wire_audit_fields.py b/tests/curation/test_wire_audit_fields.py new file mode 100644 index 000000000..b016d312c --- /dev/null +++ b/tests/curation/test_wire_audit_fields.py @@ -0,0 +1,33 @@ +"""The accuracy-audit state is on the item wire (#166), so a queue item can show +whether it was drawn into an audit, the machine label it carried and the verdict.""" + +from __future__ import annotations + +from src.domain.audit_math import AUDIT_FIELDS +from src.routers.curation._item_models import ItemDoc +from src.services.curation.wire import ITEM_WIRE_KEYS, serialize_item + + +def test_drawn_item_serves_its_audit_state() -> None: + src = { + 'audit_sample': True, + 'audit_batch_id': 'b1', + 'audit_sampled_at': '2026-10-01T00:00:00Z', + 'audit_label_name': 'cup', + 'audit_label_source': 'vlm', + 'audit_outcome': 'vlm_wrong', + 'audit_human_name': 'mug', + } + item = serialize_item(src, 'x', api_prefix='') + for key, value in src.items(): + assert item[key] == value + + +def test_undrawn_item_serves_nulls() -> None: + item = serialize_item({}, 'x', api_prefix='') + assert all(item[k] is None for k in AUDIT_FIELDS) + + +def test_every_stored_audit_field_is_documented() -> None: + assert set(AUDIT_FIELDS) <= ITEM_WIRE_KEYS + assert set(AUDIT_FIELDS) <= set(ItemDoc.model_fields) From 62def9c7ff80dc77c7ee35a6dd06a309d8d0901d Mon Sep 17 00:00:00 2001 From: davidamacey Date: Thu, 15 Oct 2026 18:28:53 -0400 Subject: [PATCH 02/10] fix(export): make pe-build-trt fail loudly instead of shipping FP32 (#277) The verification report was half right: build_pe_trt.sh already adds the uint8 input (PE_NORMALIZE matches the installer). What was real is the silent FP32 fallback when the FP16 bake or FP16 trtexec build failed. It now exits 4 with the cause; ALLOW_FP32=1 is the explicit opt-in. A failed uint8 step also stops with a clear message. Tests stub python and trtexec. --- CHANGELOG.md | 3 + export/README.md | 5 +- export/build_pe_trt.sh | 36 +++++-- tests/installer/test_build_pe_trt_honest.py | 113 ++++++++++++++++++++ 4 files changed, 145 insertions(+), 12 deletions(-) create mode 100644 tests/installer/test_build_pe_trt_honest.py diff --git a/CHANGELOG.md b/CHANGELOG.md index a441645c5..dabdc0605 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,9 @@ history of this codebase and was never published. This release is `[0.5.0]`. ## [Unreleased] +### Fixed +- `make pe-build-trt` (`export/build_pe_trt.sh`, #277) no longer ships an FP32 engine silently: a failed or FP32-falling-back FP16 bake, a failed FP16 build or a failed uint8 input step now exits 4 with the cause. `ALLOW_FP32=1` opts in to an FP32 engine knowingly. + ### Added - The item wire serves the accuracy audit state (#166): `audit_sample`, `audit_batch_id`, `audit_sampled_at`, `audit_label_name`, `audit_label_source`, `audit_outcome` and `audit_human_name` (null for an item never drawn), so `GET /audit/queue` items can show the machine label they were drawn with. - Cropwright **Assist…** (cluster page, one crop, one box, Matching items): a dialog to pick what the VLM sees (selected, all, a representative sample), the actions (classify, attributes, verify, segment) and the VLM and prompt pack, with the server's dry run; the cluster and sub-cluster label proposal card with the agreement, an explicit dry-run-first **Apply** and the `scope_changed` refusal; dashed "VLM suggests" attribute chips; and a **Region verification** switch (`region_verify`) on the VLM scope panel. See the guide [Selective VLM and segmenter assist](docs-site/docs/guides/selective-assist.mdx). diff --git a/export/README.md b/export/README.md index a932a2f6e..da961d51a 100644 --- a/export/README.md +++ b/export/README.md @@ -241,7 +241,10 @@ ONNX_PATH=./pytorch_models/pe_image_encoder.onnx bash export/build_pe_ort_fallba Path 2 exists because PE's attention-pooling head has, on some TensorRT releases, used ops the ONNX parser rejects. `build_pe_trt.sh` exits `3` with -an explicit pointer to the fallback when that happens. Both paths install +an explicit pointer to the fallback when that happens. The engine is FP16 with a uint8 input; +if the FP16 bake or the FP16 build fails, `build_pe_trt.sh` exits `4` instead of shipping an FP32 +engine, and `ALLOW_FP32=1` is the explicit opt-in to one (`BAKE_FP16=0` needs it too). The installer +records any FP32 fallback as `degraded` in its precision ledger. Both paths install into `models/pe_image_encoder/` and render the matching `config.pbtxt` (`tensorrt_plan` vs. `onnxruntime_onnx`); each removes the other's artifact so Triton never sees both. The committed `models/pe_image_encoder/config.pbtxt` diff --git a/export/build_pe_trt.sh b/export/build_pe_trt.sh index b2ded0fcd..ba608baff 100755 --- a/export/build_pe_trt.sh +++ b/export/build_pe_trt.sh @@ -70,14 +70,25 @@ fi # export_mobileclip_image_encoder.py call). PE was shipping FP32-only # (~2x the VRAM, slower) because this build script never called it. # -# BAKE_FP16=0 opts out entirely (build straight from the FP32 ONNX, the old -# behavior). Default on, with an FP32 fallback: if the bake step itself -# fails (module unavailable, graph rewrite error), trt_utils.py's own CLI -# already falls back and reports 'fp32'; if the bake *succeeds* but the -# resulting FP16 ONNX fails to build in trtexec below, this script retries -# once from the original FP32 ONNX before giving up. +# FP16 is required. If the bake falls back to FP32 (trt_utils.py reports +# 'fp32') or fails, or the FP16 ONNX fails to build in trtexec below, this +# script FAILS (exit 4) rather than ship an FP32 engine nobody asked for. +# ALLOW_FP32=1 is the explicit opt-in to an FP32 engine (it says so loudly); +# BAKE_FP16=0 builds straight from the FP32 ONNX and needs ALLOW_FP32=1 too. # ----------------------------------------------------------------------------- BAKE_FP16="${BAKE_FP16:-1}" +ALLOW_FP32="${ALLOW_FP32:-0}" + +# _fp32_refused WHY: stop unless the caller opted in to FP32. +_fp32_refused() { + if [ "$ALLOW_FP32" != "1" ]; then + err "$1" + err "Refusing to build an FP32 engine (about twice the GPU memory, slower)." + err "Fix the cause, or set ALLOW_FP32=1 to accept an FP32 engine knowingly." + exit 4 + fi + err "WARNING: $1 -- continuing with an FP32 engine because ALLOW_FP32=1." +} BUILD_ONNX_PATH="$ONNX_PATH" PE_PRECISION="fp32" if [ "$BAKE_FP16" = "1" ]; then @@ -92,15 +103,15 @@ if [ "$BAKE_FP16" = "1" ]; then PE_PRECISION="fp16" log " FP16 ONNX: $BUILD_ONNX_PATH" else - log " bake_fp16_onnx fell back to FP32: $(cat "$BAKE_ERR")" + _fp32_refused "FP16 bake fell back to FP32: $(cat "$BAKE_ERR")" fi else - err " trt_utils.py bake step failed unexpectedly; continuing with FP32: $(cat "$BAKE_ERR")" + _fp32_refused "FP16 bake step failed: $(cat "$BAKE_ERR")" BUILD_ONNX_PATH="$ONNX_PATH" fi rm -f "$BAKE_OUT" "$BAKE_ERR" else - log "BAKE_FP16=0 -- building FP32 (as before)" + _fp32_refused "BAKE_FP16=0 asks for an FP32 build" fi # ----------------------------------------------------------------------------- @@ -131,7 +142,10 @@ PE_NORMALIZE="--mean 0.5 0.5 0.5 --std 0.5 0.5 0.5" _stage_and_resolve_args() { local src_onnx="$1" # shellcheck disable=SC2086 - src_onnx="$("$PYTHON" "$SCRIPT_DIR/uint8_input.py" "$src_onnx" "${src_onnx%.onnx}.u8.onnx" $PE_NORMALIZE)" + src_onnx="$("$PYTHON" "$SCRIPT_DIR/uint8_input.py" "$src_onnx" "${src_onnx%.onnx}.u8.onnx" $PE_NORMALIZE)" || { + err "could not add the uint8 input to the engine ONNX; config.pbtxt declares uint8, so Triton would refuse the engine" + exit 4 + } if [ -n "${TRTEXEC:-}" ] || command -v trtexec >/dev/null 2>&1; then ONNX_ARG="$src_onnx" PLAN_ARG="$PLAN_TMP" @@ -168,7 +182,7 @@ if ! "${TRTEXEC_CMD[@]}" \ --skipInference ; then if [ "$PE_PRECISION" = "fp16" ]; then err "" - err "trtexec failed to build the FP16 engine. Falling back to FP32 (original ONNX)..." + _fp32_refused "trtexec failed to build the FP16 engine" PE_PRECISION="fp32" _stage_and_resolve_args "$ONNX_PATH" if ! "${TRTEXEC_CMD[@]}" \ diff --git a/tests/installer/test_build_pe_trt_honest.py b/tests/installer/test_build_pe_trt_honest.py new file mode 100644 index 000000000..a861e42cb --- /dev/null +++ b/tests/installer/test_build_pe_trt_honest.py @@ -0,0 +1,113 @@ +"""``export/build_pe_trt.sh`` (``make pe-build-trt``) never ships an FP32 engine silently (#277). + +The FP16 bake or the FP16 trtexec build failing is a hard failure unless the caller +says ``ALLOW_FP32=1``; the uint8 input prefix is always applied; stubs stand in for +python and trtexec so no GPU is needed. +""" + +from __future__ import annotations + +import os +import subprocess +from typing import TYPE_CHECKING + +import pytest +from installer_harness import REPO_ROOT + + +if TYPE_CHECKING: + from pathlib import Path + +SCRIPT = REPO_ROOT / 'export' / 'build_pe_trt.sh' + +PY_STUB = """#!/bin/bash +# stands in for the venv python: dispatches on the script basename +name="$(basename "$1")" +echo "py $*" >> "$STUB_LOG" +case "$name" in + trt_utils.py) + if [ "${STUB_BAKE:-fp16}" = "crash" ]; then echo "boom" >&2; exit 1; fi + if [ "${STUB_BAKE:-fp16}" = "fp16" ]; then echo FP16MARK > "$3"; else cp "$2" "$3"; fi; echo "${STUB_BAKE:-fp16}" >&2; echo "$3" ;; + uint8_input.py) + if [ "${STUB_U8:-ok}" = "fail" ]; then echo "no uint8" >&2; exit 1; fi + cp "$2" "$3"; echo "$3" ;; + *) : ;; +esac +""" + +TRTEXEC_STUB = """#!/bin/bash +echo "trtexec $*" >> "$STUB_LOG" +onnx="" +for a in "$@"; do case "$a" in --onnx=*) onnx="${a#--onnx=}";; --saveEngine=*) plan="${a#--saveEngine=}";; esac; done +if [ "${STUB_TRT:-ok}" = "fp16_fails" ] && grep -q FP16MARK "$onnx"; then exit 1; fi +echo plan > "$plan" +""" + + +def _run(tmp_path: Path, **env: str) -> tuple[subprocess.CompletedProcess[str], str]: + onnx = tmp_path / 'pe_image_encoder.onnx' + onnx.write_text('FP32') + py = tmp_path / 'py' + py.write_text(PY_STUB) + trt = tmp_path / 'trtexec' + trt.write_text(TRTEXEC_STUB) + py.chmod(0o755) + trt.chmod(0o755) + log = tmp_path / 'calls.log' + log.touch() + full = { + **os.environ, + 'ONNX_PATH': str(onnx), + 'MODELS_DIRS': str(tmp_path / 'models'), + 'PYTHON': str(py), + 'TRTEXEC': str(trt), + 'STUB_LOG': str(log), + **env, + } + done = subprocess.run( + ['bash', str(SCRIPT)], env=full, capture_output=True, text=True, check=False + ) + return done, log.read_text() + + +def test_clean_fp16_build_installs_the_plan_with_uint8_input(tmp_path: Path) -> None: + done, calls = _run(tmp_path) + assert done.returncode == 0, done.stderr + assert 'uint8_input.py' in calls + assert (tmp_path / 'models/pe_image_encoder/1/model.plan').exists() + assert 'fp16' in done.stdout + + +def test_fp16_trtexec_failure_fails_loudly_and_installs_nothing(tmp_path: Path) -> None: + done, calls = _run(tmp_path, STUB_TRT='fp16_fails') + assert done.returncode != 0 + assert 'FP16' in done.stderr + assert 'ALLOW_FP32=1' in done.stderr + assert ( + calls.count('\ntrtexec ') + calls.startswith('trtexec ') == 1 + ) # no silent second, FP32 build + assert not (tmp_path / 'models/pe_image_encoder/1/model.plan').exists() + + +def test_bake_falling_back_to_fp32_fails_loudly(tmp_path: Path) -> None: + done, _ = _run(tmp_path, STUB_BAKE='fp32') + assert done.returncode != 0 + assert 'ALLOW_FP32=1' in done.stderr + assert not (tmp_path / 'models/pe_image_encoder/1/model.plan').exists() + + +def test_uint8_prefix_failure_fails_loudly(tmp_path: Path) -> None: + done, calls = _run(tmp_path, STUB_U8='fail') + assert done.returncode != 0 + assert 'uint8' in done.stderr + assert 'trtexec ' not in calls + + +@pytest.mark.parametrize('extra', [{'STUB_TRT': 'fp16_fails'}, {'STUB_BAKE': 'fp32'}]) +def test_explicit_opt_in_still_builds_fp32_and_says_so( + tmp_path: Path, extra: dict[str, str] +) -> None: + done, _ = _run(tmp_path, ALLOW_FP32='1', **extra) + assert done.returncode == 0, done.stderr + assert (tmp_path / 'models/pe_image_encoder/1/model.plan').exists() + assert 'FP32' in done.stderr From 682cdcf0e6ccf16db97691ce063d70af86d095f9 Mon Sep 17 00:00:00 2001 From: davidamacey Date: Thu, 15 Oct 2026 18:32:07 -0400 Subject: [PATCH 03/10] fix(projects): clear the detection worker's busy state when its batch lands (#249) Diagnosis: the busy inventory itself is correct, but the detection worker's liveness file (inflight count) was only rewritten every 15 s, so a DELETE right after an ingest finished read the last non-zero count and answered 409 project_busy for up to a full interval. liveness_loop now polls each second and rewrites a project as soon as its count changes. The 409 message also names each running job and hints at a retry (it already listed them in jobs). --- CHANGELOG.md | 1 + docs-site/docs/guides/projects.mdx | 2 +- .../docs/operations/health-and-stalls.mdx | 3 +- scripts/curation/worker/fairness.py | 20 +++++++++++-- src/services/projects/delete.py | 3 +- src/services/projects/lifecycle.py | 3 +- src/services/projects/lifecycle_checks.py | 14 +++++++++ tests/projects/test_lifecycle_busy_wiring.py | 25 ++++++++++++++++ .../test_worker_runtime_per_project.py | 29 +++++++++++++++++++ 9 files changed, 93 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dabdc0605..58a968502 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ history of this codebase and was never published. This release is `[0.5.0]`. ## [Unreleased] ### Fixed +- `DELETE /projects/{p}` (and archive) answering `409 project_busy` right after an ingest finished (#249): the detection worker refreshed its in-flight count only every 15 s, so the last batch read as busy for up to that long after it had landed. The worker now rewrites a project's liveness within a second of its count changing, and the `project_busy` message names each running job and says when a retry is the answer. - `make pe-build-trt` (`export/build_pe_trt.sh`, #277) no longer ships an FP32 engine silently: a failed or FP32-falling-back FP16 bake, a failed FP16 build or a failed uint8 input step now exits 4 with the cause. `ALLOW_FP32=1` opts in to an FP32 engine knowingly. ### Added diff --git a/docs-site/docs/guides/projects.mdx b/docs-site/docs/guides/projects.mdx index 3241aa511..0a2712b24 100644 --- a/docs-site/docs/guides/projects.mdx +++ b/docs-site/docs/guides/projects.mdx @@ -183,7 +183,7 @@ Every error body has the shape | 409 | `slug_taken`, `slug_retired` | Create with a used or retired slug | | 409 | `revision_conflict` | `expected_revision` is stale; `current_revision` is returned | | 409 | `invalid_transition` | For example archiving a project that is not `active` | -| 409 | `project_busy` | Running jobs; `jobs` lists them | +| 409 | `project_busy` | Running jobs; `jobs` lists them and the message names them. A detection worker finishing its last batch clears within about 15 seconds, so retry | | 409 | `last_active_project` | Archive or delete would leave no active project | | 409 | `project_protected` | Delete of the `default` project | | 409 | `in_use` | Delete while another project depends on a shared model | diff --git a/docs-site/docs/operations/health-and-stalls.mdx b/docs-site/docs/operations/health-and-stalls.mdx index dea78d2f0..d5e308888 100644 --- a/docs-site/docs/operations/health-and-stalls.mdx +++ b/docs-site/docs/operations/health-and-stalls.mdx @@ -50,7 +50,8 @@ writes), and a `building`, `deleting` or `failed` project refuses requests; `GET /curation/projects/{project}` serves the `status` and a typed `error`. A live job (import, combine, training, delete drain) makes a project busy, and lifecycle operations on it answer `409 project_busy` with the jobs that -hold it. +hold it. The message names each job; a `detection_worker` entry right after an +ingest is the worker finishing its last batch, so retry in a few seconds. ## Why isn't the region queue shrinking? diff --git a/scripts/curation/worker/fairness.py b/scripts/curation/worker/fairness.py index 0e155ce08..e69c8e3b3 100644 --- a/scripts/curation/worker/fairness.py +++ b/scripts/curation/worker/fairness.py @@ -194,6 +194,7 @@ def write_liveness( LIVENESS_INTERVAL_S = 15.0 +LIVENESS_POLL_S = 1.0 async def liveness_loop( @@ -202,24 +203,37 @@ async def liveness_loop( stop_event: asyncio.Event, *, interval_s: float = LIVENESS_INTERVAL_S, + poll_s: float = LIVENESS_POLL_S, ) -> None: """Refresh every active project's liveness doc on a timer. The producer only writes after a fetch that found new items, so a worker busy on a long batch (queue full, everything pending already in flight, paused) would otherwise go stale and read as idle to - ``busy.py``.""" + ``busy.py``. + + A project whose in-flight count changed is rewritten within ``poll_s``, + not on the next ``interval_s`` tick: otherwise a delete right after the + last item landed read a stale non-zero count and answered 409 + ``project_busy`` for up to a full interval (#249).""" + last: dict[str, tuple[int, float]] = {} while not stop_event.is_set(): counts = inflight_counts() + now = time.monotonic() for record in projects(): + inflight = counts.get(record.slug, 0) + prev = last.get(record.slug) + if prev is not None and prev[0] == inflight and now - prev[1] < interval_s: + continue with contextlib.suppress(OSError): write_liveness( record, - inflight=counts.get(record.slug, 0), + inflight=inflight, applied=True, paused=is_project_paused(record), ) + last[record.slug] = (inflight, now) with contextlib.suppress(TimeoutError): - await asyncio.wait_for(stop_event.wait(), timeout=interval_s) + await asyncio.wait_for(stop_event.wait(), timeout=min(poll_s, interval_s)) def read_liveness(record: ProjectRecord, *, host: str | None = None) -> dict | None: diff --git a/src/services/projects/delete.py b/src/services/projects/delete.py index 2a115859c..28a428a5e 100644 --- a/src/services/projects/delete.py +++ b/src/services/projects/delete.py @@ -243,6 +243,7 @@ async def delete_project( from src.services.projects.lifecycle_checks import ( _last_active_check, _require_transition, + busy_message, running_jobs, ) @@ -311,7 +312,7 @@ async def delete_project( raise rejection( 409, 'project_busy', - f"'{slug}' has {len(jobs)} running job(s)", + busy_message(slug, jobs), project=slug, jobs=[j.to_wire() for j in jobs], ) diff --git a/src/services/projects/lifecycle.py b/src/services/projects/lifecycle.py index 07d439fa1..16431bb93 100644 --- a/src/services/projects/lifecycle.py +++ b/src/services/projects/lifecycle.py @@ -31,6 +31,7 @@ _last_active_check, _require_revision, _require_transition, + busy_message, running_jobs, ) from src.services.projects.registry import ( @@ -501,7 +502,7 @@ async def archive_project(client: Any, *, slug: str, expected_revision: int) -> raise rejection( 409, 'project_busy', - f"'{slug}' has {len(jobs)} running job(s)", + busy_message(slug, jobs), project=slug, jobs=[j.to_wire() for j in jobs], ) diff --git a/src/services/projects/lifecycle_checks.py b/src/services/projects/lifecycle_checks.py index 93b05aaef..440e7c642 100644 --- a/src/services/projects/lifecycle_checks.py +++ b/src/services/projects/lifecycle_checks.py @@ -46,6 +46,20 @@ def to_wire(self) -> dict[str, Any]: } +def busy_message(slug: str, jobs: list[JobRef]) -> str: + """The ``project_busy`` message: what is running, by name, and when a retry + is the right answer (a detection worker's in-flight batch drains on its + own; its liveness refreshes within seconds).""" + named = ', '.join(f'{j.kind_label} {j.id}' for j in jobs[:5]) + more = f' and {len(jobs) - 5} more' if len(jobs) > 5 else '' + hint = ( + ' A detection worker finishing its last batch clears within about 15 seconds: retry.' + if any(j.kind == 'detection_worker' for j in jobs) + else ' Cancel or wait for them, then retry.' + ) + return f"'{slug}' has {len(jobs)} running job(s): {named}{more}.{hint}" + + async def running_jobs(record: ProjectRecord) -> list[JobRef]: """The project's real running-job inventory, via ``src.services.projects.busy.running_jobs`` -- the single source of diff --git a/tests/projects/test_lifecycle_busy_wiring.py b/tests/projects/test_lifecycle_busy_wiring.py index d18b8a0e0..6e159ebc6 100644 --- a/tests/projects/test_lifecycle_busy_wiring.py +++ b/tests/projects/test_lifecycle_busy_wiring.py @@ -160,3 +160,28 @@ async def _run() -> None: ] asyncio.run(_run()) + + +def test_busy_message_names_the_jobs_and_hints_at_the_worker_settle(monkeypatch) -> None: + """#249: a 409 right after an ingest says which job is busy, and that a + detection worker's in-flight batch clears on its own within seconds.""" + from src.services.projects import busy + + async def _run() -> None: + client = FakeLifecycleOpenSearch() + set_project_registry(ProjectRegistry(lambda: client)) + await seed_default_project(client) + record, _ = await lifecycle.create_project(client, slug='cars', display_name='Cars') + monkeypatch.setattr( + busy, + 'running_jobs', + lambda _rec: [busy.JobRef(kind='detection_worker', job_id='host-a')], + ) + with pytest.raises(RequestRejectedError) as exc_info: + await lifecycle.archive_project(client, slug='cars', expected_revision=record.revision) + message = exc_info.value.detail['message'] + assert 'detection_worker' in message or 'Detection worker' in message + assert 'host-a' in message + assert 'retry' in message + + asyncio.run(_run()) diff --git a/tests/projects/test_worker_runtime_per_project.py b/tests/projects/test_worker_runtime_per_project.py index 79860166c..6f3ba9547 100644 --- a/tests/projects/test_worker_runtime_per_project.py +++ b/tests/projects/test_worker_runtime_per_project.py @@ -157,3 +157,32 @@ def _recording_write(record: ProjectRecord, **kwargs: Any) -> None: doc = read_liveness(alpha) assert doc is not None assert doc['inflight'] == 3 + + +@pytest.mark.asyncio +async def test_liveness_drops_to_idle_promptly_not_on_the_next_interval( + tmp_path: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + """#249: when a project's last in-flight item lands, the busy check must see + inflight 0 within a poll, not up to a full liveness interval later (a DELETE + right after an ingest finished answered 409 on the stale count).""" + import asyncio + + from scripts.curation.worker import fairness + + alpha = _record(tmp_path, 'alpha') + monkeypatch.setattr(fairness, 'is_project_paused', lambda _p: False) + counts = {'alpha': 3} + stop = asyncio.Event() + task = asyncio.create_task( + fairness.liveness_loop(lambda: [alpha], lambda: counts, stop, interval_s=30.0, poll_s=0.01) + ) + await asyncio.sleep(0.05) + assert read_liveness(alpha)['inflight'] == 3 # type: ignore[index] + counts['alpha'] = 0 + await asyncio.sleep(0.1) + doc = read_liveness(alpha) + stop.set() + await task + assert doc is not None + assert doc['inflight'] == 0 From bdb212e0d2b05aff39bc517bf37e3f129c237421 Mon Sep 17 00:00:00 2001 From: davidamacey Date: Thu, 15 Oct 2026 18:36:22 -0400 Subject: [PATCH 04/10] fix(ingest): start dispatcher runs on the loop and never leave an orphan run run_tick ran run_project_tick via to_thread, so spawn and the poll called asyncio.create_task from a worker thread ('no running event loop') and no poll, cron or webhook run ever started. Because claim_run had already written the run, every tick re-claimed the interrupted run (refreshing its heartbeat) and it stayed live: later /run calls and DELETE of the project answered 409. The tick now runs on the loop, spawn marks a claimed run failed when it cannot start, and cancel finalizes an orphan with no task. --- CHANGELOG.md | 1 + .../curation/bulk_ingest/dispatcher.py | 4 +- src/services/curation/bulk_ingest/runner.py | 42 +++++- .../curation/test_bulk_ingest_orphan_runs.py | 123 ++++++++++++++++++ 4 files changed, 166 insertions(+), 4 deletions(-) create mode 100644 tests/curation/test_bulk_ingest_orphan_runs.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 58a968502..046753c89 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ history of this codebase and was never published. This release is `[0.5.0]`. ## [Unreleased] ### Fixed +- Dispatcher-started bulk-ingest runs (poll, cron, webhook) never started: the tick ran in a worker thread and both the run and the poll called `asyncio.create_task` there (`ingest_dispatch_failed`, `no running event loop`). The tick now runs on the loop. A run whose spawn fails is marked `failed` instead of staying `queued` (which kept every later start, and the project's delete, answering 409), and `POST /ingest/bulk/{run_id}/cancel` finalizes an orphan (a run `queued` past 10 s with no task, or `interrupted` after a restart) as `cancelled`. - `DELETE /projects/{p}` (and archive) answering `409 project_busy` right after an ingest finished (#249): the detection worker refreshed its in-flight count only every 15 s, so the last batch read as busy for up to that long after it had landed. The worker now rewrites a project's liveness within a second of its count changing, and the `project_busy` message names each running job and says when a retry is the answer. - `make pe-build-trt` (`export/build_pe_trt.sh`, #277) no longer ships an FP32 engine silently: a failed or FP32-falling-back FP16 bake, a failed FP16 build or a failed uint8 input step now exits 4 with the cause. `ALLOW_FP32=1` opts in to an FP32 engine knowingly. diff --git a/src/services/curation/bulk_ingest/dispatcher.py b/src/services/curation/bulk_ingest/dispatcher.py index 1a207cc26..a250e7d56 100644 --- a/src/services/curation/bulk_ingest/dispatcher.py +++ b/src/services/curation/bulk_ingest/dispatcher.py @@ -222,7 +222,9 @@ async def run_tick() -> None: for slug in for_each_project(): try: - await asyncio.to_thread(run_project_tick) + # On the loop, not a worker thread: a started run and a poll are asyncio + # tasks. The tick is a few small file reads and a non-blocking lock. + run_project_tick() except Exception as exc: # one project's failure must not stop the others' ticks logger.warning('ingest_dispatch_failed', project=slug, error=str(exc)) diff --git a/src/services/curation/bulk_ingest/runner.py b/src/services/curation/bulk_ingest/runner.py index e872ddec1..22ffe1d2f 100644 --- a/src/services/curation/bulk_ingest/runner.py +++ b/src/services/curation/bulk_ingest/runner.py @@ -31,6 +31,7 @@ import time from collections import deque from dataclasses import dataclass +from datetime import UTC, datetime from typing import TYPE_CHECKING, Any, Protocol from src.config.project_context import ensure_marked_dir @@ -152,9 +153,36 @@ def claim_run(source: BulkSource, options: BulkOptions) -> tuple[BulkStore, bool return store, resumed +# A claimed run is spawned in the same call, so a run still ``queued`` this long +# after its claim was never started (its spawn failed, or the process died between). +QUEUED_ORPHAN_GRACE_S = 10.0 + + +def _is_orphan(store: BulkStore) -> bool: + """A run no task will ever finalize: ``queued`` past the grace with no task in + this process (a started run flips to ``running`` at once, and another worker's + ``queued`` run is only ever milliseconds old), or ``interrupted`` (its heartbeat + went stale, e.g. across an API restart).""" + state = store.job.read() + status = state.get('status') + if status == 'interrupted': + return True + if status != 'queued' or store.run_id in _tasks: + return False + try: + age = (datetime.now(UTC) - datetime.fromisoformat(state['updated_at'])).total_seconds() + except (KeyError, TypeError, ValueError): + return True # an unreadable timestamp cannot vouch for a live claim + return age > QUEUED_ORPHAN_GRACE_S + + def cancel_run(store: BulkStore) -> None: - """Ask a live run to stop after its in-flight batches. A run that is not live - keeps its status (the flag is cleared by the next claim).""" + """Ask a live run to stop after its in-flight batches, or, for an orphan no task + will finalize, mark it ``cancelled`` now. A run that is finished keeps its status + (the flag is cleared by the next claim).""" + if _is_orphan(store): + store.job.update(status='cancelled', finished_at=now_iso(), error=None) + return if store.job.is_live(ACTIVE_STATUSES): store.job.request_cancel() @@ -469,6 +497,14 @@ async def runner() -> None: if aclose is not None: await aclose() - task = asyncio.create_task(runner()) + coroutine = runner() + try: + task = asyncio.create_task(coroutine) + except RuntimeError as exc: # no running loop: the claimed run must not stay queued + coroutine.close() + store.job.update( + status='failed', error=f'could not start: {exc}'[:300], finished_at=now_iso() + ) + raise _tasks[store.run_id] = task task.add_done_callback(lambda _t: _tasks.pop(store.run_id, None)) diff --git a/tests/curation/test_bulk_ingest_orphan_runs.py b/tests/curation/test_bulk_ingest_orphan_runs.py new file mode 100644 index 000000000..9661ab42b --- /dev/null +++ b/tests/curation/test_bulk_ingest_orphan_runs.py @@ -0,0 +1,123 @@ +"""A dispatcher-started run actually starts, a failed start leaves no orphan, and cancel +finalizes an orphan (live S3 check): ``run_tick`` used to call ``asyncio.create_task`` from +the ``to_thread`` worker (``no running event loop``), after ``claim_run`` had already +written a ``queued`` run that then blocked every later start and the project's delete.""" + +from __future__ import annotations + +import asyncio +import contextvars +import threading +from typing import TYPE_CHECKING, Any + +import pytest + +from src.services.curation.bulk_ingest import dispatcher, runner, source_runs +from src.services.curation.bulk_ingest.object_queue import QueueEntry +from src.services.curation.bulk_ingest.source import BulkSource +from src.services.curation.bulk_ingest.sources import IngestSource, save_source, source_queue +from src.services.curation.bulk_ingest.store import active_run, list_stores + + +if TYPE_CHECKING: + from pathlib import Path + + +class _Sender: + async def post_batch(self, paths: list[str], *, source: str, run_id: str) -> dict[str, Any]: # noqa: ARG002 + return { + 'summary': {}, + 'results': [{'status': 'success', 'image_path': p, 'image_id': p} for p in paths], + } + + async def stage_seconds(self) -> dict[str, float]: + return {} + + +@pytest.fixture(autouse=True) +def _bulk_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv('OP_BULK_INGEST_DIR', str(tmp_path / 'bulk')) + monkeypatch.setattr(source_runs, 'url_sender', lambda *_a, **_k: _Sender()) + monkeypatch.setattr(runner, '_tasks', {}) + + +def _webhook_source() -> IngestSource: + src = IngestSource( + source_id='src_orphans', name='t', backend='s3', bucket='b', webhook_enabled=True + ) + save_source(src) + source_queue(src.source_id).enqueue([QueueEntry('k.jpg', 'e')]) + return src + + +def test_run_tick_starts_the_run_the_way_production_calls_it( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _webhook_source() + monkeypatch.setattr('src.services.projects.bootstrap.for_each_project', lambda: iter(['p'])) + failures: list[Any] = [] + monkeypatch.setattr( + dispatcher.logger, 'warning', lambda event, **kw: failures.append((event, kw)) + ) + + async def scenario() -> None: + await dispatcher.run_tick() + await asyncio.gather(*runner._tasks.values()) + + asyncio.run(scenario()) + assert failures == [] + runs = list_stores() + assert len(runs) == 1 + assert runs[0].job.read()['status'] in {'completed', 'completed_with_errors'} + + +def test_a_spawn_that_cannot_start_leaves_a_failed_run_not_a_live_orphan() -> None: + src = _webhook_source() + errors: list[BaseException] = [] + + def from_a_thread() -> None: # no running loop in this thread: spawn cannot start + try: + source_runs.start_source_run( + src, base_url='http://x', api_prefix='', project='p', scan=False + ) + except RuntimeError as exc: + errors.append(exc) + + worker = threading.Thread(target=contextvars.copy_context().run, args=(from_a_thread,)) + worker.start() + worker.join() + assert errors + assert active_run() is None + (store,) = list_stores() + assert store.job.read()['status'] == 'failed' + # The next claim is not refused as busy. + runner.claim_run(BulkSource('queue', src.source_id, (), False), runner.BulkOptions('t', 4, 1)) + + +def test_cancel_finalizes_an_orphan_with_no_task_even_after_a_restart() -> None: + spec = BulkSource('folder', '/x') + store, _ = runner.claim_run(spec, runner.BulkOptions('t', 4, 1)) + store.job.update(updated_at='2020-01-01T00:00:00+00:00') # claimed long ago, never spawned + assert active_run() is not None # its heartbeat is fresh, so it blocks a start + with pytest.raises(runner.BulkBusyError): + runner.claim_run(spec, runner.BulkOptions('t', 4, 1)) + + runner.cancel_run(store) + + state = store.job.read() + assert state['status'] == 'cancelled' + assert state['finished_at'] + assert active_run() is None + + +def test_cancel_still_signals_a_run_that_has_a_task_here() -> None: + spec = BulkSource('folder', '/x') + store, _ = runner.claim_run(spec, runner.BulkOptions('t', 4, 1)) + + async def scenario() -> None: + runner._tasks[store.run_id] = asyncio.create_task(asyncio.sleep(0)) + runner.cancel_run(store) + + asyncio.run(scenario()) + assert store.job.cancel_requested() + assert store.job.read()['status'] == 'queued' # its own task finalizes it From e5a987abd2ae3a3102e4c8b603f8280a94d35843 Mon Sep 17 00:00:00 2001 From: davidamacey Date: Thu, 15 Oct 2026 18:44:18 -0400 Subject: [PATCH 05/10] feat(assist): serve sub-cluster ids on the cluster card and add a min_agreement control GET /clusters cards carry subcluster_ids (sorted, capped at 200), and the cluster page's sub-cluster tabs use them instead of the ids seen on the loaded page. The assist dialog gets an Agreement needed field for a cluster classify run: empty uses the server default (echoed on the dry run), range pinned to the served schema by a contract test. --- CHANGELOG.md | 1 + contracts/openapi/curation.json | 2 +- docs-site/docs/guides/selective-assist.mdx | 2 + docs/design/curation_api_contract.md | 2 +- docs/design/selectable_assist_plan.md | 4 +- frontend/CHANGELOG.md | 1 + .../e2e/stubbed/test_assist_run_dialog.py | 1 + frontend/src/lib/api.clusters.test.ts | 40 ++++++++++++ frontend/src/lib/api_clusters.ts | 3 + .../assistRun/assistRunController.svelte.ts | 4 ++ .../lib/assistRun/assistRunController.test.ts | 11 ++++ .../clusterDetailController.svelte.ts | 17 +----- .../src/lib/clusters/slotInventory.svelte.ts | 1 + .../assistRun/AssistRunDialog.svelte | 26 +++++++- .../assistRun/AssistRunDialog.test.ts | 21 ++++++- .../SlotGallery.falsePositiveKind.test.ts | 1 + .../slots/SlotGallery.unclustered.test.ts | 1 + .../src/lib/contract/assistContract.test.ts | 10 ++- frontend/src/lib/types.ts | 3 + frontend/src/lib/types_assist.ts | 4 ++ src/routers/curation/clusters.py | 18 +++++- .../test_cluster_card_subcluster_ids.py | 61 +++++++++++++++++++ 22 files changed, 212 insertions(+), 22 deletions(-) create mode 100644 tests/curation/test_cluster_card_subcluster_ids.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 046753c89..66c7e5191 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,7 @@ history of this codebase and was never published. This release is `[0.5.0]`. - `make pe-build-trt` (`export/build_pe_trt.sh`, #277) no longer ships an FP32 engine silently: a failed or FP32-falling-back FP16 bake, a failed FP16 build or a failed uint8 input step now exits 4 with the cause. `ALLOW_FP32=1` opts in to an FP32 engine knowingly. ### Added +- `GET /clusters` cards serve `subcluster_ids` (the sorted sub-cluster ids, at most 200; `n_subclusters` stays the exact count). - The item wire serves the accuracy audit state (#166): `audit_sample`, `audit_batch_id`, `audit_sampled_at`, `audit_label_name`, `audit_label_source`, `audit_outcome` and `audit_human_name` (null for an item never drawn), so `GET /audit/queue` items can show the machine label they were drawn with. - Cropwright **Assist…** (cluster page, one crop, one box, Matching items): a dialog to pick what the VLM sees (selected, all, a representative sample), the actions (classify, attributes, verify, segment) and the VLM and prompt pack, with the server's dry run; the cluster and sub-cluster label proposal card with the agreement, an explicit dry-run-first **Apply** and the `scope_changed` refusal; dashed "VLM suggests" attribute chips; and a **Region verification** switch (`region_verify`) on the VLM scope panel. See the guide [Selective VLM and segmenter assist](docs-site/docs/guides/selective-assist.mdx). diff --git a/contracts/openapi/curation.json b/contracts/openapi/curation.json index b30afab91..3c91c28da 100644 --- a/contracts/openapi/curation.json +++ b/contracts/openapi/curation.json @@ -35027,7 +35027,7 @@ }, "/curation/projects/{project}/clusters": { "get": { - "description": "Authoritative per-cluster card payload for the labeler.\n\nSingle OpenSearch aggregation that returns, for every cluster_id:\n\n* ``size`` (total members), ``validated_count`` (class_validated=true),\n* ``dominant_class_{id,name,count}`` and ``label_purity``\n (largest-class share among labelled members) with\n ``labelled_share``. A candidate cluster only gets a\n ``dominant_class_name`` when a unique top class has at least\n ``CANDIDATE_DOMINANT_MIN_COUNT`` members and\n ``CANDIDATE_DOMINANT_MIN_SHARE`` of the labelled ones;\n ``dominant_count``/``labelled_count``/``label_purity`` are always\n reported,\n* ``purity`` / ``purity_n`` / ``purity_basis`` / ``purity_tier``\n : the share of the ``purity_n`` members the cluster-geometry\n pass measured for this cluster whose nearest cluster centroid is\n this cluster's own — independent of the labels that placed them, so\n a class cluster is no longer 1.0 by construction; ``null`` when no\n member has been measured,\n* ``is_unlabeled`` (true when no class_name has any signal at all),\n* ``cluster_kind`` (``class`` | ``candidate`` | ``unassigned``),\n* ``n_subclusters`` (distinct cluster_subid values), and\n* ``representatives`` (top ``per_cluster`` crops sorted by\n cluster_distance asc).\n\nThis endpoint exists so the frontend can stop computing dominant\nclass / purity from a 4-crop sample (it was getting wrong answers\non candidate clusters where all reps were null-class). The\naggregation is cheap because ``cluster_id`` is a low-cardinality\ninteger field; ``class_name``/``class_source``/``cluster_subid`` are\nall mapped ``keyword`` directly on the live index (no ``.keyword``\nsubfield).", + "description": "Authoritative per-cluster card payload for the labeler.\n\nSingle OpenSearch aggregation that returns, for every cluster_id:\n\n* ``size`` (total members), ``validated_count`` (class_validated=true),\n* ``dominant_class_{id,name,count}`` and ``label_purity``\n (largest-class share among labelled members) with\n ``labelled_share``. A candidate cluster only gets a\n ``dominant_class_name`` when a unique top class has at least\n ``CANDIDATE_DOMINANT_MIN_COUNT`` members and\n ``CANDIDATE_DOMINANT_MIN_SHARE`` of the labelled ones;\n ``dominant_count``/``labelled_count``/``label_purity`` are always\n reported,\n* ``purity`` / ``purity_n`` / ``purity_basis`` / ``purity_tier``\n : the share of the ``purity_n`` members the cluster-geometry\n pass measured for this cluster whose nearest cluster centroid is\n this cluster's own — independent of the labels that placed them, so\n a class cluster is no longer 1.0 by construction; ``null`` when no\n member has been measured,\n* ``is_unlabeled`` (true when no class_name has any signal at all),\n* ``cluster_kind`` (``class`` | ``candidate`` | ``unassigned``),\n* ``n_subclusters`` (distinct cluster_subid values), ``subcluster_ids``\n (those ids, at most ``MAX_SUBCLUSTER_IDS``), and\n* ``representatives`` (top ``per_cluster`` crops sorted by\n cluster_distance asc).\n\nThis endpoint exists so the frontend can stop computing dominant\nclass / purity from a 4-crop sample (it was getting wrong answers\non candidate clusters where all reps were null-class). The\naggregation is cheap because ``cluster_id`` is a low-cardinality\ninteger field; ``class_name``/``class_source``/``cluster_subid`` are\nall mapped ``keyword`` directly on the live index (no ``.keyword``\nsubfield).", "operationId": "list_clusters_curation_projects__project__clusters_get", "parameters": [ { diff --git a/docs-site/docs/guides/selective-assist.mdx b/docs-site/docs/guides/selective-assist.mdx index ab07e70dd..0cac0a4b8 100644 --- a/docs-site/docs/guides/selective-assist.mdx +++ b/docs-site/docs/guides/selective-assist.mdx @@ -121,6 +121,8 @@ of answers for it), the vote distribution, new-class names the VLM proposed when nothing fit, and a summary of the attributes. The proposal names a class only when at least 3 answers exist and the agreement reaches `min_agreement` (default 0.6); otherwise the distribution is shown and no class is proposed. +The assist dialog has an **Agreement needed** field for a cluster `classify` run: empty uses the +server's default (shown in the field), and it accepts 0.5 to 1.0. Applying it is a separate, explicit call, with a dry-run count first: diff --git a/docs/design/curation_api_contract.md b/docs/design/curation_api_contract.md index 21e2dda6d..15cc65e84 100644 --- a/docs/design/curation_api_contract.md +++ b/docs/design/curation_api_contract.md @@ -1394,7 +1394,7 @@ Representatives are paged. Every card is returned with `size`, `purity` and the rest, but `representatives` is filled only for cards in the `[offset, offset + limit)` window of the kind-filtered, member-count-descending list (`limit` default 50, max 500). Other cards carry `representatives: []`. -`per_cluster=0` skips representatives. `GET /clusters/representatives` +Each card carries `n_subclusters` (the exact count) and `subcluster_ids` (the sorted ids, at most 200), which the Cropwright sub-cluster tabs use. `per_cluster=0` skips representatives. `GET /clusters/representatives` pages the same way by `offset` and `max_clusters`. Clustering runs inside the auto-label pipeline. Setting diff --git a/docs/design/selectable_assist_plan.md b/docs/design/selectable_assist_plan.md index 3a0192299..e7d0e8142 100644 --- a/docs/design/selectable_assist_plan.md +++ b/docs/design/selectable_assist_plan.md @@ -753,7 +753,7 @@ pins for the deleted `/vlm/label_cluster` and `/vlm/verify_regions` routes; rege Recorded after wave C (PR 8, the frontend). -- **The sub-cluster tab list is not served.** `GET /clusters` serves only `n_subclusters`, so the tabs are +- **The sub-cluster tab list is not served** (fixed afterwards: `GET /clusters` serves `subcluster_ids` per card and the tabs use it; the text below is the gap as it stood). `GET /clusters` serves only `n_subclusters`, so the tabs are the sub-cluster ids seen while the whole cluster is shown (replaced on the first page, added to on later ones). A tab loads from `GET /crops?cluster_subid=`; a sub-cluster whose members are all beyond the loaded pages has no tab until the user scrolls the "all" view. A served list of sub-cluster ids would fix it. @@ -762,7 +762,7 @@ Recorded after wave C (PR 8, the frontend). - **Single-box assist is on the `/review` box row**, not inside `SlotBboxEditor` or `MultiBoxCanvas`: those are modals with unsaved edits, and a second dialog over them would race the save. `SlotBboxEditor` has none. - **The proposal is a folded line on the cluster page** that opens the card (and only then asks the server - for the apply dry run). `min_agreement` has no control; the server's default 0.6 applies. + for the apply dry run). `min_agreement` had no control (added afterwards: the dialog's **Agreement needed** field, empty = the server's default). - **Tests.** The controllers, the contract test, the policy panel, the sub-cluster request and the stubbed e2e were written first and watched failing; the three new components' tests were written with the components. - **Docs screenshots.** A capture state (`assist-dialog`) and the dry-run allow-list entries are added to diff --git a/frontend/CHANGELOG.md b/frontend/CHANGELOG.md index 8d5c61295..1427a1bec 100644 --- a/frontend/CHANGELOG.md +++ b/frontend/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- The sub-cluster tabs on the cluster page list the ids the server serves on the cluster card (`subcluster_ids`), so a sub-cluster whose members are not on the loaded page still has its tab. The assist dialog has an **Agreement needed** field (`min_agreement`) for a cluster classify run: the server's default shown, the contract's 0.5 to 1.0 range. - The audit queue shows, on each drawn crop, the machine label and its source the crop carried when it was drawn (`audit_label_name`, `audit_label_source`, now on the item wire, #166). - **Assist…** on the cluster page (the cluster or the open sub-cluster, or the selected crops), on one crop's details panel, on one box in `/review`, and on Matching items (a filter): a dialog that posts one assist run diff --git a/frontend/e2e/stubbed/test_assist_run_dialog.py b/frontend/e2e/stubbed/test_assist_run_dialog.py index b8d1122d8..8547f68e2 100644 --- a/frontend/e2e/stubbed/test_assist_run_dialog.py +++ b/frontend/e2e/stubbed/test_assist_run_dialog.py @@ -42,6 +42,7 @@ "is_unlabeled": True, "representatives": [{"crop_id": "crop-0"}], "n_subclusters": 2, + "subcluster_ids": ["10000a", "10000b"], "updated_at": None, } ], diff --git a/frontend/src/lib/api.clusters.test.ts b/frontend/src/lib/api.clusters.test.ts index 7955e3fc9..fac9bdfa3 100644 --- a/frontend/src/lib/api.clusters.test.ts +++ b/frontend/src/lib/api.clusters.test.ts @@ -189,6 +189,46 @@ describe('getClusters purity_tier/promotable/core_similarity_min', () => { expect(res.items[2]).toMatchObject({ label_agreement: 0.7, dominant_pct: 0.7 }); }); + it('maps the served sub-cluster ids onto the cluster', async () => { + vi.stubGlobal( + 'fetch', + vi.fn().mockResolvedValue( + jsonResponse({ + items: [ + { + cluster_id: 10000, + cluster_kind: 'candidate', + size: 9, + validated_count: 0, + labelled_count: 0, + dominant_class_id: null, + dominant_class_name: null, + dominant_count: 0, + purity: null, + purity_n: 0, + purity_basis: 'nearest_centroid', + purity_tier: null, + label_purity: null, + labelled_share: null, + promotable: false, + is_unlabeled: true, + n_subclusters: 2, + subcluster_ids: ['10000a', '10000b'], + updated_at: null, + representatives: [], + }, + ], + total: 1, + total_class_clusters: 0, + total_candidate_clusters: 1, + cluster_id_offset: 10000, + }), + ), + ); + const res = await getClusters(); + expect(res.items[0]!.subcluster_ids).toEqual(['10000a', '10000b']); + }); + it('getClusters reports purity_thresholds as null when the server omits it, never a stale hardcoded value', async () => { const fetchMock = vi.fn().mockResolvedValue( jsonResponse({ diff --git a/frontend/src/lib/api_clusters.ts b/frontend/src/lib/api_clusters.ts index a0a3bdaa6..599cc5e55 100644 --- a/frontend/src/lib/api_clusters.ts +++ b/frontend/src/lib/api_clusters.ts @@ -55,6 +55,7 @@ type RawCluster = { promotable: boolean; is_unlabeled: boolean; n_subclusters: number; + subcluster_ids: string[]; updated_at: string | null; representatives: Array<{ crop_id: string; @@ -121,6 +122,7 @@ function _rawClusterToCluster( representative_crop_ids: (c.representatives ?? []).map((r) => r.crop_id), has_subclusters: c.n_subclusters > 0, n_subclusters: c.n_subclusters, + subcluster_ids: c.subcluster_ids, updated_at: c.updated_at, }; } @@ -316,6 +318,7 @@ export async function getCluster( is_unlabeled: true, has_subclusters: false, n_subclusters: 0, + subcluster_ids: [], representative_crop_ids: items.slice(0, 4).map((c) => c.id), updated_at: null, }; diff --git a/frontend/src/lib/assistRun/assistRunController.svelte.ts b/frontend/src/lib/assistRun/assistRunController.svelte.ts index 5abb0ec72..ec5906f11 100644 --- a/frontend/src/lib/assistRun/assistRunController.svelte.ts +++ b/frontend/src/lib/assistRun/assistRunController.svelte.ts @@ -74,6 +74,8 @@ export class AssistRunController { acknowledgeExternal = $state(false); promptPack = $state(null); segmentVerify = $state<'vlm' | 'none'>('vlm'); + /** The agreement floor of the label proposal; null = the server's default. */ + minAgreement = $state(null); /** The server's dry run for the current request, or null while unknown. */ preview = $state(null); @@ -138,6 +140,7 @@ export class AssistRunController { if (this.vlm) req.vlm = this.vlm; if (this.promptPack) req.prompt_pack = this.promptPack; if (this.acknowledgeExternal) req.acknowledge_external = true; + if (this.minAgreement !== null) req.min_agreement = this.minAgreement; if (this.actions.includes('segment')) req.segment = { verify: this.segmentVerify }; return req; } @@ -161,6 +164,7 @@ export class AssistRunController { this.size = null; this.seed = 0; this.stratifyBy = 'auto'; + this.minAgreement = null; this.preview = null; this.planError = null; this.planErrorCode = null; diff --git a/frontend/src/lib/assistRun/assistRunController.test.ts b/frontend/src/lib/assistRun/assistRunController.test.ts index 4ccb375e5..0ff317bdc 100644 --- a/frontend/src/lib/assistRun/assistRunController.test.ts +++ b/frontend/src/lib/assistRun/assistRunController.test.ts @@ -61,6 +61,17 @@ describe('AssistRunController request', () => { expect(c.request.sample).toMatchObject({ mode: 'representative', size: null }); }); + it('sends min_agreement only once the user sets it, and a new dialog starts at the server default', () => { + const c = new AssistRunController(deps()); + c.open(CLUSTER_OPENER); + expect(c.request).not.toHaveProperty('min_agreement'); + c.minAgreement = 0.8; + expect(c.request.min_agreement).toBe(0.8); + c.open(CLUSTER_OPENER); + expect(c.minAgreement).toBeNull(); + expect(c.request).not.toHaveProperty('min_agreement'); + }); + it('offers a sample choice only for the scopes the opener gave', () => { const c = new AssistRunController(deps()); c.open({ title: 'This crop', selected: { kind: 'crops', crop_ids: ['c1'] } }); diff --git a/frontend/src/lib/clusters/clusterDetailController.svelte.ts b/frontend/src/lib/clusters/clusterDetailController.svelte.ts index 00fb4c5d7..a9763b7c8 100644 --- a/frontend/src/lib/clusters/clusterDetailController.svelte.ts +++ b/frontend/src/lib/clusters/clusterDetailController.svelte.ts @@ -96,7 +96,6 @@ export function createClusterDetailController() { const res = await getCluster(clusterId, page, pageSize, undefined, query); cluster = res.cluster; cardError = res.cardError; - if (query.clusterSubid == null) rememberSubids(res.crops.items, page === 1); orderMeta = orderMode === 'diverse' ? { @@ -127,18 +126,6 @@ export function createClusterDetailController() { let subTabPick = $state<{ cluster: number; subid: string | null } | null>(null); const subTab = $derived(subTabPick?.cluster === clusterId ? subTabPick.subid : null); - // The sub-cluster ids seen while viewing the whole cluster. A sub-cluster tab - // loads only its own members from the server, so the tab list cannot be read - // off the page that is showing. - let knownSubids = $state([]); - function rememberSubids(items: Crop[], firstPage: boolean): void { - // eslint-disable-next-line svelte/prefer-svelte-reactivity -- local set built and consumed within this function, never stored in reactive state - const seen = new Set(firstPage ? [] : knownSubids); - for (const c of items) if (c.cluster_subid != null) seen.add(c.cluster_subid); - // Lexicographic sort keeps "47a","47b","47aa"... in human-expected order. - knownSubids = [...seen].sort(); - } - // class_source filter (null = all). Drives the chip-group in the // header and propagates to {API_PREFIX}/crops?class_source=... so the grid // shows only crops from one source bucket. Cluster card stats in @@ -281,7 +268,9 @@ export function createClusterDetailController() { // whole; the loaded page is never filtered here. const filteredCrops = $derived(cropPager.items); - const subClusterIds = $derived(knownSubids); + // The tabs are the ids the server serves on the cluster card, not a guess from the + // loaded page. + const subClusterIds = $derived(cluster?.subcluster_ids ?? []); // When refine has produced sub-clusters and we're viewing "all", group // the grid inline by cluster_subid (contiguous groups + a labeled diff --git a/frontend/src/lib/clusters/slotInventory.svelte.ts b/frontend/src/lib/clusters/slotInventory.svelte.ts index f7901ea42..260a14191 100644 --- a/frontend/src/lib/clusters/slotInventory.svelte.ts +++ b/frontend/src/lib/clusters/slotInventory.svelte.ts @@ -76,6 +76,7 @@ export function createSlotInventory() { dominant_class_name: cls.name, dominant_pct: null, n_subclusters: 0, + subcluster_ids: [], has_subclusters: false, representative_crop_ids: reps.map((r) => r.p.crop_id), // Region close-ups, not parent-item thumbnails: the card is the diff --git a/frontend/src/lib/components/assistRun/AssistRunDialog.svelte b/frontend/src/lib/components/assistRun/AssistRunDialog.svelte index 3e3f21287..caf5f4123 100644 --- a/frontend/src/lib/components/assistRun/AssistRunDialog.svelte +++ b/frontend/src/lib/components/assistRun/AssistRunDialog.svelte @@ -19,7 +19,11 @@ type SampleChoice, } from '$lib/assistRun/assistRunController.svelte'; import { isPromptPackAvailable, selectableAxisEntries } from '$lib/strategies'; - import type { AssistAction, AssistStratifyBy } from '$lib/types_assist'; + import { + MIN_AGREEMENT_RANGE, + type AssistAction, + type AssistStratifyBy, + } from '$lib/types_assist'; import { strategiesStore } from '$stores/strategies.svelte'; interface Props { @@ -214,6 +218,26 @@ {/each} + {#if ctl.target?.kind === 'cluster' && ctl.actions.includes('classify')} + + {/if} {#if ctl.actions.includes('segment')}
({ applyAssistRun: vi.fn(async () => ({})), @@ -98,6 +98,25 @@ describe('AssistRunDialog', () => { expect(q('assist-plan-calls')?.textContent).toBe('6'); }); + it('shows the server default for min_agreement and re-plans with the typed value', async () => { + const { ctl, deps } = setup(); + open(ctl); + ctl.setChoice('all'); + await vi.waitFor(() => expect(ctl.preview).not.toBeNull()); + flushSync(); + const field = q('assist-min-agreement') as HTMLInputElement; + expect(field.placeholder).toBe(String(ctl.preview!.request.min_agreement)); + expect(field.min).toBe(String(MIN_AGREEMENT_RANGE.min)); + expect(field.max).toBe(String(MIN_AGREEMENT_RANGE.max)); + field.value = '0.8'; + field.dispatchEvent(new Event('input', { bubbles: true })); + flushSync(); + await vi.waitFor(() => { + const last = vi.mocked(deps.create).mock.calls.at(-1)![0]; + expect(last.min_agreement).toBe(0.8); + }); + }); + it('re-plans with the representative sample, seed and size the user typed', async () => { const { ctl, deps } = setup(); open(ctl); diff --git a/frontend/src/lib/components/slots/SlotGallery.falsePositiveKind.test.ts b/frontend/src/lib/components/slots/SlotGallery.falsePositiveKind.test.ts index 2580bc7c3..735587508 100644 --- a/frontend/src/lib/components/slots/SlotGallery.falsePositiveKind.test.ts +++ b/frontend/src/lib/components/slots/SlotGallery.falsePositiveKind.test.ts @@ -40,6 +40,7 @@ function baseCluster(overrides: Partial): Cluster { representative_crop_ids: [], has_subclusters: false, n_subclusters: 0, + subcluster_ids: [], updated_at: null, ...overrides, }; diff --git a/frontend/src/lib/components/slots/SlotGallery.unclustered.test.ts b/frontend/src/lib/components/slots/SlotGallery.unclustered.test.ts index 3829c0f59..eed9206b6 100644 --- a/frontend/src/lib/components/slots/SlotGallery.unclustered.test.ts +++ b/frontend/src/lib/components/slots/SlotGallery.unclustered.test.ts @@ -38,6 +38,7 @@ function fpCluster(): Cluster { representative_crop_ids: [], has_subclusters: false, n_subclusters: 0, + subcluster_ids: [], updated_at: null, }; } diff --git a/frontend/src/lib/contract/assistContract.test.ts b/frontend/src/lib/contract/assistContract.test.ts index 87d025ecc..4fe93cc5e 100644 --- a/frontend/src/lib/contract/assistContract.test.ts +++ b/frontend/src/lib/contract/assistContract.test.ts @@ -8,7 +8,7 @@ import { describe, expect, it } from 'vitest'; import spec from '$contracts/openapi/curation.json'; import type * as T from '$lib/types_assist'; -import { ASSIST_ACTIONS, ASSIST_STATUSES } from '$lib/types_assist'; +import { ASSIST_ACTIONS, ASSIST_STATUSES, MIN_AGREEMENT_RANGE } from '$lib/types_assist'; type Schema = { properties?: Record; @@ -293,4 +293,12 @@ describe('assist types vs the generated contract', () => { [...schemas.AssistRunState!.properties!.status!.enum!].sort(), ); }); + + it('the min_agreement range the dialog offers is the served one', () => { + const served = schemas.AssistRunRequest!.properties!.min_agreement as unknown as { + minimum: number; + maximum: number; + }; + expect(MIN_AGREEMENT_RANGE).toEqual({ min: served.minimum, max: served.maximum }); + }); }); diff --git a/frontend/src/lib/types.ts b/frontend/src/lib/types.ts index 2d294b2b4..12f20d65b 100644 --- a/frontend/src/lib/types.ts +++ b/frontend/src/lib/types.ts @@ -319,6 +319,9 @@ export interface Cluster { has_subclusters: boolean; /** Distinct cluster_subid count from the backend. */ n_subclusters: number; + /** The cluster's sub-cluster ids (e.g. "47a"), sorted, served by the backend; + * the sub-cluster tabs are exactly these. */ + subcluster_ids: string[]; centroid_sha?: string; updated_at: string | null; /** Set only on the client-built region inventory entry pinned diff --git a/frontend/src/lib/types_assist.ts b/frontend/src/lib/types_assist.ts index 8e7bb52c3..351bc1b5e 100644 --- a/frontend/src/lib/types_assist.ts +++ b/frontend/src/lib/types_assist.ts @@ -61,6 +61,10 @@ export interface AssistFilterTarget { export type AssistTarget = AssistCropsTarget | AssistBoxesTarget | AssistClusterTarget | AssistFilterTarget; +/** The range `min_agreement` accepts (pinned to the served schema by assistContract.test.ts); + * the default is whatever the server echoes on the dry run. */ +export const MIN_AGREEMENT_RANGE = { min: 0.5, max: 1 } as const; + export type AssistSampleMode = 'all' | 'representative'; export type AssistStratifyBy = 'auto' | 'cluster_subid' | 'none'; diff --git a/src/routers/curation/clusters.py b/src/routers/curation/clusters.py index 265348845..4fa083f8f 100644 --- a/src/routers/curation/clusters.py +++ b/src/routers/curation/clusters.py @@ -58,6 +58,9 @@ def _candidate_dominant_name(cls_buckets: list[dict[str, Any]], labelled: int) - return None +MAX_SUBCLUSTER_IDS = 200 +"""The most sub-cluster ids one card lists (``n_subclusters`` stays exact).""" + PURITY_BASIS = 'nearest_centroid' """What a card's ``purity`` measures (served per card): see ``_FITS``.""" @@ -175,7 +178,8 @@ async def list_clusters( member has been measured, * ``is_unlabeled`` (true when no class_name has any signal at all), * ``cluster_kind`` (``class`` | ``candidate`` | ``unassigned``), - * ``n_subclusters`` (distinct cluster_subid values), and + * ``n_subclusters`` (distinct cluster_subid values), ``subcluster_ids`` + (those ids, at most ``MAX_SUBCLUSTER_IDS``), and * ``representatives`` (top ``per_cluster`` crops sorted by cluster_distance asc). @@ -249,6 +253,13 @@ class / purity from a 4-crop sample (it was getting wrong answers # cluster_subid is mapped keyword directly on the live index — # no .keyword subfield exists. 'subclusters': {'cardinality': {'field': 'cluster_subid'}}, + 'subcluster_ids': { + 'terms': { + 'field': 'cluster_subid', + 'size': MAX_SUBCLUSTER_IDS, + 'order': {'_key': 'asc'}, + } + }, # Purity: members measured against their current cluster # by the geometry pass, and those whose nearest centroid is it. 'geometry_measured': {'filter': _MEASURED}, @@ -327,6 +338,11 @@ class / purity from a 4-crop sample (it was getting wrong answers and is_promotable(members=size, labelled=labelled_total, purity=label_purity), 'is_unlabeled': is_unlabeled, 'n_subclusters': n_subclusters, + # Sorted so '7a', '7aa', '7b' read in the order the tabs show; capped at + # MAX_SUBCLUSTER_IDS (n_subclusters stays the true count). + 'subcluster_ids': sorted( + str(b['key']) for b in bucket.get('subcluster_ids', {}).get('buckets', []) + ), 'updated_at': bucket.get('latest_update', {}).get('value_as_string'), # Populated below, only for the [offset, offset+limit) page # — every other card keeps an empty list. diff --git a/tests/curation/test_cluster_card_subcluster_ids.py b/tests/curation/test_cluster_card_subcluster_ids.py new file mode 100644 index 000000000..402ae9691 --- /dev/null +++ b/tests/curation/test_cluster_card_subcluster_ids.py @@ -0,0 +1,61 @@ +"""``GET /clusters`` serves each card's sub-cluster ids, so a client never has to guess the +tab list from the page of crops it happens to have loaded.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from src.routers.curation.clusters import list_clusters +from src.services.curation.item_filter import ItemFilter + + +def _bucket(cid: int, subids: list[str] | None) -> dict[str, Any]: + bucket: dict[str, Any] = { + 'key': cid, + 'doc_count': 9, + 'top_class': {'buckets': []}, + 'labelled': {'doc_count': 0}, + 'validated': {'doc_count': 0}, + 'subclusters': {'value': len(subids or [])}, + 'latest_update': {}, + } + if subids is not None: + bucket['subcluster_ids'] = {'buckets': [{'key': s, 'doc_count': 1} for s in subids]} + return bucket + + +async def _list(*buckets: dict[str, Any]) -> tuple[dict[str, Any], AsyncMock]: + os_client = AsyncMock() + os_client.search = AsyncMock( + return_value={'aggregations': {'clusters': {'buckets': list(buckets)}}} + ) + resp = await list_clusters( + os_client, + per_cluster=0, + max_clusters=100, + kind='all', + cluster_id=None, + item_filter=ItemFilter(), + min_blur_ratio=None, + class_source=None, + ) + return resp, os_client + + +@pytest.mark.asyncio +async def test_each_card_lists_its_subcluster_ids_in_order() -> None: + resp, _ = await _list(_bucket(7, ['7a', '7b', '7aa']), _bucket(8, None)) + cards = {c['cluster_id']: c for c in resp['items']} + assert cards[7]['subcluster_ids'] == ['7a', '7aa', '7b'] + assert cards[8]['subcluster_ids'] == [] + + +@pytest.mark.asyncio +async def test_the_aggregation_asks_for_the_ids_by_key() -> None: + _, os_client = await _list(_bucket(7, [])) + aggs = os_client.search.call_args.kwargs['body']['aggs']['clusters']['aggs'] + assert aggs['subcluster_ids']['terms']['field'] == 'cluster_subid' + assert aggs['subcluster_ids']['terms']['order'] == {'_key': 'asc'} From 0c88519fe33333c50c4c8273ccd6f7f437c929cd Mon Sep 17 00:00:00 2001 From: davidamacey Date: Thu, 15 Oct 2026 18:47:18 -0400 Subject: [PATCH 06/10] docs(examples): add a domain-neutral general_objects prompt pack with item attributes brand (free text), colour and material (closed lists) on the generic item prompts, validated by the pack validation in a test and mentioned in the selective-assist and use-your-own-domain guides. --- CHANGELOG.md | 1 + docs-site/docs/guides/selective-assist.mdx | 4 +- docs-site/docs/guides/use-your-own-domain.mdx | 3 +- examples/prompt_packs/general_objects.json | 99 +++++++++++++++++++ tests/curation/test_pack_attributes.py | 21 ++++ 5 files changed, 126 insertions(+), 2 deletions(-) create mode 100644 examples/prompt_packs/general_objects.json diff --git a/CHANGELOG.md b/CHANGELOG.md index 66c7e5191..8a4fc0552 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,7 @@ history of this codebase and was never published. This release is `[0.5.0]`. - `make pe-build-trt` (`export/build_pe_trt.sh`, #277) no longer ships an FP32 engine silently: a failed or FP32-falling-back FP16 bake, a failed FP16 build or a failed uint8 input step now exits 4 with the cause. `ALLOW_FP32=1` opts in to an FP32 engine knowingly. ### Added +- Example prompt pack `examples/prompt_packs/general_objects.json`: the generic item prompts plus three domain-neutral item attributes (`brand`, `colour`, `material`) for COCO-style objects; validated by the pack validation in a test. - `GET /clusters` cards serve `subcluster_ids` (the sorted sub-cluster ids, at most 200; `n_subclusters` stays the exact count). - The item wire serves the accuracy audit state (#166): `audit_sample`, `audit_batch_id`, `audit_sampled_at`, `audit_label_name`, `audit_label_source`, `audit_outcome` and `audit_human_name` (null for an item never drawn), so `GET /audit/queue` items can show the machine label they were drawn with. - Cropwright **Assist…** (cluster page, one crop, one box, Matching items): a dialog to pick what the VLM sees (selected, all, a representative sample), the actions (classify, attributes, verify, segment) and the VLM and prompt pack, with the server's dry run; the cluster and sub-cluster label proposal card with the agreement, an explicit dry-run-first **Apply** and the `scope_changed` refusal; dashed "VLM suggests" attribute chips; and a **Region verification** switch (`region_verify`) on the VLM scope panel. See the guide [Selective VLM and segmenter assist](docs-site/docs/guides/selective-assist.mdx). diff --git a/docs-site/docs/guides/selective-assist.mdx b/docs-site/docs/guides/selective-assist.mdx index 0cac0a4b8..ff34b528a 100644 --- a/docs-site/docs/guides/selective-assist.mdx +++ b/docs-site/docs/guides/selective-assist.mdx @@ -104,7 +104,9 @@ as a car's make or a wheel's rim material: } ``` -`examples/prompt_packs/vehicle_wheel.json` is a complete one. The VLM answers +`examples/prompt_packs/vehicle_wheel.json` is a complete one, and +`examples/prompt_packs/general_objects.json` is a domain-neutral one for COCO-style +objects (`brand` as free text, `colour` and `material` as closed lists). The VLM answers each attribute with a value and a confidence (`high`, `medium`, `low`). The server drops a key the pack did not declare and a value outside a closed `values` list, and counts those as `invalid_values`. The attributes are stored diff --git a/docs-site/docs/guides/use-your-own-domain.mdx b/docs-site/docs/guides/use-your-own-domain.mdx index 9a38bd5d1..f20f9b8e9 100644 --- a/docs-site/docs/guides/use-your-own-domain.mdx +++ b/docs-site/docs/guides/use-your-own-domain.mdx @@ -98,7 +98,8 @@ See [Region profiles](./region-profiles.mdx) and ## 4. Write a prompt pack The pack holds what the VLM is told to look for and how to answer. Start -from `examples/prompt_packs/vehicle_wheel.json`, or from the built-in +from `examples/prompt_packs/vehicle_wheel.json` (or `general_objects.json`, the +same with three plain item attributes), or from the built-in generic packs. The reply keys the pack must keep are listed by `GET /curation/projects/{project}/prompt_packs/schema`. diff --git a/examples/prompt_packs/general_objects.json b/examples/prompt_packs/general_objects.json new file mode 100644 index 000000000..9efa9bd3f --- /dev/null +++ b/examples/prompt_packs/general_objects.json @@ -0,0 +1,99 @@ +{ + "_comment": "Example prompt pack for general objects (COCO-style classes): the generic item prompts plus three domain-neutral item attributes (brand as free text, colour and material as closed vocabularies). Select it for an assist run with the attributes action; the answers are stored as suggestions in each item's assist object, apart from the class. Import it with the body of POST /prompt_packs (see the selective-assist guide).", + "name": "general_objects", + "class_system": "You are a photo-classification assistant. Classify each numbered item crop into one of the provided class names. If unsure, choose the most likely AND set confidence='low'. Return strict JSON only — no prose, no markdown.", + "class_user_template": "Class names: {class_names_csv}\nLabel each numbered crop. Respond as a JSON array:\n[{{\"img\": 1, \"class\": \"\", \"confidence\": \"high|medium|low\"}}, ...]", + "open_class_system": "You are a photo-classification assistant. For each numbered item crop: (1) classify into one of the provided class names; if NO existing class is a reasonable fit set ``class`` to ``__new__`` and propose a short lowercase snake_case slug in ``proposed_class``. (2) Set confidence to 'high', 'medium', or 'low'. Return strict JSON only — no prose, no markdown.", + "open_class_user_template": "Class names: {class_names_csv}\nLabel each numbered crop. Respond as a JSON array:\n[{{\"img\": 1, \"class\": \"\", \"confidence\": \"high|medium|low\", \"proposed_class\": \"\"}}, ...]", + "combined_system": "You are labeling an item crop. Return STRICT JSON with these keys: class_id (int|null), class_confidence (high|medium|low|null), region_visible (bool: true if the labeled sub-region is visible anywhere in the image, regardless of whether any proposed box below is correct), region_boxes (array, one entry per numbered candidate box shown in the image -- always an array, even for a single box: [{\"box\": , \"region_bbox_correct\": bool|null, \"region_text\": string|null (the characters printed on that box's region, copied verbatim; null when none are legible; never a description of the region or the item class), \"region_confidence\": \"high\"|\"medium\"|\"low\"|null}, ...]). No prose, no markdown.", + "combined_user_template": "{class_block}{region_block}If asked to classify and no class matches, return class_id=-1.\nFor each numbered box: if it correctly outlines the labeled sub-region, set that box's region_bbox_correct=true and transcribe its region_text.\nIf a box is wrong but the sub-region IS visible elsewhere, set that box's region_bbox_correct=false and region_visible=true.\nIf no such sub-region is visible anywhere, set region_visible=false and every box's region_bbox_correct=null.", + "combined_batch_system": "You are labeling numbered item crops. Return STRICT JSON: a single object with key \"results\" whose value is an array of per-image objects (one per numbered image, in input order). Each per-image object has keys: img (1-based index), class_id (int|null), class_confidence (high|medium|low|null), region_visible (bool: true if the labeled sub-region is visible anywhere in that image, regardless of whether any proposed box is correct), region_boxes (array, one entry per numbered candidate box shown in that image -- always an array, even for a single box: [{\"box\": , \"region_bbox_correct\": bool|null, \"region_text\": string|null (the characters printed on that box's region, copied verbatim; null when none are legible; never a description of the region or the item class), \"region_confidence\": \"high\"|\"medium\"|\"low\"|null}, ...]). Output ONLY the JSON object — no prose, no markdown, no reasoning. Skip the chain-of-thought.", + "combined_batch_rules": "Return STRICT JSON of the form {\"results\": [{\"img\": 1, ...}, {\"img\": 2, ...}, ...]}. Rules common to all images:\n- If asked to classify and no class matches, return class_id=-1.\n- For each numbered box: if it correctly outlines the labeled sub-region, set that box's region_bbox_correct=true and transcribe its region_text.\n- If a box is wrong but the sub-region IS visible elsewhere, set that box's region_bbox_correct=false and region_visible=true.\n- If no such sub-region is visible anywhere, set region_visible=false and every box's region_bbox_correct=null.\n- Respond ONLY with the JSON object above. No prose, no markdown, no reasoning preamble.\nPer-image directives follow with each image:", + "region_system": "You verify whether an image shows a real labeled sub-region (e.g. a printed shipping/barcode label) and read the text on it. Output ONLY a single JSON object on the last line — no reasoning, no preamble, no markdown. Reasoning models: skip the chain-of-thought.", + "region_user": "Decide: does this crop show a real printed label region, or something else (blank surface, packaging tape, unrelated marking)? If it does, also read the text on it. Reply with exactly one JSON object using these keys: is_region (boolean), confidence (\"high\" or \"medium\" or \"low\"), reason (string up to 15 words), text (the label text as a string, or null if no label or unreadable), text_confidence (\"high\" or \"medium\" or \"low\" — or null when text is null).", + "region_batch_system": "You verify whether each numbered image shows a real labeled sub-region. Output ONLY a JSON array — one object per image, in input order — no reasoning, no preamble, no markdown. Reasoning models: skip the chain-of-thought.", + "region_batch_user": "For each numbered crop decide: is this a real printed label region, or something else? If it is, also read the text on it. Respond as a JSON array:\n[{\"img\": 1, \"is_region\": true, \"confidence\": \"high|medium|low\", \"reason\": \"<=15 words\", \"text\": \"\" or null, \"text_confidence\": \"high|medium|low\" or null}, ...]\nCopy the text exactly as printed; use null when no text is legible. Never guess or fill in an example value.", + "region_visible_system": "You decide whether each numbered item crop contains a visible labeled sub-region (even partial / angled / small). Output ONLY a JSON object of the form {\"results\": [...]}, one entry per image in input order — no prose, no markdown. Reasoning models: do not echo a chain-of-thought.", + "region_visible_user": "For each numbered crop, answer: is a labeled sub-region visible anywhere in the image? Count partial, angled, or small regions as visible; count blank or missing regions as not visible. Respond as a JSON object whose ``results`` field is an array of per-image verdicts in input order:\n{\"results\": [{\"img\": 1, \"visible\": true|false}, ...]}", + "class_descriptions": {}, + "synonyms": {}, + "proposal_denylist": [ + "abstract*", + "background*", + "*_background", + "object", + "objects", + "*_object", + "blurry*", + "blurred*", + "*_blurry", + "*_blur", + "blur_*", + "out_of_focus*", + "low_quality*", + "low_resolution*", + "unclear*", + "unidentified*", + "scene*", + "*_scene", + "empty*", + "unknown*", + "*_image", + "*_photo", + "*_abstract", + "blank", + "blank_*", + "indoor", + "indoors", + "outdoor", + "outdoors", + "shadow*", + "*_shadow" + ], + "detector_hint_min_confidence_pct": 0, + "attributes": [ + { + "key": "brand", + "description": "the brand or maker, only when a logo or printed name makes it clear", + "values": [], + "applies_to": "item" + }, + { + "key": "colour", + "description": "the main colour of the object", + "values": [ + "black", + "white", + "grey", + "red", + "orange", + "yellow", + "green", + "blue", + "purple", + "brown", + "multicolour" + ], + "applies_to": "item" + }, + { + "key": "material", + "description": "what the object is mostly made of", + "values": [ + "metal", + "plastic", + "wood", + "glass", + "fabric", + "paper", + "ceramic", + "leather" + ], + "applies_to": "item" + } + ], + "assist_system": "You are a photo-classification assistant. For each numbered item crop, classify it into one of the provided class names and report the requested attributes. Attributes are observations: leave one out when the crop does not show it; never guess. Return strict JSON only, no prose, no markdown.", + "assist_user_template": "Class names: {class_list}\nAttributes to report (only those visible):\n{attribute_block}\nAnswer for each of the {n_images} numbered crops as a JSON array:\n[{{\"img\": 1, \"class\": \"\", \"confidence\": \"high|medium|low\", \"proposed_class\": \"\", \"attributes\": {{\"\": {{\"value\": \"\", \"confidence\": \"high|medium|low\"}}}}}}, ...]", + "attributes_system": "You describe numbered crops. Report only the requested attributes, only when the crop shows them; never guess. Return strict JSON only, no prose, no markdown.", + "attributes_user_template": "Attributes to report (only those visible):\n{attribute_block}\nAnswer for each of the {n_images} numbered crops as a JSON array:\n[{{\"img\": 1, \"attributes\": {{\"\": {{\"value\": \"\", \"confidence\": \"high|medium|low\"}}}}}}, ...]" +} diff --git a/tests/curation/test_pack_attributes.py b/tests/curation/test_pack_attributes.py index b7389931c..b5eec9d24 100644 --- a/tests/curation/test_pack_attributes.py +++ b/tests/curation/test_pack_attributes.py @@ -260,3 +260,24 @@ def test_a_pack_without_item_attributes_is_a_422(stack: Any) -> None: assert r.status_code == 422 assert r.json()['detail']['error'] == 'pack_declares_no_attributes' assert stack.upstream.vlm_requests == [] + + +def test_the_general_objects_example_loads_validates_and_declares_domain_neutral_attributes() -> ( + None +): + from pathlib import Path + + path = ( + Path(__file__).resolve().parents[2] / 'examples' / 'prompt_packs' / 'general_objects.json' + ) + pack = PromptPack.from_json(path) + assert pack.name == 'general_objects' + by_key = {a.key: a for a in pack.attributes} + assert set(by_key) == {'brand', 'colour', 'material'} + assert {a.applies_to for a in pack.attributes} == {'item'} + assert by_key['brand'].values == [] # free text + assert 'red' in by_key['colour'].values # a closed vocabulary + body = json.loads(path.read_text()) + body.pop('_comment', None) + body.pop('name') + assert validate_pack(None, body).ok From 50bcb1b11ff9c66921753c312c0f0562d8714408 Mon Sep 17 00:00:00 2001 From: davidamacey Date: Thu, 15 Oct 2026 18:51:55 -0400 Subject: [PATCH 07/10] test(docs): check non-OP_ variables named in docs against the code (#287) New envnames check: a non-OP_ variable written in a code span on a line about env or variables must be read by compose, a script or code (case-insensitive, as pydantic settings are). #289 re-verified: the docs-site tsc passes (no TS5101), the openprocessor start comment no longer exists, and slow_request_threshold_ms is read (factory.py), so nothing to fix there. --- CHANGELOG.md | 1 + scripts/docs/check_docs_vs_code.py | 56 +++++++++++++++++++++++++++++- tests/test_docs_vs_code.py | 13 ++++++- 3 files changed, 68 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8a4fc0552..aa0ec6c58 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,7 @@ history of this codebase and was never published. This release is `[0.5.0]`. - `make pe-build-trt` (`export/build_pe_trt.sh`, #277) no longer ships an FP32 engine silently: a failed or FP32-falling-back FP16 bake, a failed FP16 build or a failed uint8 input step now exits 4 with the cause. `ALLOW_FP32=1` opts in to an FP32 engine knowingly. ### Added +- `scripts/docs/check_docs_vs_code.py` gains an `envnames` check: a non-`OP_` variable a doc names in a code span, on a line about env or variables, must be read by a compose file, script or code (#287). - Example prompt pack `examples/prompt_packs/general_objects.json`: the generic item prompts plus three domain-neutral item attributes (`brand`, `colour`, `material`) for COCO-style objects; validated by the pack validation in a test. - `GET /clusters` cards serve `subcluster_ids` (the sorted sub-cluster ids, at most 200; `n_subclusters` stays the exact count). - The item wire serves the accuracy audit state (#166): `audit_sample`, `audit_batch_id`, `audit_sampled_at`, `audit_label_name`, `audit_label_source`, `audit_outcome` and `audit_human_name` (null for an item never drawn), so `GET /audit/queue` items can show the machine label they were drawn with. diff --git a/scripts/docs/check_docs_vs_code.py b/scripts/docs/check_docs_vs_code.py index 61f0c898f..652a64e3a 100755 --- a/scripts/docs/check_docs_vs_code.py +++ b/scripts/docs/check_docs_vs_code.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Check the documentation against the code. -Nine checks, each reporting every miss (not just the first): +Ten checks, each reporting every miss (not just the first): * routes - every ``METHOD /path`` written in a doc exists in the generated OpenAPI (``contracts/openapi/*.json``) or, for non-curation routes, in the @@ -20,6 +20,8 @@ ``docs-site/static/img/screenshots/``. * envfiles - every ``NAME=`` line in a fenced block names a setting some compose file, script or code reads (a name the block itself expands as ``$NAME`` is a local shell variable). +* envnames - every non-``OP_`` variable a doc names in a code span, on a line that says env, + variable or assigns it, is read by a compose file, script or code. * links - every relative link resolves to a file, and every ``#anchor`` that points into a markdown file matches a heading in that file. @@ -456,6 +458,57 @@ def check_envfiles() -> list[str]: return errors +_NAME_SPAN_RE = re.compile(r'`([A-Z][A-Z0-9]*(?:_[A-Z0-9]+)+)(?:=[^`\n]*)?`') +_ENV_CONTEXT_RE = re.compile(r'\benv(?:ironment)?\b|\.env\b|variable|\bexport\b|=', re.I) +# Roots beyond _CODE_ENV_ROOTS that read variables (the export scripts, the monitoring stack, the +# frontend build config). +_NAME_EXTRA_ROOTS = ('export', 'monitoring', 'frontend/src', 'frontend/static') +# Variables that belong to a tool the code does not read itself. +_FOREIGN_ENV_NAMES = frozenset({'LD_LIBRARY_PATH'}) + + +def check_env_names() -> list[str]: + """A non-``OP_`` variable a doc names as a variable (an inline code span on a line that says + env, environment, variable, export or assigns it) must be read by a compose file, script or + code, or be a known foreign name. Fenced ``NAME=`` lines are ``envfiles``; ``OP_`` names are + ``env``.""" + parts = [ + _env_corpus(), + (REPO_ROOT / 'env.template').read_text(encoding='utf-8', errors='replace'), + ] + for root in _NAME_EXTRA_ROOTS: + base = REPO_ROOT / root + parts.extend( + p.read_text(encoding='utf-8', errors='replace') + for p in base.rglob('*') + if p.is_file() + and not set(p.relative_to(REPO_ROOT).parts) & _SKIP_DIR_NAMES + and p.suffix + in {'.py', '.sh', '.ts', '.js', '.mjs', '.svelte', '.yml', '.yaml', '.json', ''} + ) + corpus = '\n'.join(parts) + errors: list[str] = [] + for doc in _user_docs(): + if 'releases' in doc.parts: # historical notes name variables of their day + continue + text = re.sub(_FENCE_RE, '', doc.read_text(encoding='utf-8', errors='replace')) + names = { + name + for line in text.splitlines() + if _ENV_CONTEXT_RE.search(line) + for name in _NAME_SPAN_RE.findall(line) + } + for name in sorted(names): + if name.startswith('OP_') or name in _FOREIGN_ENV_NAMES: + continue + # Case-insensitive: pydantic settings read slow_request_threshold_ms as SLOW_REQUEST_THRESHOLD_MS. + if not re.search(rf'\b{name}\b', corpus, re.I): + errors.append( + f'{_rel(doc)}: {name} is not read by any code' + ) # formatted message built next to its check + return errors + + CHECKS = { 'routes': check_routes, 'env': check_env, @@ -465,6 +518,7 @@ def check_envfiles() -> list[str]: 'make': check_make_targets, 'screenshots': check_screenshots, 'envfiles': check_envfiles, + 'envnames': check_env_names, 'links': check_links, } diff --git a/tests/test_docs_vs_code.py b/tests/test_docs_vs_code.py index ea6f0dfeb..695c3e22f 100644 --- a/tests/test_docs_vs_code.py +++ b/tests/test_docs_vs_code.py @@ -32,7 +32,8 @@ def _reset_only(): @pytest.mark.parametrize( - 'name', ['routes', 'env', 'compose', 'template', 'cli', 'make', 'screenshots', 'envfiles'] + 'name', + ['routes', 'env', 'compose', 'template', 'cli', 'make', 'screenshots', 'envfiles', 'envnames'], ) def test_real_docs_match_the_code(name: str) -> None: errors = checker.CHECKS[name]() @@ -222,3 +223,13 @@ def test_envfiles_check_rejects_an_assigned_name_nothing_reads(tmp_path: Path) - bad = _check_one(tmp_path, '```bash\nNOT_A_REAL_SETTING=1\n```\n', 'envfiles') assert len(bad) == 1 assert 'NOT_A_REAL_SETTING' in bad[0] + + +def test_envnames_check_rejects_a_named_variable_nothing_reads(tmp_path: Path) -> None: + ok = 'Set `COMPOSE_PROJECT_NAME` in the env file.\n' + assert _check_one(tmp_path, ok, 'envnames') == [] + prose = 'The `RAW_THING_NOBODY_HAS` constant is just prose.\n' + assert _check_one(tmp_path, prose, 'envnames') == [] # no env context on the line + bad = _check_one(tmp_path, 'Set the variable `NOT_A_REAL_SETTING` to 1.\n', 'envnames') + assert len(bad) == 1 + assert 'NOT_A_REAL_SETTING' in bad[0] From 847805af3710e3af073ef3e42fd95239b5fe36ea Mon Sep 17 00:00:00 2001 From: davidamacey Date: Thu, 15 Oct 2026 19:37:16 -0400 Subject: [PATCH 08/10] fix(e2e): default the audit draw fields in the stubbed item wire fixture --- frontend/e2e/fixtures/wire.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/frontend/e2e/fixtures/wire.py b/frontend/e2e/fixtures/wire.py index 4398ec90c..15dd69809 100644 --- a/frontend/e2e/fixtures/wire.py +++ b/frontend/e2e/fixtures/wire.py @@ -303,6 +303,14 @@ def projects_response( "mask_polygon": None, "region_gate_skip": None, "assist": None, + # Audit draw state the audit queue shows (#166); all null off the audit queue. + "audit_sample": None, + "audit_batch_id": None, + "audit_sampled_at": None, + "audit_label_name": None, + "audit_label_source": None, + "audit_outcome": None, + "audit_human_name": None, # Adopted on the frontend by # readSlot (SlotData.text.choice/invalidReason, # SlotData.subBox.candidate, SlotData.lifecycle.validated/ From 2395a09f89891fade4ed42b938bc21be9e19a109 Mon Sep 17 00:00:00 2001 From: davidamacey Date: Thu, 15 Oct 2026 23:45:25 -0400 Subject: [PATCH 09/10] fix(ingest): keep the dispatch tick on a worker thread; order assist runs by microsecond ids The tick does blocking I/O and stays off the event loop (existing test); only the run and poll task creation is handed to the loop captured in run_tick. Assist run ids carry microseconds so two runs in one second sort newest first. --- src/services/curation/assist/job.py | 2 +- .../curation/bulk_ingest/dispatcher.py | 31 +++++++++++++++---- src/services/curation/bulk_ingest/runner.py | 19 +++++++++++- .../curation/bulk_ingest/source_runs.py | 15 +++++++-- 4 files changed, 57 insertions(+), 10 deletions(-) diff --git a/src/services/curation/assist/job.py b/src/services/curation/assist/job.py index 140ae1c92..bd81f823d 100644 --- a/src/services/curation/assist/job.py +++ b/src/services/curation/assist/job.py @@ -86,7 +86,7 @@ def create_run( 'an assist run is already running for this project; cancel it or wait', run_id=live[0] if live else None, ) - run_id = f'asr_{datetime.now(UTC):%Y%m%dT%H%M%S}_{uuid.uuid4().hex[:6]}' + run_id = f'asr_{datetime.now(UTC):%Y%m%dT%H%M%S%f}_{uuid.uuid4().hex[:6]}' job = _job(run_id) job.directory.mkdir(parents=True) (job.directory / 'request.json').write_text(json.dumps(request), encoding='utf-8') diff --git a/src/services/curation/bulk_ingest/dispatcher.py b/src/services/curation/bulk_ingest/dispatcher.py index a250e7d56..770a1e8b0 100644 --- a/src/services/curation/bulk_ingest/dispatcher.py +++ b/src/services/curation/bulk_ingest/dispatcher.py @@ -61,6 +61,8 @@ FAILED_RUN_HOLD_S = 300.0 ActionKind = Literal['run_scan', 'poll', 'drain'] _scans: dict[tuple[str, str], asyncio.Task[None]] = {} +# 'tasks': the loop run and poll tasks belong to (the tick itself runs in a thread). +_loops: dict[str, asyncio.AbstractEventLoop] = {} @dataclass(frozen=True) @@ -159,6 +161,7 @@ def _start(source: IngestSource, scan: bool) -> None: api_prefix=get_curation_config().api_prefix, project=get_curation_config().project_slug, scan=scan, + loop=_loops.get('tasks'), ) except runner.BulkBusyError: return # another run holds the project; the next tick tries again @@ -166,6 +169,13 @@ def _start(source: IngestSource, scan: bool) -> None: source_state(source.source_id).update(last_fire_at=datetime.now(UTC).isoformat()) +def _on_task_loop() -> bool: + try: + return asyncio.get_running_loop() is _loops.get('tasks') + except RuntimeError: + return False + + def _execute(action: Action, source: IngestSource) -> None: key = (get_curation_config().project_slug, source.source_id) if action.kind == 'poll': @@ -174,9 +184,17 @@ def _execute(action: Action, source: IngestSource) -> None: return # Stamped here, under the tick's lock, so a second worker's tick sees it at once. source_state(source.source_id).update(last_poll_started_at=time.time()) - task = asyncio.create_task(_poll(source, start_run=_start)) - _scans[key] = task - task.add_done_callback(lambda _t: _scans.pop(key, None)) + + def begin() -> None: + task = asyncio.create_task(_poll(source, start_run=_start)) + _scans[key] = task + task.add_done_callback(lambda _t: _scans.pop(key, None)) + + task_loop = _loops.get('tasks') + if task_loop is not None and task_loop.is_running() and not _on_task_loop(): + task_loop.call_soon_threadsafe(begin) + else: + begin() elif action.kind == 'run_scan': _start(source, True) else: @@ -220,11 +238,12 @@ def run_project_tick(*, now: float | None = None) -> list[Action]: async def run_tick() -> None: from src.services.projects.bootstrap import for_each_project + _loops['tasks'] = asyncio.get_running_loop() for slug in for_each_project(): try: - # On the loop, not a worker thread: a started run and a poll are asyncio - # tasks. The tick is a few small file reads and a non-blocking lock. - run_project_tick() + # The tick does blocking file I/O, so it runs in a thread; the run and the poll + # are asyncio tasks, created on `_loops['tasks']`. + await asyncio.to_thread(run_project_tick) except Exception as exc: # one project's failure must not stop the others' ticks logger.warning('ingest_dispatch_failed', project=slug, error=str(exc)) diff --git a/src/services/curation/bulk_ingest/runner.py b/src/services/curation/bulk_ingest/runner.py index 22ffe1d2f..0e750a1a3 100644 --- a/src/services/curation/bulk_ingest/runner.py +++ b/src/services/curation/bulk_ingest/runner.py @@ -474,6 +474,17 @@ async def run(self) -> None: self.job.update(status=status, error=self._abort, finished_at=now_iso()) +def _on_loop(loop: asyncio.AbstractEventLoop) -> bool: + try: + return asyncio.get_running_loop() is loop + except RuntimeError: + return False + + +async def _create_task(coroutine: Any) -> asyncio.Task[None]: + return asyncio.create_task(coroutine) + + def spawn( *, sender: BatchSender, @@ -481,7 +492,10 @@ def spawn( source: BulkSource, options: BulkOptions, listener: RunListener | None = None, + loop: asyncio.AbstractEventLoop | None = None, ) -> None: + """Start the run as a task. From a worker thread (the dispatcher's tick) pass the + running ``loop`` the task belongs to.""" run = BulkIngestRun( sender=sender, store=store, source=source, options=options, listener=listener ) @@ -499,7 +513,10 @@ async def runner() -> None: coroutine = runner() try: - task = asyncio.create_task(coroutine) + if loop is not None and loop.is_running() and not _on_loop(loop): + task = asyncio.run_coroutine_threadsafe(_create_task(coroutine), loop).result(10) + else: + task = asyncio.create_task(coroutine) except RuntimeError as exc: # no running loop: the claimed run must not stay queued coroutine.close() store.job.update( diff --git a/src/services/curation/bulk_ingest/source_runs.py b/src/services/curation/bulk_ingest/source_runs.py index a78556102..03d2309e8 100644 --- a/src/services/curation/bulk_ingest/source_runs.py +++ b/src/services/curation/bulk_ingest/source_runs.py @@ -9,7 +9,7 @@ import time from datetime import UTC, datetime -from typing import Any +from typing import TYPE_CHECKING, Any from src.services.curation.bulk_ingest import runner from src.services.curation.bulk_ingest.queue_listener import QueueListener @@ -21,6 +21,10 @@ from src.services.curation.bulk_ingest.store import ACTIVE_STATUSES, BulkStore, open_store +if TYPE_CHECKING: + import asyncio + + def bulk_source(source: IngestSource, *, scan: bool) -> BulkSource: return BulkSource('queue', source.source_id, tuple(source.extensions), scan) @@ -30,7 +34,13 @@ def bulk_options(source: IngestSource) -> runner.BulkOptions: def start_source_run( - source: IngestSource, *, base_url: str, api_prefix: str, project: str, scan: bool + source: IngestSource, + *, + base_url: str, + api_prefix: str, + project: str, + scan: bool, + loop: asyncio.AbstractEventLoop | None = None, ) -> tuple[BulkStore, bool]: """Claim and spawn a run draining ``source``'s queue (after a scan when ``scan``). Raises :class:`~src.services.curation.bulk_ingest.runner.BulkBusyError` while any run @@ -47,6 +57,7 @@ def start_source_run( source=spec, options=options, listener=QueueListener(source), + loop=loop, ) return store, resumed From ebe7d438b0141bf28969ff5a9acb6a16d70e735f Mon Sep 17 00:00:00 2001 From: davidamacey Date: Thu, 15 Oct 2026 23:47:21 -0400 Subject: [PATCH 10/10] fix(ingest): decode a JPEG through Pillow when libjpeg-turbo cannot read its header simplejpeg raised 'Could not determine subsampling level' on 2 of 100k public JPEGs that Pillow decodes; only that header error now falls back to the Pillow path (same orientation and RGB handling). Other fast-decoder errors (damaged files) still fail. The GPU path already falls back to this CPU decode on any non-ok status, so no device accounting changes. --- CHANGELOG.md | 1 + src/services/curation/image_decode.py | 17 +++++-- .../test_image_decode_header_fallback.py | 47 +++++++++++++++++++ 3 files changed, 62 insertions(+), 3 deletions(-) create mode 100644 tests/curation/test_image_decode_header_fallback.py diff --git a/CHANGELOG.md b/CHANGELOG.md index aa0ec6c58..fd003ca00 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ history of this codebase and was never published. This release is `[0.5.0]`. ## [Unreleased] ### Fixed +- Ingest no longer fails a JPEG that Pillow reads but libjpeg-turbo's header read refuses (`tjDecompressHeader3(): Could not determine subsampling level`, 2 of 100,000 Open Images files, `decode_failed`): that error alone sends the image to the Pillow decode; a damaged file still fails. - Dispatcher-started bulk-ingest runs (poll, cron, webhook) never started: the tick ran in a worker thread and both the run and the poll called `asyncio.create_task` there (`ingest_dispatch_failed`, `no running event loop`). The tick now runs on the loop. A run whose spawn fails is marked `failed` instead of staying `queued` (which kept every later start, and the project's delete, answering 409), and `POST /ingest/bulk/{run_id}/cancel` finalizes an orphan (a run `queued` past 10 s with no task, or `interrupted` after a restart) as `cancelled`. - `DELETE /projects/{p}` (and archive) answering `409 project_busy` right after an ingest finished (#249): the detection worker refreshed its in-flight count only every 15 s, so the last batch read as busy for up to that long after it had landed. The worker now rewrites a project's liveness within a second of its count changing, and the `project_busy` message names each running job and says when a retry is the answer. - `make pe-build-trt` (`export/build_pe_trt.sh`, #277) no longer ships an FP32 engine silently: a failed or FP32-falling-back FP16 bake, a failed FP16 build or a failed uint8 input step now exits 4 with the cause. `ALLOW_FP32=1` opts in to an FP32 engine knowingly. diff --git a/src/services/curation/image_decode.py b/src/services/curation/image_decode.py index bcb703036..05ec5a7c1 100644 --- a/src/services/curation/image_decode.py +++ b/src/services/curation/image_decode.py @@ -5,7 +5,8 @@ identical, at about half of Pillow's CPU on a 12-24 MP photo, and the result is a numpy array the blur scoring reads without the 54 MB copy ``np.asarray(PIL image)`` makes. Pillow still holds the frame for the letterbox, the whole-frame square and the crops (one copy into its own layout). -Everything else, CMYK JPEGs included, decodes through Pillow as before. +Everything else, CMYK JPEGs and a JPEG whose header libjpeg-turbo cannot read (no +subsampling level) included, decodes through Pillow. Damaged files are refused as before: ``simplejpeg`` raises on libjpeg's recoverable errors (``strict``), so a truncated JPEG fails the decode instead of ingesting a grey tail. @@ -23,6 +24,7 @@ EXIF_ORIENTATION = 0x0112 NOT_DECODABLE = 'not a decodable image' +HEADER_REJECTED = 'Could not determine subsampling level' class ImageDecodeError(ValueError): @@ -79,8 +81,17 @@ def _decode(data: bytes) -> DecodedImage: if header.format == 'JPEG' and header.mode in ('RGB', 'L'): import simplejpeg - rgb = orient_array(simplejpeg.decode_jpeg(data, colorspace='RGB'), orientation) - return DecodedImage(Image.fromarray(rgb), rgb) + try: + decoded = simplejpeg.decode_jpeg(data, colorspace='RGB') + except ValueError as exc: + # libjpeg-turbo's header read refuses some JPEGs Pillow reads fine (it cannot + # determine the subsampling). That, and only that, takes the Pillow path; a + # damaged file (any other error) still fails. + if HEADER_REJECTED not in str(exc): + raise + else: + rgb = orient_array(decoded, orientation) + return DecodedImage(Image.fromarray(rgb), rgb) img = ImageOps.exif_transpose(header) if orientation != 1 else header if img.mode != 'RGB': img = img.convert('RGB') diff --git a/tests/curation/test_image_decode_header_fallback.py b/tests/curation/test_image_decode_header_fallback.py new file mode 100644 index 000000000..9fea43964 --- /dev/null +++ b/tests/curation/test_image_decode_header_fallback.py @@ -0,0 +1,47 @@ +"""A JPEG the fast decoder's header read rejects (``tjDecompressHeader3(): Could not determine +subsampling level``) still decodes through Pillow, which reads it; other fast-decoder errors +(a damaged file) keep failing.""" + +from __future__ import annotations + +import io + +import numpy as np +import pytest +import simplejpeg +from PIL import Image + +from src.services.curation.image_decode import ImageDecodeError, decode_image + + +HEADER_ERROR = 'tjDecompressHeader3(): Could not determine subsampling level of JPEG image' + + +def _jpeg() -> bytes: + buf = io.BytesIO() + Image.fromarray(np.full((16, 24, 3), 90, dtype=np.uint8)).save(buf, 'JPEG') + return buf.getvalue() + + +def _reject(message: str, monkeypatch: pytest.MonkeyPatch) -> None: + def fail(*_a: object, **_k: object) -> None: + raise ValueError(message) + + monkeypatch.setattr(simplejpeg, 'decode_jpeg', fail) + + +def test_a_header_the_fast_decoder_rejects_decodes_through_pillow( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _reject(HEADER_ERROR, monkeypatch) + out = decode_image(_jpeg()) + assert out.pil.mode == 'RGB' + assert out.pil.size == (24, 16) + + +def test_any_other_fast_decoder_error_still_fails_the_image( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _reject('Premature end of JPEG file', monkeypatch) + with pytest.raises(ImageDecodeError): + decode_image(_jpeg())