feat(kernel): ingest pipeline — hashing, blob dedup, asset origin, batch materialization (#20)

docs/README.md

@@ -9,6 +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 |
 | [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 |
@@ -18,4 +19,4 @@ contracts (kernel purity, headless annotator) are described there and enforced i
 | [releases.md](releases.md) | The immutable artifact: what a manifest is and is not, why two publishes agree byte for byte, hash verification, and the seeded split recipe |
 | [events.md](events.md) | Domain events: subscribing by type, why emission follows the commit, at-most-once delivery, and what an isolated subscriber failure does |
 | [persistence.md](persistence.md) | The metadata store: repositories, unit of work, table layout, migrations and `format_version` |
-| [examples.md](examples.md) | The runnable end-to-end example: the whole cycle in one pass, what it is built to demonstrate, and the one step that has no service yet |
+| [examples.md](examples.md) | The runnable end-to-end example: the whole cycle in one pass, and what it is built to demonstrate |

docs/events.md

@@ -94,7 +94,7 @@ custom encoder — which is what makes a webhook a subscriber rather than a rewr
 | `BatchCompleted` | `BatchService.complete` | `batch_id`, `project_id`, `asset_count` |
 | `AnnotationsWritten` | `AnnotationService.add` / `update` / `delete` | `job_id`, `batch_id`, `operation`, `asset_ids`, `annotation_ids` |
 | `ReleasePublished` | `ReleaseService.publish` | `release_id`, `dataset_id`, `project_id`, `tag`, `manifest_hash`, `schema_version`, `asset_count`, `annotation_count` |
-| `IngestCompleted` | nobody yet — M2 | `ingest_job_id`, `project_id`, `source_id`, `asset_count` |
+| `IngestCompleted` | `IngestService.ingest` | `ingest_job_id`, `project_id`, `source_id`, `asset_count` |
 
 `AnnotationsWritten` is one per **call**, not one per box: the three writes are all-or-nothing
 over a whole payload, so one call is one thing that happened. Its `asset_ids` are deduplicated —
@@ -105,10 +105,11 @@ out of the blob store without being handed it. It is also why publishing writes
 [dataset change-log](datasets.md) entry: the log records mutations of the trunk, and publishing
 mutates nothing in it. "A release happened" is an event.
 
-`IngestCompleted` is declared and emitted by nothing. Ingest is M2's; the vocabulary was settled
-in one pass so that a subscriber written today already compiles against the shape it will be
-handed, and a test asserts nothing in M1 emits it — so it cannot quietly acquire a caller before
-M2 wires one deliberately.
+`IngestCompleted` was declared in M1 and emitted by nothing until [ingest](ingest.md) wired it —
+the vocabulary was settled in one pass so a subscriber written then already compiled against the
+shape it would be handed, and a test held the line until the emitter arrived deliberately. Its
+`asset_count` is what the run **put in the batch**: assets it created plus assets whose content
+the project already held. A run that fails announces nothing at all.
 
 ### `name`, and why it is two types
 

docs/examples.md

@@ -21,8 +21,9 @@ as a plain script, which is the only way to prove it still works from a clean ch
 | Subscribe | `event_bus.subscribe(DomainEvent, ...)` — the catch-all, matched by type | [events.md](events.md) |
 | Project | `ProjectService.create` — the 1:1 dataset is created in the same transaction | [projects.md](projects.md) |
 | Schema | `SchemaService.create_version` — version 1 of the labeling contract | [schemas.md](schemas.md) |
-| Assets | six generated PNGs into the blob store, then the rows that name them | *nothing yet — see below* |
-| Batch | `BatchService.create`, then `approve(BySize(size=3))` → 2 jobs, schema pinned | [batches.md](batches.md) |
+| Source | six generated PNGs written to `incoming/`, registered as an image directory | [sources.md](sources.md) |
+| Assets | `IngestService.ingest` — hashed, stored once, and put in a draft batch | [ingest.md](ingest.md) |
+| Batch | `approve(BySize(size=3))` → 2 jobs, schema pinned | [batches.md](batches.md) |
 | Work | `JobService.start` / `next_pending` / `mark`, one asset deliberately skipped | [jobs.md](jobs.md) |
 | Labels | `AnnotationService.add` — a box, a polygon and a whole-frame tag per asset | [annotations.md](annotations.md) |
 | Trunk | `DatasetService.promote` — five assets, not six | [datasets.md](datasets.md) |
@@ -49,24 +50,27 @@ keys on `content_hash`, and the frames are generated deterministically from thei
 same pictures land in the same folds on every machine. The smoke test asserts exactly that —
 comparing folds by content hash, never by id.
 
-## The one place it reaches below a service
+## Every step goes through the service that owns it
 
-Creating an `Asset` has no door yet. `SourceService` (#18) has landed, but it registers *where*
-data comes from, not the assets themselves; the pipeline that hashes, deduplicates, extracts
-dimensions and materializes assets into a batch is #20, with `IngestJob` (#19) around it. Until
-#20 lands, `_add_assets` writes, by hand, the row that ingest will write:
+That sentence used to carry an exception. Creating an `Asset` had no door, so the example wrote
+the row by hand — through the same public port a service uses, commented as the one place it
+reached below one, and flagged in three files as something #20 would delete. It did:
 
 ```python
-hashes = [workspace.blob_store.put(BytesIO(frame_bytes(i))) for i in range(count)]
-with workspace.unit_of_work() as uow:
-    uow.assets.add(Asset(project_id=..., content_hash=..., uri=..., width=..., height=...))
+incoming = _write_frames(dest / "incoming", FRAME_COUNT)
+source = sources.register_images(project.id, incoming)
+ingested = ingest.ingest(source.id, batch_name="batch-001")
 ```
 
-This is the only step in the file that does not go through the service that owns its entity, it
-is commented as such where it lives, and it disappears when #20 lands. The blobs are written
-before the transaction opens on purpose: `BlobStore.put` is not transactional and a rollback
-cannot unwrite it — but a blob nothing points at is harmless (content-addressed, deduplicated,
-and never deleted), while a row pointing at bytes that were never stored would not be.
+The frames go to disk, the folder is registered as an origin, and [ingest](ingest.md) hashes
+them, stores the bytes once and puts them in a draft batch — which is what a real caller does with
+a real folder of photographs. `BatchService.create` disappeared from the example along with it:
+the ingest is what makes the batch now.
+
+The frames are still generated by the example's own six-line PNG encoder rather than by Pillow,
+which is a dependency these days. Their bytes are fixed, and that is the point: an asset's
+identity is the SHA-256 of its content and the split keys on content hash, so the same pictures
+land in the same folds on every machine.
 
 ## Why three classes for "two classes"
 

docs/ingest.md

@@ -0,0 +1,123 @@
+# Ingest
+
+Where a registered [source](sources.md) becomes rows. `IngestService` hashes every item, stores
+the bytes once, records what the decoder made of them, and puts the result in a draft
+[batch](batches.md) somebody can approve. Nothing else in the kernel creates an `Asset` —
+`examples/sdk_end_to_end.py` used to, and no longer does.
+
+```python
+source = sources.register_images(project.id, Path("~/dashcam/monday").expanduser())
+result = ingest.ingest(source.id, batch_name="monday")
+
+result.created  # assets new to this project
+result.deduplicated  # items whose bytes the project already held
+result.failures  # one line per item that could not be read at all
+```
+
+## One `ingest`, where registration has two methods
+
+`SourceService` split `register_images` from `register_video` because their arguments genuinely
+differ: a clip needs a decomposition rate and gets probed, a directory needs neither. Here they do
+not differ at all. The source already carries its kind, its path and its rate, so the caller
+passes one id and the service branches on `SourceKind`. A second entry point would ask callers to
+re-state what the source already knows.
+
+## Identity is content; origin is provenance
+
+An asset **is** its bytes: `content_hash` is the SHA-256 of the file, and the same bytes ingested
+twice are one asset over one blob. `uq_asset_project_content_hash` is the index under that rule.
+Per project, not global — two projects ingesting one photograph are two assets sharing one blob,
+which is exactly what makes `project_id` the asset's parent.
+
+`source_id`, `frame_index` and `frame_timestamp` are a different kind of fact. They record where
+the bytes were **first** seen, and a second sighting never rewrites them — the rule
+`Source.registered_at` already follows. One image in two registered folders keeps the first
+folder's path in its `uri` and the first source's id in its origin, because the alternative is an
+asset whose recorded origin depends on which ingest happened to run last.
+
+The deliberate consequence: an asset records **one** origin, and a duplicate across two sources
+loses the second. A join table is the honest upgrade if that information is ever wanted; it would
+be its own migration and is not needed by anything today.
+
+Re-running an ingest is therefore not an error and not a no-op worth avoiding: it creates nothing,
+reports every item as deduplicated, and is how a folder that grew by three files is caught up.
+
+## What the two paths do
+
+| | image directory | video |
+| --- | --- | --- |
+| what is read | every file at the **top level**, in filename order | one frame per extraction slot, at the rate the source records |
+| decoded by | `ImageProcessor.probe` — dimensions and format from the bytes | `VideoProcessor.frames` — ffmpeg, deterministic within one build |
+| `uri` | the file's absolute path | `/path/clip.mp4#frame=7` |
+| frame position | none | `frame_index` and `frame_timestamp` |
+| damage | one report line per file; the run carries on | the frames ffmpeg managed are kept, plus one report line |
+
+Subdirectories are stepped over and recorded nowhere. Recursion is not a per-run option but a
+question about what *the source is* — "the same source yields the same assets" — so it belongs to
+a future `register_images(..., recursive=True)` rather than here, where it would silently change
+what an already-registered source means. There is no suffix filter either: a `notes.txt` is
+reported as unsupported rather than skipped, because guessing which files an operator meant to
+offer is a policy the kernel would be inventing.
+
+**Frames are not re-probed.** `VideoProcessor` guarantees every frame is a complete image in
+`FRAME_FORMAT` at the dimensions `probe` reported, and that promise is asserted in the port's own
+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.
+
+## 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`.
+2. **No transaction.** Decode, hash and `BlobStore.put` every item.
+3. Write the asset rows, reusing whatever content the project already holds.
+4. Put them in the batch (through `BatchService`), then mark the job `completed`.
+
+Then, and only after the last block has exited, `IngestCompleted` goes on the bus — the rule every
+emitter in this kernel follows.
+
+Step 2 is outside a transaction because decoding is a Pillow pass over thousands of files or an
+out-of-process ffmpeg, and holding a write transaction open across either is how a single-writer
+SQLite store starts reporting "database is locked". The blob writes are out there too, before any
+row exists: `BlobStore.put` is not transactional and a rollback cannot unwrite it — but a blob
+nothing points at is harmless (content-addressed, shared, never deleted), while a row naming bytes
+that were never stored is not.
+
+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.
+
+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.
+
+## Failure splits by remedy, not by severity
+
+A file that is not an image, or one whose bytes will not decode, is *reported*: one
+`IngestFailure` carrying the item's name, the reason, and which of the two it was. The run carries
+on, because an operator with five thousand files needs the other four thousand nine hundred. The
+name and the reason are kept apart so a report renders as a table rather than as a list of
+sentences, and `IngestFailureKind` exists so it can be **grouped** — real data loss must not be
+buried under ordinary operator noise.
+
+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.
+
+## The target batch
+
+With no `batch_id`, the run creates a draft named `batch_name` or, failing that, after the
+source's own file or folder. With one, that batch must still be a draft — checked **before**
+anything is decoded, because finding out afterwards would mean finding out after the work.
+
+Membership is everything the run ingested, deduplicated assets included: a duplicate is not new
+data, but it is part of what the run was asked to gather. Order is ingest order, which is filename
+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.

docs/media.md

@@ -333,12 +333,12 @@ kernel — FastAPI, Typer, MCP, uvicorn — not third-party libraries, and ffmpe
 
 ## What is deliberately not here yet
 
-- **No `Asset` field.** `ImageMetadata` and `VideoMetadata` are returned, not stored; putting
-  `format` and origin on the asset row belongs with the ingest pipeline.
-- **No blob write.** `thumbnail()` and `frames()` hand back bytes. Storing them
-  content-addressed, recording a `thumbnail_hash` and writing a frame's `index`/`timestamp` onto
-  an asset are the ingest and thumbnail-cache tasks.
-
-`Source` used to be on that list and no longer is: registering a clip records its original rate
-and the decomposition parameters chosen for it, built on `VideoMetadata` exactly as anticipated.
-See [sources.md](sources.md).
+- **No thumbnail write.** `thumbnail()` hands back bytes. Storing them content-addressed and
+  recording an `asset.thumbnail_hash` is the thumbnail-cache task, for the M5 gallery.
+
+Two things used to be on that list and no longer are. `Source` came off it with registration,
+which records a clip's original rate and the decomposition parameters chosen for it, built on
+`VideoMetadata` exactly as anticipated — see [sources.md](sources.md). The `Asset` fields came off
+it with [ingest](ingest.md): what a probe reported is now stored as `asset.format`, and a frame's
+`index`/`timestamp` land on the asset as `frame_index`/`frame_timestamp` beside the source it was
+cut from. Both ports are called from exactly one place, and that is where.