From 511554c78e715b3c877abea1e69998b485137a95 Mon Sep 17 00:00:00 2001 From: Armando Anaya Date: Mon, 27 Jul 2026 03:58:22 -0700 Subject: [PATCH] =?UTF-8?q?feat(kernel):=20SourceService=20=E2=80=94=20sou?= =?UTF-8?q?rce=20registration=20with=20provenance=20(#18)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the M1 `Source` placeholder with a real record of where a project's raw data came from, and adds `SourceService` as the one door to it. - `domain/source.py`: `SourceKind` (image_directory | video), frozen `VideoProvenance` wrapping the port's `VideoMetadata` plus the chosen `extraction_fps`, and a `Source` carrying `path`, tz-aware `registered_at` and opaque `capture_params`. `uri` is renamed to `path` — it holds `canonical_path()`, an absolute resolved local path. - `SourceService.register_images` / `register_video`: two methods because a clip needs a rate and a probe and a directory needs neither. The probe runs before the transaction opens. Registration is idempotent on `(kind, path, extraction_fps)`; differing capture params or a replaced clip refresh the matched source in place rather than forking it. - Migration 7 rebuilds `source` rather than altering it: `registered_at` is NOT NULL with no honest default, and a pre-#18 row's `kind='local_folder'` is not a value `SourceKind` has. It counts `ingest_job` as well as `source`, because `DROP TABLE` under `PRAGMA foreign_keys = ON` cascades silently. FORMAT_VERSION 6 -> 7. - `SOURCES` becomes a hand-written mapper pair (timestamp + nested JSON). - New `docs/sources.md`; persistence, media, examples docs brought current. No new domain event, no VERSION bump, no openapi drift. Closes #18 --- docs/README.md | 1 + docs/examples.md | 7 +- docs/media.md | 6 +- docs/persistence.md | 38 +- docs/sources.md | 126 ++++++ examples/sdk_end_to_end.py | 9 +- src/visionset/kernel/__init__.py | 2 + src/visionset/kernel/adapters/_mappers.py | 44 ++- src/visionset/kernel/adapters/_tables.py | 17 +- src/visionset/kernel/adapters/migrations.py | 81 +++- src/visionset/kernel/domain/__init__.py | 10 +- src/visionset/kernel/domain/source.py | 160 +++++++- src/visionset/kernel/errors.py | 14 + src/visionset/kernel/services/__init__.py | 2 + .../kernel/services/source_service.py | 233 +++++++++++ tests/kernel/test_metadata_store.py | 61 ++- tests/kernel/test_migrations.py | 87 +++++ tests/kernel/test_project_service.py | 5 +- tests/kernel/test_source_service.py | 363 ++++++++++++++++++ 19 files changed, 1222 insertions(+), 44 deletions(-) create mode 100644 docs/sources.md create mode 100644 src/visionset/kernel/services/source_service.py create mode 100644 tests/kernel/test_source_service.py diff --git a/docs/README.md b/docs/README.md index 82364ccb..673e2b1f 100644 --- a/docs/README.md +++ b/docs/README.md @@ -8,6 +8,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 | | [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 | diff --git a/docs/examples.md b/docs/examples.md index 5cd8f233..be746e7c 100644 --- a/docs/examples.md +++ b/docs/examples.md @@ -51,9 +51,10 @@ comparing folds by content hash, never by id. ## The one place it reaches below a service -Creating an `Asset` has no door yet. Ingest is M2 — `SourceService` (#18), `IngestJob` (#19), and -the pipeline that hashes, deduplicates, extracts dimensions and materializes assets into a batch -(#20) — so `_add_assets` writes, by hand, the row that ingest will write: +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: ```python hashes = [workspace.blob_store.put(BytesIO(frame_bytes(i))) for i in range(count)] diff --git a/docs/media.md b/docs/media.md index c498a253..64891d18 100644 --- a/docs/media.md +++ b/docs/media.md @@ -338,5 +338,7 @@ kernel — FastAPI, Typer, MCP, uvicorn — not third-party libraries, and ffmpe - **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. -- **No `Source`.** Nothing yet records that a clip was registered, at what original rate, or with - what decomposition parameters. `VideoMetadata.fps` is what that record will be built from. + +`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). diff --git a/docs/persistence.md b/docs/persistence.md index 47ff4871..9b3f6ef7 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -95,8 +95,9 @@ MIGRATIONS: list[Migration] = [ Migration(version=4, name="annotation_job_asset_position", upgrade=...), Migration(version=5, name="annotation_attributes", upgrade=...), Migration(version=6, name="release_manifest_pointer", upgrade=...), + Migration(version=7, name="source_provenance", upgrade=...), ] -FORMAT_VERSION: int = MIGRATIONS[-1].version # 6 +FORMAT_VERSION: int = MIGRATIONS[-1].version # 7 ``` `initialize()` reads the version stamped in `_visionset_meta` and runs whatever is @@ -138,24 +139,35 @@ one, where `batch_asset.position`, which was there from migration 001, does not. And a column arriving by `ALTER` must be declared **last** on its row class, because SQLite appends it: `batch.schema_version`, `annotation_job_asset.position` and `annotation.attributes` -all sit at the end of their tables for that reason alone. Declared anywhere else, the +all sit at the end of their tables for that reason alone. `source` is exempt — migration 007 +rebuilds it from `_tables` rather than altering it, so both paths run the same `CREATE TABLE` and +the rule has nothing to bite on. 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 006 is the one that drops a table**, and the bar it had to clear is worth writing -down. `release` gained three `NOT NULL` columns with no honest default — an `ALTER` would have -baked `manifest_hash DEFAULT ''` into every fresh database forever — and, decisively, a pre-#12 -release row carries its manifest as a JSON column with *no blob behind it*, so there is no value -`manifest_hash` could be given that `verify` would ever accept. Adding the columns would have -manufactured rows that are broken by construction. It is idempotent the same way 003–005 are (an -inspector check for `manifest_hash`), and the emptiness it relies on is **checked** rather than -argued: a workspace that somehow holds a release row raises `WorkspaceCorrupt` instead of being -quietly emptied. A migration that would lose real data does not clear this bar. +**Migrations 006 and 007 drop a table**, and the bar they had to clear is worth writing down. +`release` gained three `NOT NULL` columns with no honest default — an `ALTER` would have baked +`manifest_hash DEFAULT ''` into every fresh database forever — and, decisively, a pre-#12 release +row carries its manifest as a JSON column with *no blob behind it*, so there is no value +`manifest_hash` could be given that `verify` would ever accept. `source` is the same shape of +argument twice over: `registered_at` is `NOT NULL` with no honest default, and a pre-#18 row's +`kind` reads `'local_folder'`, which is not a value `SourceKind` has — those rows would come back +as validation errors rather than as sources. Adding the columns would have manufactured rows that +are broken by construction. Both are idempotent the same way 003–005 are (an inspector check for +the new column), and the emptiness each relies on is **checked** rather than argued: a workspace +that somehow holds such a row raises `WorkspaceCorrupt` instead of being quietly emptied. A +migration that would lose real data does not clear this bar. + +Migration 007 carries one extra obligation that 006 did not. `ingest_job.source_id` is +`ON DELETE CASCADE`, and this store sets `PRAGMA foreign_keys = ON` for every connection — so +`DROP TABLE source` runs an implicit `DELETE FROM source` that takes the ingest jobs with it, +**silently, without raising**. A rebuild of a table with children has to count the children too. +`release` had none, which is why the precedent alone was not enough. 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. -Migration 006 is the one place that undo cannot borrow its DDL from `_tables`, because `_tables` -no longer describes the shape it is restoring. +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. `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 — diff --git a/docs/sources.md b/docs/sources.md new file mode 100644 index 00000000..87eaa806 --- /dev/null +++ b/docs/sources.md @@ -0,0 +1,126 @@ +# Sources + +A **source** is the record that raw data was offered to a project: a directory of stills, or a +video file. It is not annotatable and holds no pixels. Assets are what an ingest *materializes* +from a source; the source is the receipt that says where they came from. + +`SourceService` is the one door to a `Source`. Nothing else writes `uow.sources`. + +## Two registration methods, not one + +```python +sources = SourceService(workspace) + +stills = sources.register_images(project.id, Path("~/captures/2026-07").expanduser()) +clip = sources.register_video(project.id, Path("~/captures/drive.mp4"), extraction_fps=5.0) +``` + +There is no `register(kind=...)`, because the arguments genuinely differ. A clip needs a +decomposition rate and gets probed; a directory needs neither and is not walked. That is the same +argument that made `ImageProcessor` and `VideoProcessor` two protocols instead of one — see +[media.md](media.md). A single entry point would have to accept parameters that are meaningless +for half its callers. + +`register_images` checks that the directory exists and is a directory, and stops there. What is +*in* it is read at ingest, because a count taken at registration would be stale by the time +anything used it. + +`register_video` probes the file through the workspace's `VideoProcessor` and stores the answer. +The probe runs **before** the transaction opens: it is an out-of-process decoder, and holding a +write transaction open across a subprocess is how a single-writer SQLite store ends up reporting +"database is locked". + +## What a source records + +| Field | Meaning | +| --- | --- | +| `kind` | `image_directory` or `video` — a `SourceKind` | +| `path` | the canonical absolute path of the origin | +| `registered_at` | timezone-aware UTC, the **first** registration | +| `capture_params` | opaque operator-supplied provenance; nothing branches on it | +| `video` | a `VideoProvenance`, present exactly when `kind` is `video` | + +`VideoProvenance` is the port's own `VideoMetadata` — original fps, duration, displayed +dimensions, codec — plus the `extraction_fps` a decomposition will run at. The probe result is +kept whole rather than re-spelled field by field, because `metadata.fps` is the rate the file was +*shot* at and `extraction_fps` is the rate we chose to *cut* it at, and re-declaring the first +beside the second is how the two come to be confused. + +The `video`/`kind` pairing is an invariant, enforced on construction **and** on assignment — +`Source` is the only model in the domain with `validate_assignment` on, because a +`model_validator` does not re-run when you assign to a field. Reading it goes through +`source.require_video()`, which raises `WorkspaceCorrupt` rather than handing back a `None` that +every caller would have to assert away. + +## Decomposition parameters live on the source, not on the job + +A source can be ingested more than once, and the promise is that the same source yields the same +assets. That promise only means something if the parameters are part of what "the same source" +*is* — put them on the ingest job and two runs of one source could legitimately disagree, leaving +idempotency with nothing to be measured against. + +The consequence is deliberate: **one clip registered at 1 fps and again at 5 fps is two sources +over one file**, not one source with a history. + +## Registration is idempotent + +The match key is `(kind, path, extraction_fps)`. Registering the same origin twice returns the +same `Source` rather than a second one, so that once ingest gives `asset.source_id` a target, +"which source did this asset come from?" has one answer. + +Two things the key deliberately leaves out: + +- **`capture_params`.** Fragmenting one directory into two sources because an operator typed a + different lens note would defeat the point. Differing params are written onto the matched + source instead. +- **The probed `VideoMetadata`.** A clip replaced at a known path is still that path's source, so + its recorded provenance is *refreshed in place* rather than left describing a file that is + gone. `registered_at` is never rewritten — it is the first registration, not the last. + +That second rule has a corollary worth knowing: re-registering an already-known clip still needs +ffmpeg, because the fresh probe is what keeps the record honest. + +### The gap this leaves, and when to close it + +`docs/persistence.md` says a rule with no backstop is a wish, and every other uniqueness rule in +this store has a unique index behind it. This one does not. Two concurrent registrations of one +folder can both pass the pre-check and both insert. + +That is tolerated today because no row references a source, so a duplicate is inert. It stops +being tolerable when ingest gives `asset.source_id` a target and the winner of a race starts +deciding an asset's recorded origin — **that is when this needs an index under it**. It sits +alongside the store's other known concurrency gap, the untranslated `OperationalError`. + +## Paths are canonicalized once + +`canonical_path` is `str(Path.resolve(strict=True))`: absolute, symlinks followed, so `./data`, +`../project/data` and `/abs/data` are one source. Two things it does not do: + +- **It does not normalize case.** On a case-insensitive filesystem — macOS by default, Windows + always — `/Data` and `/data` are one directory and would register as two sources. Lower-casing + would be wrong on Linux, where they are genuinely two. +- **It does not look at the content.** Two hard links to one inode read as two origins. What the + bytes *are* is asked at ingest, where the answer is a content hash. + +`strict=True` means an origin that is not on disk is a `FileNotFoundError`, and a file offered +where a directory was wanted is a `NotADirectoryError`. Both are about the machine rather than +the workspace, so both stay outside the `VisionSetError` tree — the same line +`MediaToolUnavailable` sits on. + +## Registration is not a validation pass + +`register_video` probes; it does not decode. A clip whose tail has been truncated still has a +readable header, so it registers successfully and records the duration the intact file would have +had. The damage surfaces when frames are actually extracted. Anything downstream that treats a +successful registration as proof the file will decode is wrong. + +## What is deliberately not here yet + +- **No delete.** A source disappears with its project's cascade and no sooner. Nothing yet + references one, so there is no orphan to reason about; when ingest gives `asset.source_id` a + target, deletion becomes a real question with a real answer. +- **No event.** Registering a source announces nothing. `IngestCompleted` is the event this area + will emit, and the ingest pipeline owns it. +- **No remote kinds.** `SourceKind` has two members and grows by a deliberate kernel change with + a service method behind it — see the enum's own docstring for why it is an enum where + `DatasetChange.operation` is a plain `str`. diff --git a/examples/sdk_end_to_end.py b/examples/sdk_end_to_end.py index 71483bee..5d31947d 100644 --- a/examples/sdk_end_to_end.py +++ b/examples/sdk_end_to_end.py @@ -14,10 +14,11 @@ library to lean on (Pillow arrives with the media processor in M2, #16). **One call in here reaches below a service, on purpose.** Creating an ``Asset`` -has no door yet — ingest is M2 (#18 SourceService, #19 IngestJob, #20 the -pipeline that hashes, deduplicates and materializes into a batch). Until #20 -lands, ``_add_assets`` writes the row that ingest would write, through the same -public port a service uses. Every other step below goes through the service that +has no door yet. #18's ``SourceService`` registers *where* data comes from, not +the assets themselves; the pipeline that hashes, deduplicates and materializes +into a batch is #20, with #19's ``IngestJob`` around it. Until #20 lands, +``_add_assets`` writes the row that ingest would write, through the same public +port a service uses. Every other step below goes through the service that owns it, which is how the rest of the SDK is meant to be used. """ diff --git a/src/visionset/kernel/__init__.py b/src/visionset/kernel/__init__.py index 947db614..b1362e6f 100644 --- a/src/visionset/kernel/__init__.py +++ b/src/visionset/kernel/__init__.py @@ -45,6 +45,7 @@ SchemaChangeWouldOrphan, SchemaNotFound, SchemaVersionConflict, + SourceNotFound, UnknownAttribute, UnserializableManifest, UnsupportedGeometry, @@ -95,6 +96,7 @@ "SchemaChangeWouldOrphan", "SchemaNotFound", "SchemaVersionConflict", + "SourceNotFound", "UnknownAttribute", "UnserializableManifest", "UnsupportedGeometry", diff --git a/src/visionset/kernel/adapters/_mappers.py b/src/visionset/kernel/adapters/_mappers.py index e3614f2c..5fe0fa8c 100644 --- a/src/visionset/kernel/adapters/_mappers.py +++ b/src/visionset/kernel/adapters/_mappers.py @@ -7,14 +7,15 @@ fourteen times against fourteen tables. Most entities are flat — every field is a column — and share -``_flat_mapping``. The six that are not say so explicitly: +``_flat_mapping``. The seven that are not say so explicitly: - ``AnnotationSchema`` and ``Annotation`` 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`` and ``Release`` encode a timezone-aware timestamp, which a - ``String`` column must be handed as text rather than as a ``datetime``. +- ``DatasetChange``, ``Release`` and ``Source`` encode a timezone-aware + timestamp, which a ``String`` column must be handed as text rather than as a + ``datetime``. ``Source`` also carries a nested ``VideoProvenance`` as JSON. """ from __future__ import annotations @@ -46,8 +47,10 @@ Project, Release, Source, + SourceKind, SplitRecipe, TaskGroup, + VideoProvenance, Workspace, ) @@ -172,6 +175,34 @@ def _change_to_domain(_: Session, row: Any) -> DatasetChange: ) +def _source_to_row(entity: Source) -> t.Base: + return t.SourceRow( + id=entity.id, + project_id=entity.project_id, + kind=entity.kind, + path=entity.path, + # Spelled out for the reason ``_release_to_row`` is: ``_flat_mapping`` + # dumps in python mode and would hand a ``datetime`` to a ``String`` + # column, which sqlite3 accepts through a deprecated adapter and writes + # in a second timestamp format. + registered_at=entity.registered_at.isoformat(), + capture_params=dict(entity.capture_params), + video=None if entity.video is None else entity.video.model_dump(mode="json"), + ) + + +def _source_to_domain(_: Session, row: Any) -> Source: + return Source( + id=row.id, + project_id=row.project_id, + kind=SourceKind(row.kind), + path=row.path, + registered_at=datetime.fromisoformat(row.registered_at), + capture_params=row.capture_params, + video=None if row.video is None else VideoProvenance.model_validate(row.video), + ) + + def _release_to_row(entity: Release) -> t.Base: return t.ReleaseRow( id=entity.id, @@ -278,7 +309,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") -SOURCES = _flat_mapping(Source, t.SourceRow, "project_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") @@ -297,6 +327,12 @@ def _job_sync_children(session: Session, entity: AnnotationJob) -> None: to_row=_annotation_to_row, to_domain=_annotation_to_domain, ) +SOURCES: EntityMapping[Source] = EntityMapping( + row=t.SourceRow, + parent_column="project_id", + to_row=_source_to_row, + to_domain=_source_to_domain, +) RELEASES: EntityMapping[Release] = EntityMapping( row=t.ReleaseRow, parent_column="dataset_id", diff --git a/src/visionset/kernel/adapters/_tables.py b/src/visionset/kernel/adapters/_tables.py index db5966e3..33de1420 100644 --- a/src/visionset/kernel/adapters/_tables.py +++ b/src/visionset/kernel/adapters/_tables.py @@ -98,6 +98,15 @@ class AnnotationSchemaRow(Base): class SourceRow(Base): + """Registered origins. Rebuilt by migration 7, so column order is free here. + + Every other table with a migrated column declares it last, because SQLite's + ``ALTER TABLE ... ADD COLUMN`` appends and the two creation paths would + otherwise emit different DDL. Migration 7 drops and re-creates this table + from this class instead, so both paths run the same ``CREATE TABLE`` and the + rule has nothing to bite on. + """ + __tablename__ = "source" id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) @@ -105,7 +114,13 @@ class SourceRow(Base): SaUuid, ForeignKey("project.id", ondelete="CASCADE"), index=True, nullable=False ) kind: Mapped[str] = mapped_column(String, nullable=False) - uri: Mapped[str] = mapped_column(String, nullable=False) + #: The canonical absolute path of the origin — see ``domain.canonical_path``. + path: Mapped[str] = mapped_column(String, nullable=False) + #: ISO-8601 with offset, never SQLite ``DATETIME``. See the module docstring. + registered_at: Mapped[str] = mapped_column(String, nullable=False) + capture_params: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + #: A ``VideoProvenance``, or NULL for anything that is not a clip. + video: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) class IngestJobRow(Base): diff --git a/src/visionset/kernel/adapters/migrations.py b/src/visionset/kernel/adapters/migrations.py index 5ea3aaf6..9d379827 100644 --- a/src/visionset/kernel/adapters/migrations.py +++ b/src/visionset/kernel/adapters/migrations.py @@ -21,11 +21,15 @@ migration (rather than repeating the DDL) is what keeps the two paths from drifting; ``tests/kernel/test_migrations.py`` proves they agree. -**A migration may drop, but it has to earn it.** Migration 6 rebuilds the -``release`` table instead of altering it, because the columns it needed could -not be added honestly and because the rows it discards could not have been made -correct. It checks that there are none rather than taking the argument on trust. -That is the bar; a migration that would lose real data does not clear it. +**A migration may drop, but it has to earn it.** Migrations 6 and 7 rebuild the +``release`` and ``source`` tables instead of altering them, because the columns +they needed could not be added honestly and because the rows they discard could +not have been made correct. Both check that there are none rather than taking +the argument on trust. That is the bar; a migration that would lose real data +does not clear it — and note that under ``PRAGMA foreign_keys = ON``, which this +store sets on every connection, ``DROP TABLE`` runs an implicit ``DELETE`` that +cascades to children *silently*. A rebuild of a table with children has to count +those too. Migration 7 does; migration 6 had none to count. Migrations only run forward. A workspace stamped ahead of this build is rejected (``WorkspaceFormatTooNew``) rather than silently downgraded. @@ -47,6 +51,7 @@ Base, BatchRow, ReleaseRow, + SourceRow, ) from visionset.kernel.errors import WorkspaceCorrupt @@ -125,9 +130,9 @@ def _add_annotation_attributes(connection: Connection) -> None: def _repoint_release_at_its_manifest_blob(connection: Connection) -> None: """Rebuild ``release`` around a manifest that lives in the blob store. - The only migration here that drops a table, and the only one that could not - have been an ``ALTER``. Three of the columns it adds are ``NOT NULL`` with no - honest default — SQLite refuses such a column without one, so the ``ALTER`` + The first migration here that drops a table (migration 7 is the other), and + it could not have been an ``ALTER``. Three of the columns it adds are + ``NOT NULL`` with no honest default — SQLite refuses such a column without one, so the ``ALTER`` route would have baked ``manifest_hash DEFAULT ''`` and two more fictions into every fresh database forever. And decisively: a pre-#12 row carries its manifest as a JSON column with *no blob behind it*, so there is no value @@ -162,6 +167,61 @@ def _repoint_release_at_its_manifest_blob(connection: Connection) -> None: table.create(connection) +def _rebuild_source_with_provenance(connection: Connection) -> None: + """Rebuild ``source`` around real provenance: where, when, and at what rate. + + Before this, a source was ``(kind, uri)`` and nothing else — a placeholder + with no service behind it. It gains a registration timestamp, the capture + parameters an operator supplied, and for a clip the probe result plus the + decomposition rate; ``uri`` becomes ``path``, which is what it always held. + + A rebuild rather than five ``ALTER``s, on migration 6's terms. + ``registered_at`` is ``NOT NULL`` with no honest default — SQLite refuses + such a column without one, so the ``ALTER`` route would bake a fictional + epoch into every fresh database forever. And decisively: ``kind`` on a + pre-#18 row reads ``'local_folder'``, which is not a value ``SourceKind`` + has, so those rows would come back as validation errors rather than as + sources. Adding the columns would manufacture rows that are broken by + construction. + + Idempotent the way migrations 3 to 6 are: the inspector check *is* the + ``checkfirst``, because migration 1 is ``create_all`` of current metadata. + That check comes first so the fresh path never reaches the counts below. + + **Two counts, not one.** ``ingest_job.source_id`` is ``ON DELETE CASCADE`` + and the store sets ``PRAGMA foreign_keys = ON`` on every connection, so + ``DROP TABLE source`` performs an implicit ``DELETE FROM source`` that takes + the ingest jobs with it — silently, without an error. Migration 6 dropped a + table with no children and so never met this. Counting only ``source`` would + let a workspace with orphan-parented jobs pass the check and lose them. + + Nothing could write either row before ``SourceService`` existed, but "nothing + could" is a claim about a build, not about a file on disk — so a workspace + that somehow holds one is refused rather than quietly emptied. + """ + # ``__table__`` is declared as the general ``FromClause``; for a mapped class + # it is always the ``Table``, which is what ``drop``/``create`` need. + table = cast(Table, SourceRow.__table__) + stored = {existing["name"] for existing in inspect(connection).get_columns(table.name)} + if "registered_at" in stored: + return + # Raw text: the table still has its pre-#18 shape here, so the mapped columns + # in ``_tables`` no longer describe the thing being counted. + counts = { + name: connection.execute(text(f"SELECT count(*) FROM {name}")).scalar_one() + for name in ("source", "ingest_job") + } + if any(counts.values()): + raise WorkspaceCorrupt( + "this workspace holds source rows written before SourceService existed " + f"({counts['source']} sources, {counts['ingest_job']} ingest jobs). They record no " + "registration date and their 'local_folder' kind no longer exists, so there is " + "nothing to migrate them to; register the sources again instead." + ) + table.drop(connection) + table.create(connection) + + MIGRATIONS: list[Migration] = [ Migration(version=1, name="initial_schema", upgrade=_create_initial_schema), Migration( @@ -189,6 +249,11 @@ def _repoint_release_at_its_manifest_blob(connection: Connection) -> None: name="release_manifest_pointer", upgrade=_repoint_release_at_its_manifest_blob, ), + Migration( + version=7, + name="source_provenance", + upgrade=_rebuild_source_with_provenance, + ), ] FORMAT_VERSION: int = MIGRATIONS[-1].version diff --git a/src/visionset/kernel/domain/__init__.py b/src/visionset/kernel/domain/__init__.py index ba752e0d..eb5f8bba 100644 --- a/src/visionset/kernel/domain/__init__.py +++ b/src/visionset/kernel/domain/__init__.py @@ -74,7 +74,12 @@ SchemaDiff, diff_classes, ) -from visionset.kernel.domain.source import Source +from visionset.kernel.domain.source import ( + Source, + SourceKind, + VideoProvenance, + canonical_path, +) from visionset.kernel.domain.task import ( ASSET_PROGRESS_TRANSITIONS, JOB_TRANSITIONS, @@ -142,14 +147,17 @@ "SchemaDiff", "SingleJob", "Source", + "SourceKind", "SplitAssignment", "SplitRecipe", "TaskGroup", "VideoFrame", "VideoMetadata", + "VideoProvenance", "Workspace", "assign_split", "canonical_bytes", + "canonical_path", "diff_classes", "normalize_name", "partition_assets", diff --git a/src/visionset/kernel/domain/source.py b/src/visionset/kernel/domain/source.py index 711622aa..e575f311 100644 --- a/src/visionset/kernel/domain/source.py +++ b/src/visionset/kernel/domain/source.py @@ -1,16 +1,164 @@ -# usage: from visionset.kernel.domain import Source +# usage: from visionset.kernel.domain import Source, SourceKind, VideoProvenance +"""Where a project's raw data came from, and what we know about it. + +A :class:`Source` is the record that some bytes were *offered* to a project. It +is not annotatable and holds no pixels: it names an origin on disk, says when it +was registered, and — for a clip — carries what ``VideoProcessor.probe`` read off +it plus the rate a decomposition will run at. Assets are what an ingest +*materializes* from it; this is the receipt. + +**Decomposition parameters live here, not on the ingest job.** A source can be +ingested more than once, and the promise is that the same source yields the same +assets. That promise only means something if the parameters are part of what +"the same source" *is* — put them on the job and two runs of one source could +legitimately disagree, leaving idempotency with nothing to be measured against. +The consequence is deliberate: one clip registered at 1 fps and again at 5 fps is +two sources over one file, not one source with a history. + +**Paths are canonicalized once**, by :func:`canonical_path`, so ``./data`` and +``/abs/data`` are one source rather than two. See that function for what +canonicalization does and does not promise. +""" + from __future__ import annotations -from typing import Literal +from datetime import UTC, datetime +from enum import StrEnum +from pathlib import Path from uuid import UUID, uuid4 -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +from visionset.kernel.domain.media import VideoMetadata +from visionset.kernel.errors import WorkspaceCorrupt + + +class SourceKind(StrEnum): + """The shapes of raw input VisionSet accepts. + + An enum, where ``DatasetChange.operation`` and ``VideoMetadata.codec`` are + plain ``str``. That doctrine turns on one question — *can something outside + this build write the value?* A change-log entry outlives the release that + wrote it and a codec name is whatever ffmpeg decides to call it, so both have + to stay readable when they name something this build never heard of. + + Neither applies here. ``SourceService`` is the only door to a ``Source``, so + no foreign writer exists; the kernel **branches** on this value, in the two + registration methods and in the invariant tying :attr:`Source.video` to + :attr:`SourceKind.VIDEO`, and a branch on a magic string is the shape this + codebase replaces with a table; and the set grows by a deliberate kernel + change with a service method behind it. That is ``ImageFormat`` / + ``BatchState`` / ``IngestState`` territory, and it costs persistence nothing + — a ``StrEnum`` member *is* a ``str``. + """ + + IMAGE_DIRECTORY = "image_directory" + VIDEO = "video" + + +def canonical_path(path: Path) -> str: + """The one spelling of an origin, so two names for it are one source. + + ``resolve()`` makes the path absolute and follows symlinks, which is what + makes ``./data``, ``../project/data`` and ``/abs/data`` agree. Two things it + deliberately does not do: + + - **It does not normalize case.** On a case-insensitive filesystem — macOS by + default, Windows always — ``/Data`` and ``/data`` are one directory and + would register as two sources. Lower-casing here would be wrong on Linux, + where they are genuinely two. + - **It does not look at the content.** Two hard links to one inode read as two + origins. What the bytes *are* is asked at ingest, where the answer is a + content hash. + + Raises: + FileNotFoundError: nothing exists at that path. An origin that is not + there is a provenance nobody can ever check. + """ + return str(path.resolve(strict=True)) + + +class VideoProvenance(BaseModel): + """What a clip was, and how we chose to cut it. + + :attr:`metadata` is ``VideoProcessor.probe``'s answer, kept whole rather than + re-spelled field by field: ``fps`` there is the *original* rate the file was + shot at, which is provenance, and re-declaring it beside + :attr:`extraction_fps` is how the two come to be confused. Note that + video-derived asset identity is reproducible within one ffmpeg build and not + across builds — see ``ports/video_processor.py`` — so these numbers describe + the file, not a promise about what a later re-ingest will produce. + + Frozen, like every other value in the domain that is a pure function of some + bytes and a choice. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + metadata: VideoMetadata + extraction_fps: float = Field(gt=0) class Source(BaseModel): - """Where assets come from. Only local folders today; typed to extend.""" + """One registered origin of raw data for a project. + + ``validate_assignment`` is on, and this is the only model in the domain that + turns it on. The reason is the cross-field rule below: a + ``model_validator(mode="after")`` runs at construction and **not** on + attribute assignment, so without it ``source.kind = IMAGE_DIRECTORY`` would + leave a populated :attr:`video` behind and the mapper would write a video + provenance blob onto an image-directory row. Every other mutable model here + validates field by field, where assignment is already covered. + + :attr:`registered_at` is timezone-aware UTC, the convention for every + timestamp in the domain, and it records the **first** registration: + re-registering a known origin refreshes what was probed, never this. + + :attr:`capture_params` is opaque operator-supplied provenance — lens, rig, + site, whatever the person running the ingest wants on the record. Nothing in + the kernel branches on it and nothing validates it, which is exactly why the + values are ``str``: a typed value would imply someone was checking. + """ + + model_config = ConfigDict(validate_assignment=True) id: UUID = Field(default_factory=uuid4) project_id: UUID - kind: Literal["local_folder"] = "local_folder" - uri: str + kind: SourceKind + path: str + registered_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + capture_params: dict[str, str] = Field(default_factory=dict) + video: VideoProvenance | None = None + + @field_validator("registered_at") + @classmethod + def _registered_at_is_timezone_aware(cls, value: datetime) -> datetime: + if value.tzinfo is None: + raise ValueError("registered_at must be timezone-aware (UTC)") + return value.astimezone(UTC) + + @model_validator(mode="after") + def _video_provenance_matches_the_kind(self) -> Source: + if (self.video is not None) != (self.kind is SourceKind.VIDEO): + carry = "carry" if self.kind is SourceKind.VIDEO else "not carry" + raise ValueError(f"a {self.kind.value} source must {carry} video provenance") + return self + + def require_video(self) -> VideoProvenance: + """The clip's provenance, or refuse because this is not a clip. + + mypy cannot see the validator above, so ``if source.kind is VIDEO`` never + narrows ``VideoProvenance | None`` and every caller would grow its own + ``assert``. One spelling of the rule instead — the reason + ``ProjectService.require_dataset`` exists in the same shape. + + Raises: + WorkspaceCorrupt: this source carries no video provenance. For a + clip that means the invariant failed on disk; for anything else + it means the caller asked the wrong question of the wrong row. + """ + if self.video is None: + raise WorkspaceCorrupt( + f"source {self.id} is a {self.kind.value} and has no video provenance" + ) + return self.video diff --git a/src/visionset/kernel/errors.py b/src/visionset/kernel/errors.py index 1d2c7fd8..d170255f 100644 --- a/src/visionset/kernel/errors.py +++ b/src/visionset/kernel/errors.py @@ -298,6 +298,20 @@ class AssetNotFound(VisionSetError): """ +class SourceNotFound(VisionSetError): + """A source id does not belong to the project or workspace it was used in. + + Same rule as every other cross-scope reference in the kernel: a source in a + different project reads as missing, not as forbidden. + + Note what this is *not*. A path that does not exist on disk is a + ``FileNotFoundError`` and a path that is a file where a directory was wanted + is a ``NotADirectoryError`` — both are about the machine, not about the + workspace, and both stay outside the ``VisionSetError`` tree for the reason + ``MediaToolUnavailable`` sits outside ``MediaError``. + """ + + class SchemaVersionConflict(VisionSetError): """Two writers raced for the same next version number, and this one lost. diff --git a/src/visionset/kernel/services/__init__.py b/src/visionset/kernel/services/__init__.py index fb7cef6b..286950a1 100644 --- a/src/visionset/kernel/services/__init__.py +++ b/src/visionset/kernel/services/__init__.py @@ -14,6 +14,7 @@ from visionset.kernel.services.project_service import ProjectService from visionset.kernel.services.release_service import ReleaseService from visionset.kernel.services.schema_service import SchemaService +from visionset.kernel.services.source_service import SourceService from visionset.kernel.services.workspace_service import ( BLOBS_DIRNAME, DB_FILENAME, @@ -30,5 +31,6 @@ "ProjectService", "ReleaseService", "SchemaService", + "SourceService", "WorkspaceService", ] diff --git a/src/visionset/kernel/services/source_service.py b/src/visionset/kernel/services/source_service.py new file mode 100644 index 00000000..df95a801 --- /dev/null +++ b/src/visionset/kernel/services/source_service.py @@ -0,0 +1,233 @@ +# usage: from visionset.kernel.services import SourceService +"""Sources: the one door to the record that raw data was offered to a project. + +A ``Source`` is a receipt, not a payload — it says *this directory* or *this +clip* is where a project's assets came from, when it was registered, and what a +probe made of it. Materializing assets out of it is the ingest pipeline's job +(#20); this service only ever writes one row. + +**Two registration methods, not one ``register(kind=...)``.** The arguments +genuinely differ: a clip needs a decomposition rate and gets probed, a directory +needs neither and is not walked. That is the same argument that made +``ImageProcessor`` and ``VideoProcessor`` two protocols instead of one — a single +entry point would have to accept parameters that are meaningless for half its +callers. + +**Registration is idempotent, and the match key is ``(kind, path, +extraction_fps)``.** Registering the same origin twice returns the same +``Source`` rather than a second one, so that once #20 gives ``asset.source_id`` a +target, "which source did this asset come from?" has one answer. The key +deliberately excludes ``capture_params``: fragmenting one directory into two +sources because an operator typed a different lens note would defeat the point. +It also excludes the probed ``VideoMetadata`` — a clip replaced at a known path +is still that path's source, so its recorded provenance is **refreshed in +place** rather than left describing a file that is gone. ``registered_at`` is +never rewritten; it is the first registration. + +**The idempotency has no constraint underneath it, and that is a deliberate, +named gap.** ``docs/persistence.md`` says a rule with no backstop is a wish, and +every other uniqueness rule in this store has a unique index behind it. This one +does not, because the natural key includes a float parameter and a nullable one +at that, and because two concurrent registrations of one folder are not yet able +to hurt anything: no row references a source, so a duplicate is inert. That +stops being true at #20, when ``asset.source_id`` gets a target and the winner of +a race starts deciding an asset's recorded origin. **#20 is where this needs an +index under it**, alongside the store's other known concurrency gap (#80, the +untranslated ``OperationalError``). + +Composition follows the rule in ``docs/workspaces.md``: this service takes an +open ``WorkspaceService`` and nothing else, and reaches ``video_processor`` +through it. It never names an adapter. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from pathlib import Path +from uuid import UUID + +from visionset.kernel.domain import Project, Source, SourceKind, VideoProvenance, canonical_path +from visionset.kernel.errors import ProjectNotFound, SourceNotFound +from visionset.kernel.ports import DEFAULT_EXTRACTION_FPS, UnitOfWork +from visionset.kernel.services.workspace_service import WorkspaceService + + +class SourceService: + """Register, read and list the origins of one project's raw data.""" + + def __init__(self, workspace: WorkspaceService) -> None: + self._workspace = workspace + + # --- reading ----------------------------------------------------------- + + def get(self, source_id: UUID) -> Source: + """The source with that id. + + Raises: + SourceNotFound: no such source in this workspace. + """ + with self._workspace.unit_of_work() as uow: + return self.require_source(uow, source_id) + + # --- writing ----------------------------------------------------------- + + def register_images( + self, + project_id: UUID, + directory: Path, + *, + capture_params: Mapping[str, str] | None = None, + ) -> Source: + """Record a directory of stills as an origin for this project. + + The directory is checked to exist and to be a directory, and that is all: + what is *in* it is read at ingest, because a count taken now would be + stale by the time anything used it. + + Registering a directory already registered for this project returns the + existing source. Differing ``capture_params`` are written onto it rather + than making a second one — see the module docstring. + + Raises: + ProjectNotFound: no such project in this workspace. + FileNotFoundError: there is nothing at ``directory``. + NotADirectoryError: ``directory`` is there but is not one. + """ + path = canonical_path(directory) + if not Path(path).is_dir(): + raise NotADirectoryError(f"{path} is not a directory") + return self._register( + project_id, + SourceKind.IMAGE_DIRECTORY, + path, + video=None, + capture_params=capture_params, + ) + + def register_video( + self, + project_id: UUID, + clip: Path, + *, + extraction_fps: float = DEFAULT_EXTRACTION_FPS, + capture_params: Mapping[str, str] | None = None, + ) -> Source: + """Record a video file as an origin, with what a probe makes of it. + + The probe runs **before** the transaction opens. It is an out-of-process + decoder, and holding a write transaction open across a subprocess is how + a single-writer SQLite store ends up reporting "database is locked" — + the same reason ``examples/sdk_end_to_end.py`` puts its blob writes + outside the ``unit_of_work``. + + The consequence is worth knowing: re-registering an already-known clip + still needs ffmpeg, because the freshly probed metadata is what keeps the + stored provenance honest when the file behind the path has changed. + + ``extraction_fps`` is part of the source's identity, not a per-run + option: the same clip at 1 fps and at 5 fps is two sources. See + ``domain/source.py`` for why the parameters live here and not on the job. + + Raises: + ProjectNotFound: no such project in this workspace. + FileNotFoundError: there is nothing at ``clip``. + ValueError: ``extraction_fps`` is not positive. + MediaToolUnavailable: ffmpeg is not installed on this machine. + UnsupportedMedia: the file is intact and is not a video we read. + CorruptMedia: the file is a video we read, and it is damaged. + """ + if extraction_fps <= 0: + raise ValueError(f"extraction_fps must be positive, got {extraction_fps}") + path = canonical_path(clip) + metadata = self._workspace.video_processor.probe(Path(path)) + return self._register( + project_id, + SourceKind.VIDEO, + path, + video=VideoProvenance(metadata=metadata, extraction_fps=extraction_fps), + capture_params=capture_params, + ) + + # --- lookups shared by the operations above ---------------------------- + + def require_source(self, uow: UnitOfWork, source_id: UUID) -> Source: + """The source, checked through its project so workspaces stay separate. + + Public, and taking a ``uow``, for the reason ``JobService.require_job`` + is: #20's ingest has to resolve a source *inside its own transaction* + before it writes assets against it, and a second spelling of this ladder + is a second place for it to be got wrong. + + Raises: + SourceNotFound: no such source in this workspace. + """ + source = uow.sources.get(source_id) + if source is not None: + project = uow.projects.get(source.project_id) + if project is not None and project.workspace_id == self._workspace.workspace_id: + return source + raise SourceNotFound( + f"no source {source_id} in workspace {self._workspace.workspace.name!r}" + ) + + def _register( + self, + project_id: UUID, + kind: SourceKind, + path: str, + *, + video: VideoProvenance | None, + capture_params: Mapping[str, str] | None, + ) -> Source: + """Add the source, or return the one that already stands for this origin.""" + params = dict(capture_params or {}) + extraction_fps = None if video is None else video.extraction_fps + with self._workspace.unit_of_work() as uow: + self._require_project(uow, project_id) + for stored in uow.sources.list(project_id): + if stored.kind is not kind or stored.path != path: + continue + if ( + None if stored.video is None else stored.video.extraction_fps + ) != extraction_fps: + continue + if stored.video == video and stored.capture_params == params: + return stored + # The path is the same and the parameters are the same, so this + # is the same source; the file behind it moved on. Refresh what + # was read off it rather than leaving a record that describes + # bytes nobody can produce any more. + return uow.sources.update( + stored.model_copy(update={"video": video, "capture_params": params}) + ) + return uow.sources.add( + Source( + project_id=project_id, + kind=kind, + path=path, + capture_params=params, + video=video, + ) + ) + + def _require_project(self, uow: UnitOfWork, project_id: UUID) -> Project: + """The project, or refuse because this workspace does not have it.""" + project = uow.projects.get(project_id) + if project is None or project.workspace_id != self._workspace.workspace_id: + raise ProjectNotFound( + f"no project {project_id} in workspace {self._workspace.workspace.name!r}" + ) + return project + + # ``list`` shadows the builtin for every annotation below it in a class body, + # so it is declared last. See ``BatchService`` for the precedent. + + def list(self, project_id: UUID) -> list[Source]: + """Every source registered for that project, in registration order. + + Raises: + ProjectNotFound: no such project in this workspace. + """ + with self._workspace.unit_of_work() as uow: + self._require_project(uow, project_id) + return uow.sources.list(project_id) diff --git a/tests/kernel/test_metadata_store.py b/tests/kernel/test_metadata_store.py index 75129723..3297578a 100644 --- a/tests/kernel/test_metadata_store.py +++ b/tests/kernel/test_metadata_store.py @@ -44,8 +44,11 @@ Project, Release, Source, + SourceKind, SplitRecipe, TaskGroup, + VideoMetadata, + VideoProvenance, Workspace, ) from visionset.kernel.ports import UNINITIALIZED, MetadataStore, UnitOfWork @@ -61,7 +64,25 @@ def _seed(uow: UnitOfWork) -> list[tuple[str, UUID]]: """Persist one of every entity, wired into a valid parent chain.""" workspace = uow.workspaces.add(Workspace(name="w", root_dir="/tmp/w")) project = uow.projects.add(Project(workspace_id=workspace.id, name="p", description="d")) - source = uow.sources.add(Source(project_id=project.id, uri="file:///images")) + # A *video* source, and an explicit timestamp, for the reason the release + # below carries a real ``SplitRecipe`` and a fixed ``created_at``: seeding + # the simple shape would leave the ``video`` JSON column NULL in every store + # test, and nothing here would ever exercise the nested round trip. + source = uow.sources.add( + Source( + project_id=project.id, + kind=SourceKind.VIDEO, + path="/data/clip.mp4", + registered_at=datetime(2026, 7, 27, 8, 0, tzinfo=UTC), + capture_params={"lens": "24mm"}, + video=VideoProvenance( + metadata=VideoMetadata( + width=64, height=48, fps=29.97, duration_seconds=2.0, codec="h264" + ), + extraction_fps=5.0, + ), + ) + ) ingest = uow.ingest_jobs.add(IngestJob(source_id=source.id)) first = uow.assets.add( Asset(project_id=project.id, content_hash="a" * 64, uri="file:///1.png", width=8, height=6) @@ -279,6 +300,44 @@ def test_schema_classes_and_attributes_round_trip(tmp_path: Path) -> None: store.close() +def test_a_source_video_provenance_and_timestamp_round_trip(tmp_path: Path) -> None: + """The nested probe result survives the JSON column, offset and all.""" + store = _store(tmp_path) + with store.unit_of_work() as uow: + source = uow.sources.get(_seed(uow)[2][1]) + assert source is not None + assert source.kind is SourceKind.VIDEO + assert source.path == "/data/clip.mp4" + assert source.capture_params == {"lens": "24mm"} + assert source.require_video() == VideoProvenance( + metadata=VideoMetadata( + width=64, height=48, fps=29.97, duration_seconds=2.0, codec="h264" + ), + extraction_fps=5.0, + ) + assert source.registered_at == datetime(2026, 7, 27, 8, 0, tzinfo=UTC) + assert source.registered_at.tzinfo is not None + store.close() + + +def test_a_source_without_video_provenance_round_trips_as_none(tmp_path: Path) -> None: + store = _store(tmp_path) + with store.unit_of_work() as uow: + project_id = _seed(uow)[1][1] + stored = uow.sources.add( + Source( + project_id=project_id, + kind=SourceKind.IMAGE_DIRECTORY, + path="/data/stills", + ) + ) + read_back = uow.sources.get(stored.id) + assert read_back is not None + assert read_back.video is None + assert read_back.capture_params == {} + store.close() + + def test_a_release_points_at_its_manifest_rather_than_carrying_it(tmp_path: Path) -> None: """The document is in the blob store; the row keeps its hash and a read cache.""" store = _store(tmp_path) diff --git a/tests/kernel/test_migrations.py b/tests/kernel/test_migrations.py index 5992128f..cd3533b6 100644 --- a/tests/kernel/test_migrations.py +++ b/tests/kernel/test_migrations.py @@ -59,6 +59,23 @@ def _downgrade_to_version_one(store: SqliteMetadataStore) -> None: ) ) connection.execute(text("CREATE INDEX ix_release_dataset_id ON release (dataset_id)")) + # Migration 7 rebuilt ``source`` for the same kind of reason, so its undo + # is hand-written too. ``ingest_job`` keeps a ``REFERENCES source (id)`` + # across this window; SQLite resolves foreign-key targets at DML time, + # not at DDL time, so the gap between the drop and the create is safe. + connection.execute(text("drop table source")) + connection.execute( + text( + "CREATE TABLE source (" + " id CHAR(32) NOT NULL," + " project_id CHAR(32) NOT NULL," + " kind VARCHAR NOT NULL," + " uri VARCHAR NOT NULL," + " PRIMARY KEY (id)," + " FOREIGN KEY(project_id) REFERENCES project (id) ON DELETE CASCADE)" + ) + ) + connection.execute(text("CREATE INDEX ix_source_project_id ON source (project_id)")) connection.execute(text("update _visionset_meta set format_version = 1")) @@ -146,6 +163,76 @@ def test_migration_six_refuses_a_workspace_that_still_holds_a_pre_release_row( store.close() +def test_migration_seven_gives_a_source_its_provenance(tmp_path: Path) -> None: + store = SqliteMetadataStore(tmp_path / "visionset.db") + store.initialize() + with store.engine.connect() as connection: + columns = {c["name"] for c in inspect(connection).get_columns("source")} + assert {"path", "registered_at", "capture_params", "video"} <= columns + assert "uri" not in columns + store.close() + + +def test_migration_seven_keeps_the_source_project_index(tmp_path: Path) -> None: + """The rebuild re-creates the table from ``_tables``, indexes included. + + Worth its own test: migration 6 dropped a table carrying a + ``UniqueConstraint``, not an ``Index``, so this path was never exercised. + """ + store = SqliteMetadataStore(tmp_path / "visionset.db") + store.initialize() + assert any("ix_source_project_id" in sql for sql in _schema(store)) + store.close() + + +@pytest.mark.parametrize( + ("table", "insert"), + [ + ( + "source", + "insert into source (id, project_id, kind, uri) " + "values ('s', 'p', 'local_folder', '/in')", + ), + ( + "ingest_job", + "insert into source (id, project_id, kind, uri) " + "values ('s', 'p', 'local_folder', '/in');" + "insert into ingest_job (id, source_id, state) values ('j', 's', 'pending')", + ), + ], +) +def test_migration_seven_refuses_a_workspace_that_still_holds_pre_provenance_rows( + tmp_path: Path, table: str, insert: str +) -> None: + """Neither the source nor its children are dropped on the quiet. + + ``ingest_job.source_id`` is ``ON DELETE CASCADE`` and the store turns foreign + keys on for every connection, so ``DROP TABLE source`` would take the jobs + with it *without raising*. Counting only ``source`` would let a workspace + with jobs slip through, which is why the migration counts both — and why the + second case here exists at all. + """ + store = SqliteMetadataStore(tmp_path / "visionset.db") + store.initialize() + _downgrade_to_version_one(store) + with store.engine.begin() as connection: + # A real parent chain, because foreign keys are on. + connection.execute(text("insert into workspace (id, name) values ('w', 'ws')")) + connection.execute( + text("insert into project (id, workspace_id, name) values ('p', 'w', 'proj')") + ) + for statement in insert.split(";"): + connection.execute(text(statement)) + + with pytest.raises(WorkspaceCorrupt, match="before SourceService existed"): + store.initialize() + + # And the rows are still there — refused, not emptied. + with store.engine.connect() as connection: + assert connection.execute(text(f"select count(*) from {table}")).scalar_one() == 1 + store.close() + + def test_a_fresh_database_and_a_migrated_one_have_the_same_schema(tmp_path: Path) -> None: """Migration 1 is ``create_all`` of *current* metadata, so the two paths differ. diff --git a/tests/kernel/test_project_service.py b/tests/kernel/test_project_service.py index 5aab107c..d70abff5 100644 --- a/tests/kernel/test_project_service.py +++ b/tests/kernel/test_project_service.py @@ -33,6 +33,7 @@ LabelClass, Release, Source, + SourceKind, ) from visionset.kernel.ports import UnitOfWork from visionset.kernel.services import ProjectService, WorkspaceService @@ -65,7 +66,9 @@ def _populate(workspace: WorkspaceService, project_id: UUID, dataset_id: UUID) - classes=[LabelClass(name="sign", geometry=GeometryType.BBOX)], ) ) - uow.sources.add(Source(project_id=project_id, kind="local_folder", uri="/tmp/in")) + uow.sources.add( + Source(project_id=project_id, kind=SourceKind.IMAGE_DIRECTORY, path="/tmp/in") + ) asset = uow.assets.add( Asset(project_id=project_id, content_hash=content_hash, uri="/tmp/in/a.png") ) diff --git a/tests/kernel/test_source_service.py b/tests/kernel/test_source_service.py new file mode 100644 index 00000000..7ae8516f --- /dev/null +++ b/tests/kernel/test_source_service.py @@ -0,0 +1,363 @@ +"""`SourceService`: registration, provenance, and the idempotency rule. + +Two things shape this file. + +The ffmpeg requirement arrives through `write_video`, which calls `require_ffmpeg` +itself — there is deliberately no module-level skip. The tests that need no clip +(directory registration, the not-found ladders, the domain invariant) have to run +on a machine without ffmpeg, and a module-level skip would take them with it. + +The idempotency assertions compare `id`s rather than counting rows wherever they +can, because "returns the same source" is the contract; "wrote one row" is how it +happens to be implemented. +""" + +from pathlib import Path +from uuid import uuid4 + +import pytest +from pydantic import ValidationError +from tests.fixtures.media import ( + GeneratedVideo, + write_corrupt_video, + write_unsupported_file, + write_video, +) + +from visionset.kernel import ( + ProjectNotFound, + SourceNotFound, + UnsupportedMedia, + WorkspaceCorrupt, +) +from visionset.kernel.domain import Source, SourceKind, VideoMetadata, VideoProvenance +from visionset.kernel.ports import DEFAULT_EXTRACTION_FPS +from visionset.kernel.services import ProjectService, SourceService, WorkspaceService + + +class Fixture: + """A workspace with one project and a directory of stills to point at.""" + + def __init__(self, tmp_path: Path, name: str = "ws") -> None: + self.tmp_path = tmp_path + self.workspace = WorkspaceService.init(tmp_path / name) + self.projects = ProjectService(self.workspace) + self.sources = SourceService(self.workspace) + self.project = self.projects.create(f"{name}-project") + self.stills = tmp_path / f"{name}-stills" + self.stills.mkdir() + + def clip(self, name: str = "clip.mp4", **kwargs: object) -> GeneratedVideo: + return write_video(self.tmp_path / name, **kwargs) # type: ignore[arg-type] + + def close(self) -> None: + self.workspace.close() + + +# --- registering a directory of stills -------------------------------------- + + +def test_an_image_directory_source_persists_and_rehydrates_completely(tmp_path: Path) -> None: + fx = Fixture(tmp_path) + registered = fx.sources.register_images( + fx.project.id, fx.stills, capture_params={"site": "yard-3"} + ) + fx.close() + + reopened = WorkspaceService.open(tmp_path / "ws") + read_back = SourceService(reopened).get(registered.id) + assert read_back.kind is SourceKind.IMAGE_DIRECTORY + assert read_back.path == str(fx.stills.resolve()) + assert read_back.capture_params == {"site": "yard-3"} + assert read_back.video is None + assert read_back.registered_at == registered.registered_at + assert read_back.registered_at.tzinfo is not None + reopened.close() + + +def test_registering_a_directory_that_is_not_there_is_a_file_not_found_error( + tmp_path: Path, +) -> None: + fx = Fixture(tmp_path) + with pytest.raises(FileNotFoundError): + fx.sources.register_images(fx.project.id, tmp_path / "absent") + fx.close() + + +def test_registering_a_file_as_an_image_directory_is_a_not_a_directory_error( + tmp_path: Path, +) -> None: + fx = Fixture(tmp_path) + plain = tmp_path / "notes.txt" + plain.write_text("hello") + with pytest.raises(NotADirectoryError): + fx.sources.register_images(fx.project.id, plain) + fx.close() + + +def test_a_relative_path_and_its_absolute_form_are_one_source( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Canonicalization is the whole reason `path` is not stored as given.""" + fx = Fixture(tmp_path) + absolute = fx.sources.register_images(fx.project.id, fx.stills) + monkeypatch.chdir(tmp_path) + relative = fx.sources.register_images(fx.project.id, Path(f"./{fx.stills.name}")) + assert relative.id == absolute.id + assert fx.sources.list(fx.project.id) == [absolute] + fx.close() + + +# --- registering a clip ------------------------------------------------------ + + +def test_a_video_source_stores_the_original_fps_from_the_probe(tmp_path: Path) -> None: + """Issue #18's second acceptance criterion, asserted against the generator.""" + fx = Fixture(tmp_path) + clip = fx.clip(fps=10, duration_seconds=2.0) + registered = fx.sources.register_video(fx.project.id, clip.path, extraction_fps=5.0) + + provenance = registered.require_video() + assert provenance.metadata.fps == pytest.approx(clip.fps) + assert provenance.metadata.width == clip.width + assert provenance.metadata.height == clip.height + assert provenance.metadata.duration_seconds == pytest.approx(clip.duration_seconds, abs=0.2) + assert provenance.metadata.codec + assert provenance.extraction_fps == 5.0 + fx.close() + + +def test_a_video_source_rehydrates_its_whole_provenance(tmp_path: Path) -> None: + fx = Fixture(tmp_path) + clip = fx.clip() + registered = fx.sources.register_video(fx.project.id, clip.path, extraction_fps=2.5) + fx.close() + + reopened = WorkspaceService.open(tmp_path / "ws") + read_back = SourceService(reopened).get(registered.id) + assert read_back == registered + assert read_back.require_video() == registered.require_video() + reopened.close() + + +def test_registering_a_video_defaults_to_the_port_extraction_rate(tmp_path: Path) -> None: + fx = Fixture(tmp_path) + registered = fx.sources.register_video(fx.project.id, fx.clip().path) + assert registered.require_video().extraction_fps == DEFAULT_EXTRACTION_FPS + fx.close() + + +def test_a_non_positive_extraction_rate_is_refused_before_anything_is_probed( + tmp_path: Path, +) -> None: + fx = Fixture(tmp_path) + with pytest.raises(ValueError, match="extraction_fps must be positive"): + fx.sources.register_video(fx.project.id, tmp_path / "never-read.mp4", extraction_fps=0) + fx.close() + + +def test_registering_something_that_is_not_a_video_stores_nothing(tmp_path: Path) -> None: + fx = Fixture(tmp_path) + not_a_clip = write_unsupported_file(tmp_path / "notes.mp4") + with pytest.raises(UnsupportedMedia): + fx.sources.register_video(fx.project.id, not_a_clip) + assert fx.sources.list(fx.project.id) == [] + fx.close() + + +def test_a_truncated_clip_registers_because_a_probe_only_reads_the_header( + tmp_path: Path, +) -> None: + """Registration is not a validation pass, and #19/#20 must not assume it is. + + `write_corrupt_video` truncates a faststart clip, so the index at the front + still describes the whole thing and ffprobe answers happily; ffmpeg only + fails once a decode runs off the end of the bytes. So damage surfaces at + extraction, not here — and the duration this source records is the one the + intact file would have had. + """ + fx = Fixture(tmp_path) + broken = write_corrupt_video(tmp_path / "broken.mp4") + registered = fx.sources.register_video(fx.project.id, broken.path) + assert registered.require_video().metadata.duration_seconds == pytest.approx( + broken.duration_seconds, abs=0.2 + ) + fx.close() + + +# --- idempotency ------------------------------------------------------------- + + +def test_registering_the_same_directory_twice_returns_the_same_source(tmp_path: Path) -> None: + fx = Fixture(tmp_path) + first = fx.sources.register_images(fx.project.id, fx.stills) + second = fx.sources.register_images(fx.project.id, fx.stills) + assert second == first + assert fx.sources.list(fx.project.id) == [first] + fx.close() + + +def test_registering_the_same_clip_at_the_same_rate_returns_the_same_source( + tmp_path: Path, +) -> None: + fx = Fixture(tmp_path) + clip = fx.clip() + first = fx.sources.register_video(fx.project.id, clip.path, extraction_fps=5.0) + second = fx.sources.register_video(fx.project.id, clip.path, extraction_fps=5.0) + assert second == first + assert len(fx.sources.list(fx.project.id)) == 1 + fx.close() + + +def test_the_same_clip_at_a_different_rate_is_a_second_source(tmp_path: Path) -> None: + """The decomposition rate is part of the source's identity, deliberately.""" + fx = Fixture(tmp_path) + clip = fx.clip() + slow = fx.sources.register_video(fx.project.id, clip.path, extraction_fps=1.0) + fast = fx.sources.register_video(fx.project.id, clip.path, extraction_fps=5.0) + assert fast.id != slow.id + assert {s.id for s in fx.sources.list(fx.project.id)} == {slow.id, fast.id} + fx.close() + + +def test_differing_capture_params_update_the_source_rather_than_forking_it( + tmp_path: Path, +) -> None: + """A typo in a lens note must not give one directory two origins.""" + fx = Fixture(tmp_path) + first = fx.sources.register_images(fx.project.id, fx.stills, capture_params={"lens": "24mm"}) + second = fx.sources.register_images(fx.project.id, fx.stills, capture_params={"lens": "35mm"}) + assert second.id == first.id + assert second.capture_params == {"lens": "35mm"} + assert fx.sources.list(fx.project.id) == [second] + fx.close() + + +def test_a_replaced_clip_refreshes_its_provenance_and_keeps_its_identity(tmp_path: Path) -> None: + """The path is the source; the bytes behind it are what a re-probe is for.""" + fx = Fixture(tmp_path) + clip = fx.clip(fps=10, duration_seconds=2.0) + first = fx.sources.register_video(fx.project.id, clip.path, extraction_fps=1.0) + + write_video(clip.path, fps=25, duration_seconds=1.0) + second = fx.sources.register_video(fx.project.id, clip.path, extraction_fps=1.0) + + assert second.id == first.id + assert second.registered_at == first.registered_at + assert second.require_video().metadata.fps == pytest.approx(25) + assert second.require_video().metadata.fps != first.require_video().metadata.fps + assert len(fx.sources.list(fx.project.id)) == 1 + fx.close() + + +# --- scope and lookups ------------------------------------------------------- + + +def test_registering_against_an_unknown_project_is_refused(tmp_path: Path) -> None: + fx = Fixture(tmp_path) + with pytest.raises(ProjectNotFound): + fx.sources.register_images(uuid4(), fx.stills) + fx.close() + + +def test_getting_an_unknown_source_is_refused(tmp_path: Path) -> None: + fx = Fixture(tmp_path) + with pytest.raises(SourceNotFound): + fx.sources.get(uuid4()) + fx.close() + + +def test_listing_an_unknown_project_is_refused(tmp_path: Path) -> None: + fx = Fixture(tmp_path) + with pytest.raises(ProjectNotFound): + fx.sources.list(uuid4()) + fx.close() + + +def test_sources_are_scoped_to_their_project(tmp_path: Path) -> None: + fx = Fixture(tmp_path) + other = fx.projects.create("other") + mine = fx.sources.register_images(fx.project.id, fx.stills) + assert fx.sources.list(other.id) == [] + assert fx.sources.list(fx.project.id) == [mine] + fx.close() + + +def test_a_source_in_another_workspace_reads_as_missing(tmp_path: Path) -> None: + """Not as forbidden — this service speaks for one workspace.""" + theirs = Fixture(tmp_path, name="theirs") + stranger = theirs.sources.register_images(theirs.project.id, theirs.stills) + theirs.close() + + mine = Fixture(tmp_path, name="mine") + with pytest.raises(SourceNotFound): + mine.sources.get(stranger.id) + mine.close() + + +def test_require_source_resolves_inside_a_callers_transaction(tmp_path: Path) -> None: + """The shape #20's ingest needs: one lookup, one transaction, no re-entry.""" + fx = Fixture(tmp_path) + registered = fx.sources.register_images(fx.project.id, fx.stills) + with fx.workspace.unit_of_work() as uow: + assert fx.sources.require_source(uow, registered.id) == registered + with pytest.raises(SourceNotFound): + fx.sources.require_source(uow, uuid4()) + fx.close() + + +# --- the domain invariant ---------------------------------------------------- + + +def _provenance() -> VideoProvenance: + return VideoProvenance( + metadata=VideoMetadata(width=4, height=4, fps=30.0, duration_seconds=1.0, codec="h264"), + extraction_fps=1.0, + ) + + +def test_an_image_directory_source_may_not_carry_video_provenance() -> None: + with pytest.raises(ValidationError, match="must not carry video provenance"): + Source( + project_id=uuid4(), + kind=SourceKind.IMAGE_DIRECTORY, + path="/data", + video=_provenance(), + ) + + +def test_a_video_source_must_carry_video_provenance() -> None: + with pytest.raises(ValidationError, match="must carry video provenance"): + Source(project_id=uuid4(), kind=SourceKind.VIDEO, path="/data/clip.mp4") + + +def test_the_invariant_survives_assignment_not_only_construction() -> None: + """`validate_assignment` is on for exactly this: a model validator does not + re-run on attribute assignment, so without it the pair could drift apart + after a valid construction.""" + source = Source( + project_id=uuid4(), + kind=SourceKind.VIDEO, + path="/data/clip.mp4", + video=_provenance(), + ) + with pytest.raises(ValidationError, match="must not carry video provenance"): + source.kind = SourceKind.IMAGE_DIRECTORY + + +def test_a_naive_registration_timestamp_is_refused() -> None: + from datetime import datetime + + with pytest.raises(ValidationError, match="timezone-aware"): + Source( + project_id=uuid4(), + kind=SourceKind.IMAGE_DIRECTORY, + path="/data", + registered_at=datetime(2026, 7, 27, 9, 0), + ) + + +def test_require_video_refuses_a_source_that_is_not_a_clip() -> None: + source = Source(project_id=uuid4(), kind=SourceKind.IMAGE_DIRECTORY, path="/data") + with pytest.raises(WorkspaceCorrupt, match="no video provenance"): + source.require_video()