feat(kernel): IngestJob lifecycle — transition table, persisted progress, per-file report, resume (#19)

docs/README.md

@@ -9,7 +9,7 @@ contracts (kernel purity, headless annotator) are described there and enforced i
 | [workspaces.md](workspaces.md) | The workspace on disk: layout, `init`/`open`, project-name uniqueness, and how services are composed |
 | [projects.md](projects.md) | The project lifecycle: the 1:1 dataset, renaming, and what deletion does and does not destroy |
 | [sources.md](sources.md) | Where raw data comes from: the two registration methods, what a video source records from the probe, why decomposition parameters live on the source, and the idempotency rule and its named uniqueness gap |
-| [ingest.md](ingest.md) | Turning a source into rows: content identity versus recorded origin, the two source paths, why the decode happens outside a transaction, and the per-file report |
+| [ingest.md](ingest.md) | Turning a source into rows: content identity versus recorded origin, the two source paths, why the decode happens outside a transaction, the run's lifecycle and pollable progress, and the per-file report |
 | [schemas.md](schemas.md) | The annotation schema: immutable monotonic versions, additive vs destructive change, and the two gates on narrowing |
 | [batches.md](batches.md) | The unit of annotation work: the state machine, membership frozen at approval, the schema pin, and the exact partition into jobs |
 | [jobs.md](jobs.md) | Annotation jobs: the job and per-asset progress machines, what counts as settled, ordered `next_pending`, and derived progress |

docs/ingest.md

@@ -64,6 +64,77 @@ offer is a policy the kernel would be inventing.
 tests. Decoding each one again to re-confirm it would also route our own encoder's output into an
 operator's per-file report — a failure nobody could act on.
 
+## The run has a lifecycle, and it is a table
+
+`INGEST_TRANSITIONS` in `domain/ingest.py` is the whole of what is legal. `IngestService`
+consults it through `require_move`; nothing restates it.
+
+```
+pending ──▶ running ──▶ completed
+   │           │
+   └────────▶ failed ──▶ running        (resume)
+```
+
+A job is created `pending` and moved to `running` by whoever picks the work up. Today that is
+the same call, and the state is over in microseconds — it is spelled out anyway because it is
+the vocabulary a queue needs, and adding it later would mean changing what a stored row means.
+
+**`failed → running` is the only backward edge in this kernel**, and the argument against
+reopening a [batch](batches.md) does not carry over. A batch pins a schema version at approval
+and its jobs are already cut against that pin, so un-freezing one would invalidate work already
+done. Nothing is pinned against an ingest run. It is a record of work, not an artifact with
+dependents — so resuming is the same unit of work continuing, on the same row. A second row per
+attempt would fork `batch_id` and turn `IngestService.list` into a list of retries.
+
+**`running → running` is deliberately missing**, so a run stuck at `running` cannot be resumed.
+That state is a process that died without reporting anything, not a failure anybody can read.
+The remedy already exists — ingest the source again, which creates nothing — and it leaves the
+stuck row as the only evidence the crash left.
+
+## Progress a caller can poll
+
+`processed` and `total` are written to the row **as the run goes**, so
+`IngestService.get(job_id)` answers "where is it now" rather than "where did it end". That is
+the contract the HTTP API and the UI will reuse; nothing about it is specific to being in the
+same process.
+
+| | what it means |
+| --- | --- |
+| `processed` | items dealt with — decoded and stored, or reported as unreadable |
+| `total` | items the source offered, or **NULL** when that is not knowable up front |
+
+A directory can be listed, so it states its total before the first file; an empty one records
+`0 of 0` rather than nothing. A clip cannot: `VideoMetadata` carries no frame count by design —
+it would be a guess for a variable-rate clip, and the number an ingest wants is what extraction
+actually produced — so `total` stays NULL and `processed` climbs alone.
+
+The counter is written **once per item**, not on a cadence. An interval that suits five files
+and one that suits fifty thousand are different numbers, and this service cannot know which it
+is looking at; the cost of not choosing is one small commit beside a decode and a hash that
+cost an order of magnitude more.
+
+## Resuming a failed run
+
+`IngestService.resume(job_id)` re-runs a failed job on its own row, into the batch the first
+attempt was headed for. What qualifies is whatever the table says can reach `running` — `failed`,
+and also `pending`, which a synchronous run never leaves behind but a queued one would. A
+`completed` or `running` job is refused with an ordinary `InvalidTransition` rather than an error
+of its own.
+
+It is a **redo, not a skip**. There is no per-file record of what the previous attempt managed,
+and there does not need to be: blobs are content-addressed and assets are deduplicated by
+content, so re-reading the whole source creates nothing it created before. The cost is
+re-hashing what is already stored; what it buys is that resume has no second code path to get
+wrong.
+
+The counters, the per-file report and the fatal `error` are reset when the attempt starts, so
+they describe the run somebody is watching rather than the one that failed. A run that *failed*
+keeps them exactly where they stopped, which is the first thing anyone reading a failure wants.
+
+`batch_name` is a column for this reason alone: a run that died during the decode reached no
+batch, so without it a resumed run would fall back to naming the batch after the source folder
+and quietly lose the name the caller asked for.
+
 ## Four transactions, and the middle of the run is in none of them
 
 1. Resolve the source, decide the target batch, and insert the `IngestJob` as `running`.
@@ -81,9 +152,14 @@ row exists: `BlobStore.put` is not transactional and a rollback cannot unwrite i
 nothing points at is harmless (content-addressed, shared, never deleted), while a row naming bytes
 that were never stored is not.
 
+The progress writes that happen *between* items are not a contradiction. Each is one `UPDATE`
+that opens and commits while nothing is being decoded; what the single-writer warning is about
+is a transaction held **across** the decode, not the existence of writes during that phase.
+
 The honest consequence, stated rather than hidden: a process killed between transactions can leave
-assets in the project with no batch and a job stuck at `running`. That is recoverable, and finding
-it is what the job record is for.
+assets in the project with no batch and a job stuck at `running`. Finding it is what the job
+record is for, and ingesting the source again is what fixes it — see the lifecycle above for why
+that is the remedy rather than a resume.
 
 Within a run, each file is **probed before it is stored**, so a file that is going to be refused
 never leaves a blob behind.
@@ -97,9 +173,14 @@ name and the reason are kept apart so a report renders as a table rather than as
 sentences, and `IngestFailureKind` exists so it can be **grouped** — real data loss must not be
 buried under ordinary operator noise.
 
+The report is on the row as well as in the return value, written as the run goes rather than at
+the end: a report that only appeared once the run finished would be invisible for exactly as
+long as it is interesting.
+
 A missing ffmpeg is not a file's fault at all. `MediaToolUnavailable` is recorded as the job's
-`error`, the job is marked `failed`, and it is re-raised — which is precisely why it sits outside
-the `MediaError` family. One broken machine is not five thousand broken files.
+`error` — a separate column from the per-file `failures`, which stays empty — the job is marked
+`failed`, and it is re-raised, which is precisely why it sits outside the `MediaError` family.
+One broken machine is not five thousand broken files.
 
 ## The target batch
 
@@ -113,11 +194,10 @@ order for a directory and frame order for a clip.
 
 ## What is deliberately not here yet
 
-- **No state machine, no persisted progress, no persisted report.** The job is written straight to
-  `completed` or `failed`, and `IngestResult` lives only in memory. The transition table, the
-  processed/total counters a caller can poll mid-run, and the report as columns are #19's, which
-  adds them *around* this working path rather than rewriting it.
 - **No thumbnails.** Generating one per asset at ingest and recording `asset.thumbnail_hash` is
   #21, for the M5 gallery.
 - **No background execution.** A run is synchronous and in-process. The service API is shaped so
-  that moving it behind a queue changes the caller's waiting, not its vocabulary.
+  that moving it behind a queue changes the caller's waiting, not its vocabulary — which is why
+  a job is created `pending` and why progress is read off the row rather than off a callback.
+- **No cross-attempt history.** The report and the counters describe the current attempt. A
+  resumed run overwrites them, and a log of every attempt would be its own table.

docs/persistence.md

@@ -115,8 +115,9 @@ MIGRATIONS: list[Migration] = [
     Migration(version=6, name="release_manifest_pointer", upgrade=...),
     Migration(version=7, name="source_provenance", upgrade=...),
     Migration(version=8, name="ingest_pipeline", upgrade=...),
+    Migration(version=9, name="ingest_job_progress", upgrade=...),
 ]
-FORMAT_VERSION: int = MIGRATIONS[-1].version  # 8
+FORMAT_VERSION: int = MIGRATIONS[-1].version  # 9
 ```
 
 `initialize()` reads the version stamped in `_visionset_meta` and runs whatever is
@@ -167,6 +168,15 @@ at 7 and will never re-run it. Declared anywhere else, the
 `create_all` path and the `ALTER` path emit different `CREATE TABLE` text and the
 fresh-versus-migrated test fails — which is exactly what it is for.
 
+Migration 009 is where that expiry actually bit. `ingest_job` was rebuilt by 008, so its column
+order was free *then*; by 009 the table holds real rows and its four new columns —
+`batch_name`, `processed`, `total`, `failures` — arrive by `ALTER` and sit last, in the order
+the migration adds them. That path has a test of its own
+(`test_migration_nine_alters_a_table_migration_eight_rebuilt`), because the fresh-versus-migrated
+test walks back to generation 1, from where 008 re-creates the table whole and 009 finds its
+columns already present. A migration whose only exercise is through an earlier rebuild is not
+exercised at all.
+
 **A column that carries a foreign key cannot arrive by `ALTER` at all.** SQLite spells an added
 key inline on the column; `create_all` spells one as a table constraint. The two texts differ, so
 the fresh-versus-migrated test fails — and dropping the key instead is not free either.
@@ -203,10 +213,20 @@ predates the ingest pipeline", where a `server_default` would invent a format no
 it does refuse is data its two new unique indexes cannot accept, counted before either index is
 created so an `IntegrityError` never escapes `initialize()`.
 
+Migration 009 is the plainest in the file, and the plainness is the point after 008: four
+columns, none carrying a foreign key, so `ALTER` can express all of them — and every one has an
+honest value for a row written before it. A pre-#19 run counted nothing and reported nothing,
+which is exactly what `0` and `[]` say; NULL is what a run that named no batch meant. So unlike
+006, 007 and 008 it refuses nothing and drops nothing. `failures` is a JSON column rather than a
+child table on the criteria above: a per-file report is an immutable value read whole, and
+nothing queries a single failed file in SQL.
+
 The fresh-versus-migrated test is only as strong as how far back
 `_downgrade_to_version_one` walks, so every migration added there needs its undo added too.
 Migrations 006 and 007 are the two places that undo cannot borrow its DDL from `_tables`,
-because `_tables` no longer describes the shape it is restoring.
+because `_tables` no longer describes the shape it is restoring. Migration 009 is the one
+place that needs no undo of its own: its columns live on `ingest_job`, which 008's undo rebuilds
+from scratch, so restoring the generation-1 shape removes them along with everything else.
 
 `format_version` here is the *database* generation. Validating the on-disk workspace layout
 around it — directories, the blob-store root, what makes a directory a workspace at all —

src/visionset/kernel/adapters/_mappers.py

@@ -7,10 +7,10 @@
 fourteen times against fourteen tables.
 
 Most entities are flat — every field is a column — and share
-``_flat_mapping``. The seven that are not say so explicitly:
+``_flat_mapping``. The eight that are not say so explicitly:
 
-- ``AnnotationSchema`` and ``Annotation`` hold immutable nested values, encoded
-  as JSON.
+- ``AnnotationSchema``, ``Annotation`` and ``IngestJob`` hold immutable nested
+  values, encoded as JSON.
 - ``Batch`` and ``AnnotationJob`` own child tables, so their mappings carry a
   ``sync_children`` hook and rebuild their collections on read.
 - ``DatasetChange``, ``Release`` and ``Source`` encode a timezone-aware
@@ -42,7 +42,9 @@
     DatasetChange,
     DatasetMember,
     Geometry,
+    IngestFailure,
     IngestJob,
+    IngestState,
     LabelClass,
     Project,
     Release,
@@ -125,6 +127,37 @@ def _schema_to_domain(_: Session, row: Any) -> AnnotationSchema:
     )
 
 
+def _ingest_job_to_row(entity: IngestJob) -> t.Base:
+    """Spelled out rather than left to ``_flat_mapping``, which dumps in python
+    mode and would hand a tuple of ``IngestFailure`` models to a ``JSON`` column.
+    """
+    return t.IngestJobRow(
+        id=entity.id,
+        source_id=entity.source_id,
+        state=entity.state,
+        error=entity.error,
+        batch_id=entity.batch_id,
+        batch_name=entity.batch_name,
+        processed=entity.processed,
+        total=entity.total,
+        failures=[failure.model_dump(mode="json") for failure in entity.failures],
+    )
+
+
+def _ingest_job_to_domain(_: Session, row: Any) -> IngestJob:
+    return IngestJob(
+        id=row.id,
+        source_id=row.source_id,
+        state=IngestState(row.state),
+        error=row.error,
+        batch_id=row.batch_id,
+        batch_name=row.batch_name,
+        processed=row.processed,
+        total=row.total,
+        failures=tuple(IngestFailure.model_validate(f) for f in row.failures),
+    )
+
+
 def _annotation_to_row(entity: Annotation) -> t.Base:
     return t.AnnotationRow(
         id=entity.id,
@@ -309,7 +342,6 @@ def _job_sync_children(session: Session, entity: AnnotationJob) -> None:
 
 WORKSPACES = _flat_mapping(Workspace, t.WorkspaceRow, None)
 PROJECTS = _flat_mapping(Project, t.ProjectRow, "workspace_id")
-INGEST_JOBS = _flat_mapping(IngestJob, t.IngestJobRow, "source_id")
 ASSETS = _flat_mapping(Asset, t.AssetRow, "project_id")
 TASK_GROUPS = _flat_mapping(TaskGroup, t.TaskGroupRow, "batch_id")
 DATASETS = _flat_mapping(Dataset, t.DatasetRow, "project_id")
@@ -327,6 +359,12 @@ def _job_sync_children(session: Session, entity: AnnotationJob) -> None:
     to_row=_annotation_to_row,
     to_domain=_annotation_to_domain,
 )
+INGEST_JOBS: EntityMapping[IngestJob] = EntityMapping(
+    row=t.IngestJobRow,
+    parent_column="source_id",
+    to_row=_ingest_job_to_row,
+    to_domain=_ingest_job_to_domain,
+)
 SOURCES: EntityMapping[Source] = EntityMapping(
     row=t.SourceRow,
     parent_column="project_id",

src/visionset/kernel/adapters/_tables.py

@@ -11,8 +11,9 @@
   from the asset side, which a JSON blob cannot serve.
 - Collections that are *immutable value objects* get JSON columns —
   ``annotation_schema.classes``, ``annotation.geometry``, ``annotation.attributes``,
-  ``release.split``. A schema version must rehydrate byte-identical, and nothing
-  ever queries a single ``LabelClass`` by name in SQL.
+  ``release.split``, ``source.video``, ``ingest_job.failures``. A schema version
+  must rehydrate byte-identical, and nothing ever queries a single
+  ``LabelClass`` by name — or a single failed file — in SQL.
 - A value object too large to belong in a row at all goes in the blob store, and
   the row keeps its hash — that is ``release.manifest_hash``. The line between
   the two is size and verifiability, not shape.
@@ -157,17 +158,23 @@ class SourceRow(Base):
 
 
 class IngestJobRow(Base):
-    """Ingestion runs. Rebuilt by migration 8, so column order is free here.
-
-    Rebuilt rather than altered because ``batch_id`` carries a foreign key, and
-    an ``ALTER TABLE ... ADD COLUMN`` cannot express one the way ``create_all``
-    does: SQLite spells an added key inline on the column, while a created table
-    spells it as a table constraint. The two texts differ, so the fresh-versus-
-    migrated test would fail — and dropping the key instead would leave a run
-    pointing at a batch somebody deleted. This table has no children and is
-    provably empty at that point (nothing wrote an ingest job before this
-    build), which is what makes the rebuild affordable; migration 8 counts
-    rather than assuming it.
+    """Ingestion runs. Rebuilt by migration 8; migration 9 then altered it.
+
+    Migration 8 rebuilt rather than altered because ``batch_id`` carries a
+    foreign key, and an ``ALTER TABLE ... ADD COLUMN`` cannot express one the way
+    ``create_all`` does: SQLite spells an added key inline on the column, while a
+    created table spells it as a table constraint. The two texts differ, so the
+    fresh-versus-migrated test would fail — and dropping the key instead would
+    leave a run pointing at a batch somebody deleted. This table had no children
+    and was provably empty at that point (nothing wrote an ingest job before that
+    build), which is what made the rebuild affordable; migration 8 counts rather
+    than assuming it.
+
+    **That exemption expired the moment this build started writing rows.** A
+    database stamped at 8 will never re-run migration 8, so anything added after
+    it reaches that database by ``ALTER TABLE`` after all — which is why
+    migration 9's four columns are declared **last, in the order it adds them**.
+    A rebuild is no longer available here either: these rows are legitimate.
     """
 
     __tablename__ = "ingest_job"
@@ -177,6 +184,7 @@ class IngestJobRow(Base):
         SaUuid, ForeignKey("source.id", ondelete="CASCADE"), index=True, nullable=False
     )
     state: Mapped[str] = mapped_column(String, nullable=False)
+    #: The fatal cause that stopped the run, as opposed to ``failures`` below.
     error: Mapped[str | None] = mapped_column(String, nullable=True)
     #: The batch this run materialized into. NULL until it reaches one — a run
     #: that dies during the decode never does, which is why this is nullable
@@ -185,6 +193,23 @@ class IngestJobRow(Base):
     batch_id: Mapped[UUID | None] = mapped_column(
         SaUuid, ForeignKey("batch.id", ondelete="SET NULL"), nullable=True
     )
+    #: The name a batch this run creates will take, so a resumed run lands where
+    #: the first attempt meant it to. Nullable, and honestly so: a row written
+    #: before migration 9 never recorded one.
+    batch_name: Mapped[str | None] = mapped_column(String, nullable=True)
+    #: Items read so far. Carries a ``server_default`` for the reason
+    #: ``AnnotationJobAssetRow.position`` does — SQLite refuses ``ADD COLUMN``
+    #: ``NOT NULL`` without a value for the rows already there — and ``0`` is
+    #: what a finished pre-#19 run in fact recorded: nothing.
+    processed: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
+    #: Items the source offered, or NULL when that is not knowable up front —
+    #: a directory can be listed, a clip cannot. See ``IngestJob.total``.
+    total: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    #: The per-file report: a list of ``IngestFailure``. JSON rather than a child
+    #: table because it is read whole and never queried by field, the rule
+    #: ``source.video`` follows. ``server_default`` on
+    #: ``AnnotationRow.attributes``' terms.
+    failures: Mapped[list[Any]] = mapped_column(JSON, nullable=False, server_default=text("'[]'"))
 
 
 class AssetRow(Base):