diff --git a/docs/api.md b/docs/api.md index 80fde049..5c11e564 100644 --- a/docs/api.md +++ b/docs/api.md @@ -38,11 +38,26 @@ GET /projects/{project_id}/schema the version in force POST /projects/{project_id}/schema/versions GET /projects/{project_id}/schema/versions GET /projects/{project_id}/schema/versions/{version} +POST /projects/{project_id}/sources/images multipart +POST /projects/{project_id}/sources/video multipart +GET /projects/{project_id}/sources +GET /sources/{source_id} +POST /sources/{source_id}/ingest-jobs launch +GET /sources/{source_id}/ingest-jobs +GET /ingest-jobs/{job_id} poll +POST /ingest-jobs/{job_id}/resume +GET /batches/{batch_id}/assets ``` The active schema is the collection's **parent**, not a member of it, because "in force" is a property of the schema rather than a version number a client could guess. +A **collection** hangs off whatever owns it; an individually addressable **resource** does not. +A source belongs to one project, so listing and creating happen under it — but a source has an +id of its own, and nesting `/projects/{p}/sources/{s}/ingest-jobs/{j}` would put four segments +in front of a job that one id already identifies. A schema version has no such id, which is why +it stays nested all the way down. + **Ids are UUIDs**, canonical hyphenated form, in the path. One deliberate exception: a **schema version is an integer 1..N**, because that is the handle the domain itself uses — an annotation records `schema_version`, and a batch pins one at approval. A malformed UUID never reaches a @@ -68,7 +83,37 @@ pre-check either one: the flag goes to the SDK and the SDK's refusal is what car `CONFIRMATION_REQUIRED` or `DESTRUCTIVE_SCHEMA_CHANGE`. **Statuses.** 201 with the created resource in the body; 200 for a read or an update; 204 with an -empty body for a delete. +empty body for a delete. And **202 when the work has not happened yet** — see below. + +**A long operation is launched and then polled.** Ingest is the first one, and the shape it set +is the one every later long operation uses: + +``` +POST /sources/{id}/ingest-jobs → 202 Accepted + Location: /ingest-jobs/{job_id} + { "id": …, "state": "pending", "processed": 0, … } + +GET /ingest-jobs/{job_id} → 200 { "state": "running", "processed": 12, … } +GET /ingest-jobs/{job_id} → 200 { "state": "completed", "batch_id": …, … } +``` + +**202, not 201**: the row exists, the work does not. The row is what makes the id worth handing +back — it is written and committed before the response is sent, so the first poll always finds +something. That is also why anything the request can refuse is refused *synchronously*: an +unknown source is a 404 here rather than a 202 pointing at a job nobody wrote, and resuming a +`completed` run is a 409 here rather than a background no-op a client could not distinguish from +a redo. Everything that goes wrong *after* the launch is reported on the job — `error` for the +one fatal cause, `failures` for the per-item report — because by then there is no request left +to answer. + +**Uploads are multipart, and the only non-JSON request shape.** Registering a source means +sending the bytes: one `files` part per image, or one `file` part plus an `extraction_fps` field +for a clip. VisionSet sets **no size limit of its own** — parts are spooled to disk past 1 MiB +and streamed from there, so memory does not grow with the file — which means the real ceilings +are your reverse proxy's (`client_max_body_size` in nginx) and free disk. Uploaded bytes are +staged under `/uploads//` and, like blobs, are **never deleted**: a workspace +grows with what was offered to it, not only with what it kept. Re-uploading identical files +under identical names is free — it stages to the same path and returns the same source. **Request bodies forbid unknown fields.** A misspelled key is a 422 `VALIDATION_ERROR`, never a silently ignored one — a typo that looked like it worked is worse than a refusal. diff --git a/docs/batches.md b/docs/batches.md index 33440976..6a1b4e81 100644 --- a/docs/batches.md +++ b/docs/batches.md @@ -45,6 +45,13 @@ Membership is a **set** — adding an asset the batch already holds changes noth removing one it does not hold is a no-op. Order is the order assets were first added. Every asset must belong to the batch's project, else `AssetNotFound`. +Reading it back is `batches.assets(batch_id)`, which answers with the `Asset` rows in that +stored order — `DatasetService.assets` over the trunk, applied to a batch. It is how "what did +that ingest actually gather" is answered, and a member whose asset is not stored is +`WorkspaceCorrupt` rather than a silently shorter list: `batch_asset.asset_id` cascades from +`asset`, so that cannot happen while foreign keys are on, and a batch quietly holding less than +it says is worse than a refusal. + Excluding an asset after approval is a different act: it is marked **`skipped`**, a per-asset progress decision the record keeps rather than a membership edit that erases it. Somebody decided not to label that asset, and that decision is worth more than a tidy list. The diff --git a/docs/ingest.md b/docs/ingest.md index b6b4b391..91984696 100644 --- a/docs/ingest.md +++ b/docs/ingest.md @@ -67,6 +67,29 @@ 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. +## Asking for a run and doing it are two calls + +```python +job = ingest.enqueue(source.id, batch_name="monday") # refuses now, reads nothing +result = ingest.resume(job.id) # does the work +``` + +`ingest(...)` is those two composed, and that is all it is. The split exists because a caller +that cannot wait — the [REST API](api.md), and one day a queue — needs the **row before the +work**: the id it hands back has to name something the next request can find. Every refusal +`enqueue` can make it makes before the insert, so a launch that fails leaves no job at all and +a launch that succeeds leaves one that is already pollable. + +`resume` is what picks a `pending` job up, which is why `pending → running` was in the +transition table from the start. `resumable(job_id)` is the same friendly pre-check without the +work, for a caller that runs the second half elsewhere and needs the refusal on its own thread — +a launch that answered "accepted" and only discovered in a worker that the job was already +`completed` would give nobody a way to tell a redo from a no-op. + +Nothing here decides *when* the second half runs. That is deliberately not the kernel's +business: the API supplies a single background worker (`server/runner.py`), the CLI just calls +`ingest`, and neither arrangement is visible in this module. + ## The run has a lifecycle, and it is a table `INGEST_TRANSITIONS` in `domain/ingest.py` is the whole of what is legal. `IngestService` @@ -78,9 +101,11 @@ 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. +A job is created `pending` and moved to `running` by whoever picks the work up. Through +`ingest(...)` those are the same call and the state is over in microseconds; through +`enqueue` + `resume` they are not, and a `pending` row is a run somebody asked for that has not +started. That is why the state was spelled out before anything left one behind — adding it later +would have meant 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 @@ -98,8 +123,8 @@ stuck row as the only evidence the crash left. `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 `GET /ingest-jobs/{id}` returns and what the UI will poll; nothing about it is specific to +being in the same process. | | what it means | | --- | --- | @@ -255,10 +280,38 @@ Membership is everything the run ingested, deduplicated assets included: a dupli 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. +## Over HTTP + +The [API](api.md) is `enqueue` and `resume` with a worker between them. + +``` +POST /projects/{id}/sources/video multipart: the clip + extraction_fps → 201 SourceOut +POST /sources/{id}/ingest-jobs → 202 IngestJobOut +GET /ingest-jobs/{id} → 200 IngestJobOut +GET /batches/{id}/assets → 200 the assets +``` + +The launch calls `enqueue` on the request thread and hands the `pending` job to a **single +background worker** — one, so that runs serialize against a single-writer store rather than +racing each other. What that buys the *reader* is what [#80's concurrency posture](workspaces.md) +was for: a client polling while the worker holds a write transaction reads through WAL instead +of waiting on it. + +**Where a refusal appears depends on when it can be known.** An unknown source or a blank batch +name is refused synchronously, with a 404 or a 422 — the launch never returns 202 pointing at a +job row nobody wrote. Everything after that is on the job: `state` becomes `failed` and `error` +carries the cause, while individual unreadable items sit in `failures` and do not fail the run +at all. That split is the same one this service already makes; HTTP just changes where you read +it. + +Registration over HTTP is **upload-only**, and the bytes are staged content-addressed — see +[sources.md](sources.md). Targeting an existing draft batch is not on the wire yet: batches have +no endpoints until the batch and job API lands, so there is nothing for a client to name. + ## What is deliberately not here yet -- **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 — which is why - a job is created `pending` and why progress is read off the row rather than off a callback. +- **No scheduler.** `enqueue` and `resume` are two calls; nothing in the kernel decides when the + second one runs. The API supplies one background worker, the CLI supplies the calling thread, + and a queue would be a third arrangement neither of them would notice. - **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. diff --git a/docs/sources.md b/docs/sources.md index c08cc679..8a5c19af 100644 --- a/docs/sources.md +++ b/docs/sources.md @@ -117,6 +117,27 @@ where a directory was wanted is a `NotADirectoryError`. Both are about the machi the workspace, so both stay outside the `VisionSetError` tree — the same line `MediaToolUnavailable` sits on. +## Over HTTP, a path is an upload + +`SourceService` registers by path, and an HTTP client has bytes rather than a path. So the +[REST API](api.md) takes multipart — one `files` part per image, or one `file` part plus an +`extraction_fps` field for a clip — writes the parts under `/uploads/`, and registers +what it wrote. There is **no route that accepts a server-side path**: it would hand every token +holder an arbitrary-directory read, and the two surfaces that legitimately hold real paths, the +CLI and MCP, call the SDK in-process and never go through HTTP. + +The staging directory is named by a **digest of the whole part set** — sha-256 over the sorted +`name:sha256` lines — which is what makes the idempotency above survive the trip. The same files +under the same names stage to the same path, so a repeated upload returns the *same* `Source` +instead of a second one over a second copy on disk. Different bytes, or the same bytes under a +different filename, are a different offer and stage apart. + +That upload-only choice has a quiet dividend: because the server just wrote the file, the +`FileNotFoundError` and `NotADirectoryError` below are unreachable from HTTP. Neither is a +`VisionSetError`, so neither has an entry in the API's error table — and neither needs one. + +A client never sees `path`. `SourceOut` publishes the filename and nothing about the machine. + ## Registration is not a validation pass `register_video` probes; it does not decode. A clip whose tail has been truncated still has a diff --git a/docs/workspaces.md b/docs/workspaces.md index 1ead1483..56d2a65d 100644 --- a/docs/workspaces.md +++ b/docs/workspaces.md @@ -10,9 +10,17 @@ happens in the context of exactly one, so `WorkspaceService` is both the way in visionset.db-wal SQLite's write-ahead log — present only while the workspace is open visionset.db-shm its shared-memory index — likewise blobs/ FilesystemBlobStore root, sharded // + uploads/ written only by the REST server — see below ``` -Nothing else is written. The store runs in **WAL mode**, which is why the two sidecars are +Nothing the kernel writes is outside those four entries. `uploads/` is the exception and it +belongs to somebody else: the [REST API](api.md) stages multipart uploads there, named by a +digest of the part set, so that `SourceService` — which registers a source by *path* — has a +path to be given. The kernel neither writes nor reads it, `open` simply tolerates it the way it +tolerates anything else beside the database and `blobs/`, and the CLI and MCP surfaces never +create one because they already hold real paths. Like blobs, staged uploads are never deleted. + +The store runs in **WAL mode**, which is why the two sidecars are part of the format: `close()` checkpoints them into `visionset.db` and removes them, so a workspace at rest is still just the database and the blobs — but a workspace that is *open*, or one whose process was killed, is all four entries. diff --git a/openapi.json b/openapi.json index fee94a3a..d189a17f 100644 --- a/openapi.json +++ b/openapi.json @@ -1,6 +1,144 @@ { "components": { "schemas": { + "AssetOut": { + "description": "One ingested item.", + "properties": { + "content_hash": { + "title": "Content Hash", + "type": "string" + }, + "format": { + "anyOf": [ + { + "$ref": "#/components/schemas/ImageFormat" + }, + { + "type": "null" + } + ] + }, + "frame_index": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Frame Index" + }, + "frame_timestamp": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Frame Timestamp" + }, + "height": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Height" + }, + "id": { + "format": "uuid", + "title": "Id", + "type": "string" + }, + "modality": { + "const": "image", + "title": "Modality", + "type": "string" + }, + "project_id": { + "format": "uuid", + "title": "Project Id", + "type": "string" + }, + "source_id": { + "anyOf": [ + { + "format": "uuid", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Source Id" + }, + "thumbnail_hash": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Thumbnail Hash" + }, + "width": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Width" + } + }, + "required": [ + "id", + "project_id", + "modality", + "content_hash", + "width", + "height", + "format", + "source_id", + "frame_index", + "frame_timestamp", + "thumbnail_hash" + ], + "title": "AssetOut", + "type": "object" + }, + "AssetPage": { + "description": "A page of assets.", + "properties": { + "items": { + "items": { + "$ref": "#/components/schemas/AssetOut" + }, + "title": "Items", + "type": "array" + }, + "total": { + "title": "Total", + "type": "integer" + } + }, + "required": [ + "items", + "total" + ], + "title": "AssetPage", + "type": "object" + }, "AttributeBody": { "additionalProperties": false, "description": "A typed attribute on a label class.", @@ -63,6 +201,46 @@ "title": "AttributeBody", "type": "object" }, + "Body_register_image_source": { + "properties": { + "files": { + "description": "The images, as one multipart part each.", + "items": { + "contentMediaType": "application/octet-stream", + "type": "string" + }, + "title": "Files", + "type": "array" + } + }, + "required": [ + "files" + ], + "title": "Body_register_image_source", + "type": "object" + }, + "Body_register_video_source": { + "properties": { + "extraction_fps": { + "default": 1.0, + "description": "Frames per second to cut the clip at. One per second by default.", + "exclusiveMinimum": 0.0, + "title": "Extraction Fps", + "type": "number" + }, + "file": { + "contentMediaType": "application/octet-stream", + "description": "The clip.", + "title": "File", + "type": "string" + } + }, + "required": [ + "file" + ], + "title": "Body_register_video_source", + "type": "object" + }, "ErrorBody": { "description": "The one error shape this API emits, at every status.", "properties": { @@ -112,6 +290,186 @@ "title": "GeometryType", "type": "string" }, + "ImageFormat": { + "description": "Every still-image encoding VisionSet accepts. See the module docstring.\n\nA ``StrEnum`` rather than a ``Literal``, unlike ``Asset.modality``: that one\nhas a single member, where an enum would be ceremony, and this one is a\nclosed set whose whole purpose is to grow deliberately. It costs the\npersistence layer nothing \u2014 a ``StrEnum`` member *is* a ``str``, and the\ntables already store every other enum as ``String``.", + "enum": [ + "jpeg", + "png" + ], + "title": "ImageFormat", + "type": "string" + }, + "IngestFailureKind": { + "description": "Why one item did not become an asset, split by what to do about it.\n\nAn enum rather than a plain ``str``, on exactly ``SourceKind``'s terms: the\nset is closed, no writer outside this build produces a value, and the kernel\nbranches on it. What makes it worth a type at all is that a report has to be\n**grouped**, not read \u2014 ``CorruptMedia``'s docstring is explicit that a\nreport unable to separate the two would bury real data loss under ordinary\noperator noise, and a reason sentence cannot be grouped on.", + "enum": [ + "unsupported", + "corrupt" + ], + "title": "IngestFailureKind", + "type": "string" + }, + "IngestFailureOut": { + "description": "One item a run could not read, and why.", + "properties": { + "kind": { + "$ref": "#/components/schemas/IngestFailureKind" + }, + "name": { + "title": "Name", + "type": "string" + }, + "reason": { + "title": "Reason", + "type": "string" + } + }, + "required": [ + "name", + "kind", + "reason" + ], + "title": "IngestFailureOut", + "type": "object" + }, + "IngestJobOut": { + "description": "One run of one source, and how far it has got.", + "properties": { + "batch_id": { + "anyOf": [ + { + "format": "uuid", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Batch Id" + }, + "batch_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Batch Name" + }, + "error": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Error" + }, + "failures": { + "items": { + "$ref": "#/components/schemas/IngestFailureOut" + }, + "title": "Failures", + "type": "array" + }, + "id": { + "format": "uuid", + "title": "Id", + "type": "string" + }, + "processed": { + "title": "Processed", + "type": "integer" + }, + "source_id": { + "format": "uuid", + "title": "Source Id", + "type": "string" + }, + "state": { + "$ref": "#/components/schemas/IngestState" + }, + "total": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Total" + } + }, + "required": [ + "id", + "source_id", + "state", + "error", + "batch_id", + "batch_name", + "processed", + "total", + "failures" + ], + "title": "IngestJobOut", + "type": "object" + }, + "IngestJobPage": { + "description": "A page of ingest jobs.", + "properties": { + "items": { + "items": { + "$ref": "#/components/schemas/IngestJobOut" + }, + "title": "Items", + "type": "array" + }, + "total": { + "title": "Total", + "type": "integer" + } + }, + "required": [ + "items", + "total" + ], + "title": "IngestJobPage", + "type": "object" + }, + "IngestStart": { + "additionalProperties": false, + "description": "What launching a run needs, which is almost nothing.", + "properties": { + "batch_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Batch Name" + } + }, + "title": "IngestStart", + "type": "object" + }, + "IngestState": { + "description": "Lifecycle: pending -> running -> (completed | failed) -> running.\n\n``IngestService`` owns the moves; ``INGEST_TRANSITIONS`` below is the whole\nof what is legal.", + "enum": [ + "pending", + "running", + "completed", + "failed" + ], + "title": "IngestState", + "type": "string" + }, "LabelClassBody": { "additionalProperties": false, "description": "One labelable class, bound to a geometry.", @@ -310,28 +668,234 @@ ], "title": "SchemaVersionPage", "type": "object" - } - }, - "securitySchemes": { - "HTTPBearer": { - "description": "A workspace API token, created with `visionset token create`. Sent as `Authorization: Bearer `.", - "scheme": "bearer", - "type": "http" - } - } - }, - "info": { - "description": "REST surface of the VisionSet SDK. The committed openapi.json is the contract.", - "title": "Robomous VisionSet API", - "version": "0.0.1.dev0" - }, - "openapi": "3.1.0", - "paths": { - "/health": { - "get": { - "description": "Liveness probe. Public \u2014 no token required.", - "operationId": "health", - "responses": { + }, + "SourceKind": { + "description": "The shapes of raw input VisionSet accepts.\n\nAn enum, where ``DatasetChange.operation`` and ``VideoMetadata.codec`` are\nplain ``str``. That doctrine turns on one question \u2014 *can something outside\nthis build write the value?* A change-log entry outlives the release that\nwrote it and a codec name is whatever ffmpeg decides to call it, so both have\nto stay readable when they name something this build never heard of.\n\nNeither applies here. ``SourceService`` is the only door to a ``Source``, so\nno foreign writer exists; the kernel **branches** on this value, in the two\nregistration methods and in the invariant tying :attr:`Source.video` to\n:attr:`SourceKind.VIDEO`, and a branch on a magic string is the shape this\ncodebase replaces with a table; and the set grows by a deliberate kernel\nchange with a service method behind it. That is ``ImageFormat`` /\n``BatchState`` / ``IngestState`` territory, and it costs persistence nothing\n\u2014 a ``StrEnum`` member *is* a ``str``.", + "enum": [ + "image_directory", + "video" + ], + "title": "SourceKind", + "type": "string" + }, + "SourceOut": { + "description": "A registered origin: a folder of stills, or a clip.", + "properties": { + "id": { + "format": "uuid", + "title": "Id", + "type": "string" + }, + "kind": { + "$ref": "#/components/schemas/SourceKind" + }, + "name": { + "title": "Name", + "type": "string" + }, + "project_id": { + "format": "uuid", + "title": "Project Id", + "type": "string" + }, + "registered_at": { + "format": "date-time", + "title": "Registered At", + "type": "string" + }, + "video": { + "anyOf": [ + { + "$ref": "#/components/schemas/VideoProvenanceOut" + }, + { + "type": "null" + } + ] + } + }, + "required": [ + "id", + "project_id", + "kind", + "name", + "registered_at", + "video" + ], + "title": "SourceOut", + "type": "object" + }, + "SourcePage": { + "description": "A page of sources.", + "properties": { + "items": { + "items": { + "$ref": "#/components/schemas/SourceOut" + }, + "title": "Items", + "type": "array" + }, + "total": { + "title": "Total", + "type": "integer" + } + }, + "required": [ + "items", + "total" + ], + "title": "SourcePage", + "type": "object" + }, + "VideoProvenanceOut": { + "description": "What a clip turned out to be, and the rate it is decomposed at.", + "properties": { + "codec": { + "title": "Codec", + "type": "string" + }, + "duration_seconds": { + "title": "Duration Seconds", + "type": "number" + }, + "extraction_fps": { + "title": "Extraction Fps", + "type": "number" + }, + "fps": { + "title": "Fps", + "type": "number" + }, + "height": { + "title": "Height", + "type": "integer" + }, + "width": { + "title": "Width", + "type": "integer" + } + }, + "required": [ + "width", + "height", + "fps", + "duration_seconds", + "codec", + "extraction_fps" + ], + "title": "VideoProvenanceOut", + "type": "object" + } + }, + "securitySchemes": { + "HTTPBearer": { + "description": "A workspace API token, created with `visionset token create`. Sent as `Authorization: Bearer `.", + "scheme": "bearer", + "type": "http" + } + } + }, + "info": { + "description": "REST surface of the VisionSet SDK. The committed openapi.json is the contract.", + "title": "Robomous VisionSet API", + "version": "0.0.1.dev0" + }, + "openapi": "3.1.0", + "paths": { + "/batches/{batch_id}/assets": { + "get": { + "description": "Everything in the batch, in membership order.\n\nThe order is stored, so reading twice gives the same sequence and an ingest\ninto an existing batch appends rather than reshuffles. An empty batch is a\n200 with an empty list, never a 404.\n\nBytes are not here: an asset is named by its `content_hash` and its\n`thumbnail_hash`, and downloading either is a later capability.", + "operationId": "list_batch_assets", + "parameters": [ + { + "in": "path", + "name": "batch_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Batch Id", + "type": "string" + } + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/AssetPage" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Missing or invalid bearer token" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "No such resource" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The request payload is not processable" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Unhandled server error, with an incident id" + }, + "503": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The workspace is busy; retry after the header says" + } + }, + "security": [ + { + "HTTPBearer": [] + } + ], + "summary": "List Batch Assets", + "tags": [ + "batches" + ] + } + }, + "/health": { + "get": { + "description": "Liveness probe. Public \u2014 no token required.", + "operationId": "health", + "responses": { "200": { "content": { "application/json": { @@ -376,20 +940,759 @@ }, "description": "The workspace is busy; retry after the header says" } - }, - "summary": "Health" - } - }, - "/projects": { - "get": { - "description": "Every project in this workspace, in the order they were created.", - "operationId": "list_projects", + }, + "summary": "Health" + } + }, + "/ingest-jobs/{job_id}": { + "get": { + "description": "Where a run is now.\n\n`processed` and `total` are written as the run goes, so this answers \"where\nis it\" rather than \"where did it end\". `total` is null for a clip \u2014 a video's\nframe count is a guess before extraction, so it is not reported.\n\nTerminal states are `completed` and `failed`. A `failed` job keeps its\ncounters exactly where they stopped, and `error` says why; unreadable\nindividual items are in `failures` and never fail a run on their own.", + "operationId": "get_ingest_job", + "parameters": [ + { + "in": "path", + "name": "job_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Job Id", + "type": "string" + } + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/IngestJobOut" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Missing or invalid bearer token" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "No such resource" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The request payload is not processable" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Unhandled server error, with an incident id" + }, + "503": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The workspace is busy; retry after the header says" + } + }, + "security": [ + { + "HTTPBearer": [] + } + ], + "summary": "Get Ingest Job", + "tags": [ + "ingest" + ] + } + }, + "/ingest-jobs/{job_id}/resume": { + "post": { + "description": "Run a failed job again, on the same row and into the same batch.\n\nA redo, not a skip: the whole source is read again. That creates nothing it\ncreated before \u2014 content is addressed by hash and assets are deduplicated \u2014\nso the cost is re-reading and the gain is that resume has no second code path.\n\nA `completed` job cannot be resumed, and neither can one stuck at `running`:\nthat is a process that died without reporting, so ingest the source again\ninstead, which creates nothing and leaves the stuck row as the record it is.\nBoth answer 409 `INVALID_TRANSITION`.", + "operationId": "resume_ingest", + "parameters": [ + { + "in": "path", + "name": "job_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Job Id", + "type": "string" + } + } + ], + "responses": { + "202": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/IngestJobOut" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Missing or invalid bearer token" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "No such resource" + }, + "409": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The resource's state refuses this request" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The request payload is not processable" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Unhandled server error, with an incident id" + }, + "503": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The workspace is busy; retry after the header says" + } + }, + "security": [ + { + "HTTPBearer": [] + } + ], + "summary": "Resume Ingest", + "tags": [ + "ingest" + ] + } + }, + "/projects": { + "get": { + "description": "Every project in this workspace, in the order they were created.", + "operationId": "list_projects", + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProjectPage" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Missing or invalid bearer token" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The request payload is not processable" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Unhandled server error, with an incident id" + }, + "503": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The workspace is busy; retry after the header says" + } + }, + "security": [ + { + "HTTPBearer": [] + } + ], + "summary": "List Projects", + "tags": [ + "projects" + ] + }, + "post": { + "description": "Add a project and its empty dataset, both or neither.", + "operationId": "create_project", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProjectCreate" + } + } + }, + "required": true + }, + "responses": { + "201": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProjectOut" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Missing or invalid bearer token" + }, + "409": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The resource's state refuses this request" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The request payload is not processable" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Unhandled server error, with an incident id" + }, + "503": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The workspace is busy; retry after the header says" + } + }, + "security": [ + { + "HTTPBearer": [] + } + ], + "summary": "Create Project", + "tags": [ + "projects" + ] + } + }, + "/projects/{project_id}": { + "delete": { + "description": "Remove a project and everything under it.\n\nMetadata only: content blobs are shared and are never deleted. Without\n`confirm=true` this answers 409 `CONFIRMATION_REQUIRED` and destroys nothing.", + "operationId": "delete_project", + "parameters": [ + { + "in": "path", + "name": "project_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Project Id", + "type": "string" + } + }, + { + "description": "Required to destroy data. The kernel refuses the request without it.", + "in": "query", + "name": "confirm", + "required": false, + "schema": { + "default": false, + "description": "Required to destroy data. The kernel refuses the request without it.", + "title": "Confirm", + "type": "boolean" + } + } + ], + "responses": { + "204": { + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Missing or invalid bearer token" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "No such resource" + }, + "409": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The resource's state refuses this request" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The request payload is not processable" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Unhandled server error, with an incident id" + }, + "503": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The workspace is busy; retry after the header says" + } + }, + "security": [ + { + "HTTPBearer": [] + } + ], + "summary": "Delete Project", + "tags": [ + "projects" + ] + }, + "get": { + "description": "The project with that id.", + "operationId": "get_project", + "parameters": [ + { + "in": "path", + "name": "project_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Project Id", + "type": "string" + } + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProjectOut" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Missing or invalid bearer token" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "No such resource" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The request payload is not processable" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Unhandled server error, with an incident id" + }, + "503": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The workspace is busy; retry after the header says" + } + }, + "security": [ + { + "HTTPBearer": [] + } + ], + "summary": "Get Project", + "tags": [ + "projects" + ] + }, + "patch": { + "description": "Rename a project, and its dataset with it. The only field that moves.", + "operationId": "rename_project", + "parameters": [ + { + "in": "path", + "name": "project_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Project Id", + "type": "string" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProjectRename" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProjectOut" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Missing or invalid bearer token" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "No such resource" + }, + "409": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The resource's state refuses this request" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The request payload is not processable" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Unhandled server error, with an incident id" + }, + "503": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The workspace is busy; retry after the header says" + } + }, + "security": [ + { + "HTTPBearer": [] + } + ], + "summary": "Rename Project", + "tags": [ + "projects" + ] + } + }, + "/projects/{project_id}/schema": { + "get": { + "description": "The version in force: the highest one.\n\nA project that has no schema yet answers 404 `SCHEMA_NOT_FOUND`, which is a\ndifferent code from the 404 `PROJECT_NOT_FOUND` an unknown project gets.\nSame status, two situations.", + "operationId": "get_active_schema", + "parameters": [ + { + "in": "path", + "name": "project_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Project Id", + "type": "string" + } + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SchemaVersionOut" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Missing or invalid bearer token" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "No such resource" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The request payload is not processable" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "Unhandled server error, with an incident id" + }, + "503": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "The workspace is busy; retry after the header says" + } + }, + "security": [ + { + "HTTPBearer": [] + } + ], + "summary": "Get Active Schema", + "tags": [ + "schemas" + ] + } + }, + "/projects/{project_id}/schema/versions": { + "get": { + "description": "Every version, oldest first. An empty page is the ordinary starting state.", + "operationId": "list_schema_versions", + "parameters": [ + { + "in": "path", + "name": "project_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Project Id", + "type": "string" + } + } + ], "responses": { "200": { "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/ProjectPage" + "$ref": "#/components/schemas/SchemaVersionPage" } } }, @@ -405,6 +1708,16 @@ }, "description": "Missing or invalid bearer token" }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "No such resource" + }, "422": { "content": { "application/json": { @@ -441,19 +1754,43 @@ "HTTPBearer": [] } ], - "summary": "List Projects", + "summary": "List Schema Versions", "tags": [ - "projects" + "schemas" ] }, "post": { - "description": "Add a project and its empty dataset, both or neither.", - "operationId": "create_project", + "description": "Append the next version of the project's schema.\n\nThe body is the whole proposed version; versions are never edited in place.\n\nRemoving a class or an attribute answers 409 `DESTRUCTIVE_SCHEMA_CHANGE`\nuntil `allow_destructive=true` says so deliberately. If annotations already\nexist under an affected class it answers 409 `SCHEMA_CHANGE_WOULD_ORPHAN`\ninstead, and **no flag overrides that one** \u2014 which is why a client branches\non `code` and not on the status.", + "operationId": "create_schema_version", + "parameters": [ + { + "in": "path", + "name": "project_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Project Id", + "type": "string" + } + }, + { + "description": "Required when the new version narrows the labeling contract.", + "in": "query", + "name": "allow_destructive", + "required": false, + "schema": { + "default": false, + "description": "Required when the new version narrows the labeling contract.", + "title": "Allow Destructive", + "type": "boolean" + } + } + ], "requestBody": { "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/ProjectCreate" + "$ref": "#/components/schemas/SchemaVersionCreate" } } }, @@ -464,7 +1801,7 @@ "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/ProjectOut" + "$ref": "#/components/schemas/SchemaVersionOut" } } }, @@ -480,6 +1817,16 @@ }, "description": "Missing or invalid bearer token" }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorBody" + } + } + }, + "description": "No such resource" + }, "409": { "content": { "application/json": { @@ -526,16 +1873,16 @@ "HTTPBearer": [] } ], - "summary": "Create Project", + "summary": "Create Schema Version", "tags": [ - "projects" + "schemas" ] } }, - "/projects/{project_id}": { - "delete": { - "description": "Remove a project and everything under it.\n\nMetadata only: content blobs are shared and are never deleted. Without\n`confirm=true` this answers 409 `CONFIRMATION_REQUIRED` and destroys nothing.", - "operationId": "delete_project", + "/projects/{project_id}/schema/versions/{version}": { + "get": { + "description": "One version of a project's schema.", + "operationId": "get_schema_version", "parameters": [ { "in": "path", @@ -548,33 +1895,30 @@ } }, { - "description": "Required to destroy data. The kernel refuses the request without it.", - "in": "query", - "name": "confirm", - "required": false, + "description": "A schema version, 1..N.", + "in": "path", + "name": "version", + "required": true, "schema": { - "default": false, - "description": "Required to destroy data. The kernel refuses the request without it.", - "title": "Confirm", - "type": "boolean" + "description": "A schema version, 1..N.", + "minimum": 1, + "title": "Version", + "type": "integer" } } ], "responses": { - "204": { - "description": "Successful Response" - }, - "401": { + "200": { "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/ErrorBody" + "$ref": "#/components/schemas/SchemaVersionOut" } } }, - "description": "Missing or invalid bearer token" + "description": "Successful Response" }, - "404": { + "401": { "content": { "application/json": { "schema": { @@ -582,9 +1926,9 @@ } } }, - "description": "No such resource" + "description": "Missing or invalid bearer token" }, - "409": { + "404": { "content": { "application/json": { "schema": { @@ -592,7 +1936,7 @@ } } }, - "description": "The resource's state refuses this request" + "description": "No such resource" }, "422": { "content": { @@ -630,14 +1974,16 @@ "HTTPBearer": [] } ], - "summary": "Delete Project", + "summary": "Get Schema Version", "tags": [ - "projects" + "schemas" ] - }, + } + }, + "/projects/{project_id}/sources": { "get": { - "description": "The project with that id.", - "operationId": "get_project", + "description": "Every source of that project, in registration order.", + "operationId": "list_sources", "parameters": [ { "in": "path", @@ -655,7 +2001,7 @@ "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/ProjectOut" + "$ref": "#/components/schemas/SourcePage" } } }, @@ -717,14 +2063,16 @@ "HTTPBearer": [] } ], - "summary": "Get Project", + "summary": "List Sources", "tags": [ - "projects" + "sources" ] - }, - "patch": { - "description": "Rename a project, and its dataset with it. The only field that moves.", - "operationId": "rename_project", + } + }, + "/projects/{project_id}/sources/images": { + "post": { + "description": "Offer a project a folder of stills.\n\nThe parts are staged as one directory and that directory becomes the source.\nUploading the same files again returns the **same** source rather than a\nsecond one: staging is content-addressed, so identical bytes under identical\nfilenames land on the same path, and registration is idempotent on that path.\n\nNothing is decoded here \u2014 what the files turn out to be is read at ingest,\nand a file that is not an image is reported there rather than refused now.", + "operationId": "register_image_source", "parameters": [ { "in": "path", @@ -739,20 +2087,20 @@ ], "requestBody": { "content": { - "application/json": { + "multipart/form-data": { "schema": { - "$ref": "#/components/schemas/ProjectRename" + "$ref": "#/components/schemas/Body_register_image_source" } } }, "required": true }, "responses": { - "200": { + "201": { "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/ProjectOut" + "$ref": "#/components/schemas/SourceOut" } } }, @@ -778,16 +2126,6 @@ }, "description": "No such resource" }, - "409": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ErrorBody" - } - } - }, - "description": "The resource's state refuses this request" - }, "422": { "content": { "application/json": { @@ -824,16 +2162,16 @@ "HTTPBearer": [] } ], - "summary": "Rename Project", + "summary": "Register Image Source", "tags": [ - "projects" + "sources" ] } }, - "/projects/{project_id}/schema": { - "get": { - "description": "The version in force: the highest one.\n\nA project that has no schema yet answers 404 `SCHEMA_NOT_FOUND`, which is a\ndifferent code from the 404 `PROJECT_NOT_FOUND` an unknown project gets.\nSame status, two situations.", - "operationId": "get_active_schema", + "/projects/{project_id}/sources/video": { + "post": { + "description": "Offer a project a clip, to be cut at `extraction_fps`.\n\nThe clip is probed on the way in, so a file that is not a video, or one\nwhose bytes will not decode, is 422 here rather than a run that fails later.\n\nThe rate is part of what the source *is*: the same clip registered at 1 fps\nand again at 5 fps is two sources over one file, which is what makes \"the\nsame source yields the same assets\" mean anything.", + "operationId": "register_video_source", "parameters": [ { "in": "path", @@ -846,12 +2184,22 @@ } } ], + "requestBody": { + "content": { + "multipart/form-data": { + "schema": { + "$ref": "#/components/schemas/Body_register_video_source" + } + } + }, + "required": true + }, "responses": { - "200": { + "201": { "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/SchemaVersionOut" + "$ref": "#/components/schemas/SourceOut" } } }, @@ -913,24 +2261,24 @@ "HTTPBearer": [] } ], - "summary": "Get Active Schema", + "summary": "Register Video Source", "tags": [ - "schemas" + "sources" ] } }, - "/projects/{project_id}/schema/versions": { + "/sources/{source_id}": { "get": { - "description": "Every version, oldest first. An empty page is the ordinary starting state.", - "operationId": "list_schema_versions", + "description": "The source with that id.", + "operationId": "get_source", "parameters": [ { "in": "path", - "name": "project_id", + "name": "source_id", "required": true, "schema": { "format": "uuid", - "title": "Project Id", + "title": "Source Id", "type": "string" } } @@ -940,7 +2288,7 @@ "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/SchemaVersionPage" + "$ref": "#/components/schemas/SourceOut" } } }, @@ -1002,54 +2350,34 @@ "HTTPBearer": [] } ], - "summary": "List Schema Versions", + "summary": "Get Source", "tags": [ - "schemas" + "sources" ] - }, - "post": { - "description": "Append the next version of the project's schema.\n\nThe body is the whole proposed version; versions are never edited in place.\n\nRemoving a class or an attribute answers 409 `DESTRUCTIVE_SCHEMA_CHANGE`\nuntil `allow_destructive=true` says so deliberately. If annotations already\nexist under an affected class it answers 409 `SCHEMA_CHANGE_WOULD_ORPHAN`\ninstead, and **no flag overrides that one** \u2014 which is why a client branches\non `code` and not on the status.", - "operationId": "create_schema_version", + } + }, + "/sources/{source_id}/ingest-jobs": { + "get": { + "description": "Every run of that source, in the order they were asked for.", + "operationId": "list_ingest_jobs", "parameters": [ { "in": "path", - "name": "project_id", + "name": "source_id", "required": true, "schema": { "format": "uuid", - "title": "Project Id", + "title": "Source Id", "type": "string" } - }, - { - "description": "Required when the new version narrows the labeling contract.", - "in": "query", - "name": "allow_destructive", - "required": false, - "schema": { - "default": false, - "description": "Required when the new version narrows the labeling contract.", - "title": "Allow Destructive", - "type": "boolean" - } } ], - "requestBody": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/SchemaVersionCreate" - } - } - }, - "required": true - }, "responses": { - "201": { + "200": { "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/SchemaVersionOut" + "$ref": "#/components/schemas/IngestJobPage" } } }, @@ -1075,16 +2403,6 @@ }, "description": "No such resource" }, - "409": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ErrorBody" - } - } - }, - "description": "The resource's state refuses this request" - }, "422": { "content": { "application/json": { @@ -1121,46 +2439,49 @@ "HTTPBearer": [] } ], - "summary": "Create Schema Version", + "summary": "List Ingest Jobs", "tags": [ - "schemas" + "sources" ] - } - }, - "/projects/{project_id}/schema/versions/{version}": { - "get": { - "description": "One version of a project's schema.", - "operationId": "get_schema_version", + }, + "post": { + "description": "Launch a run over the source and answer at once with the job to poll.\n\n**202, not 201**: the row exists, the work does not. Poll\n`GET /ingest-jobs/{id}` \u2014 the `Location` header names it \u2014 and watch\n`processed` climb until `state` is `completed` or `failed`.\n\nA run that could not even be recorded is refused here; everything that goes\nwrong afterwards is reported *on the job*, which is the whole point of the\nshape. Unreadable files land in `failures` and do not fail the run; a\nmissing ffmpeg does, in `error`.", + "operationId": "start_ingest", "parameters": [ { "in": "path", - "name": "project_id", + "name": "source_id", "required": true, "schema": { "format": "uuid", - "title": "Project Id", + "title": "Source Id", "type": "string" } - }, - { - "description": "A schema version, 1..N.", - "in": "path", - "name": "version", - "required": true, - "schema": { - "description": "A schema version, 1..N.", - "minimum": 1, - "title": "Version", - "type": "integer" - } } ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "anyOf": [ + { + "$ref": "#/components/schemas/IngestStart" + }, + { + "type": "null" + } + ], + "title": "Body" + } + } + } + }, "responses": { - "200": { + "202": { "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/SchemaVersionOut" + "$ref": "#/components/schemas/IngestJobOut" } } }, @@ -1222,9 +2543,9 @@ "HTTPBearer": [] } ], - "summary": "Get Schema Version", + "summary": "Start Ingest", "tags": [ - "schemas" + "sources" ] } } diff --git a/pyproject.toml b/pyproject.toml index 1bec4948..8b6e7740 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,6 +30,12 @@ dependencies = [ # so it is allowed inside kernel/adapters/ and needs no import-linter change. The 11.0 # floor is the first release with Python 3.13 wheels, which the classifiers promise. "pillow>=11.0", + # Multipart parsing for the upload routes (#28). FastAPI requires it for + # `UploadFile` and raises at import time without it. It is already installed + # here as a transitive dependency of `mcp` — which is exactly why it is + # declared: an upload surface resting on somebody else's dependency tree + # breaks the day that tree changes. + "python-multipart>=0.0.9", ] [project.urls] diff --git a/src/visionset/kernel/services/batch_service.py b/src/visionset/kernel/services/batch_service.py index 7228b168..bd45a13b 100644 --- a/src/visionset/kernel/services/batch_service.py +++ b/src/visionset/kernel/services/batch_service.py @@ -38,6 +38,7 @@ BATCH_TRANSITIONS, AnnotationJob, AnnotationJobState, + Asset, AssetProgress, Batch, BatchApproved, @@ -59,6 +60,7 @@ ConfirmationRequired, EmptyBatch, ProjectNotFound, + WorkspaceCorrupt, ) from visionset.kernel.ports import UnitOfWork from visionset.kernel.services.schema_service import SchemaService @@ -98,6 +100,22 @@ def jobs(self, batch_id: UUID) -> list[AnnotationJob]: with self._workspace.unit_of_work() as uow: return jobs_of(uow, self.require_batch(uow, batch_id)) + def assets(self, batch_id: UUID) -> list[Asset]: + """Everything in the batch, in membership order. + + The read behind "what did that ingest actually gather" — membership + order is the stored ``batch_asset.position``, so a caller reading the + batch twice sees the same sequence and an ``add_assets`` appends rather + than reshuffles. ``DatasetService.assets`` is the same method over the + trunk, and answers to the same rule about a member that is not there. + + Raises: + BatchNotFound: no such batch in this workspace. + WorkspaceCorrupt: the batch holds an asset that is not stored. + """ + with self._workspace.unit_of_work() as uow: + return assets_of(uow, self.require_batch(uow, batch_id)) + # ``list`` shadows the builtin for every annotation after it in this class # body, so it comes last here and the helpers that need ``list[...]`` live # at module level. @@ -367,6 +385,27 @@ def _subject(batch: Batch) -> str: return f"batch {batch.name!r}" +def assets_of(uow: UnitOfWork, batch: Batch) -> list[Asset]: + """Every asset in the batch, in membership order. + + Module-level and public beside ``jobs_of``, for the same reason: the read is + one line and the *rule about a member that is not stored* is the part worth + having one copy of. ``batch_asset.asset_id`` cascades from ``asset``, so a + deleted asset takes its membership row with it and this cannot happen while + foreign keys are on — dropping the id quietly would turn that guarantee + failing into a batch that silently holds less than it says. + """ + assets = [] + for asset_id in batch.asset_ids: + asset = uow.assets.get(asset_id) + if asset is None: + raise WorkspaceCorrupt( + f"batch {batch.name!r} holds asset {asset_id}, which is not stored" + ) + assets.append(asset) + return assets + + def jobs_of(uow: UnitOfWork, batch: Batch) -> list[AnnotationJob]: """Every job under the batch, task group by task group, in segment order. diff --git a/src/visionset/kernel/services/ingest_service.py b/src/visionset/kernel/services/ingest_service.py index 5a8c9b9d..efd793e7 100644 --- a/src/visionset/kernel/services/ingest_service.py +++ b/src/visionset/kernel/services/ingest_service.py @@ -62,11 +62,14 @@ four-transaction discipline applied to a different job — read the ids, render outside any transaction, write once at the end. -**What is still deliberately not here.** No background execution: a run is -synchronous and in-process, and the API is shaped -so that putting it behind a queue changes the caller's waiting rather than its -vocabulary — which is why a job is created ``pending`` and moved to ``running`` -by whoever picks it up, even though today that is the same call. +**Asking for a run and doing it are two calls.** ``enqueue`` refuses everything +refusable and returns a ``pending`` job; ``resume`` picks that job up and does +the work; ``ingest`` is the two composed, for a caller that can wait. The split +is what the ``pending`` state was reserved for from the start, and it is what +lets the HTTP surface hand back a job id before the first byte is read. **What +is still not here is a scheduler**: nothing in this module decides *when* the +second half runs, and a caller that wants it off the calling thread supplies +that itself. """ from __future__ import annotations @@ -131,6 +134,45 @@ def get(self, job_id: UUID) -> IngestJob: # --- ingesting --------------------------------------------------------- + def enqueue( + self, + source_id: UUID, + *, + batch_id: UUID | None = None, + batch_name: str | None = None, + ) -> IngestJob: + """Record that a run was asked for, and refuse it now if it cannot happen. + + Everything :meth:`ingest` can refuse before reading a byte is refused + here — an unknown source, a project this workspace does not have, a + frozen target batch, a blank name — so a run that fails fast leaves **no + job row at all** and a caller that got a row got one it can poll. + + What comes back is ``pending``. Whoever picks the work up moves it to + ``running``, which today is :meth:`resume`; that vocabulary was written + into ``INGEST_TRANSITIONS`` from the start for a queue that did not exist + yet, and this is the half of it that was missing. A caller wanting the + whole run in one call still uses :meth:`ingest`. + + ``batch_id`` is stored on the row rather than held by the caller, + because between here and the run there is no caller to hold it: the row + is the only thing that crosses. :meth:`resume` already reads it as "the + batch this attempt was headed for". + + Raises: + SourceNotFound: no such source in this workspace. + BatchNotFound: ``batch_id`` names no batch in this workspace. + BatchNotEditable: the target batch is past ``draft``. + InvalidName: ``batch_name`` is blank once stripped. + """ + with self._workspace.unit_of_work() as uow: + source = self._sources.require_source(uow, source_id) + self._require_project(uow, source.project_id) + name = self._target_name(uow, source, batch_id, batch_name) + return uow.ingest_jobs.add( + IngestJob(source_id=source.id, batch_id=batch_id, batch_name=name) + ) + def ingest( self, source_id: UUID, @@ -171,17 +213,12 @@ def ingest( job records it and is marked failed before it is re-raised — one broken machine is not five thousand broken files. """ - with self._workspace.unit_of_work() as uow: - source = self._sources.require_source(uow, source_id) - self._require_project(uow, source.project_id) - name = self._target_name(uow, source, batch_id, batch_name) - # ``pending``, not ``running``: the row exists before anybody picks - # the work up, which is the vocabulary a queue will need and costs - # nothing today. Every refusal above happens before the insert, so a - # run that fails fast leaves no job row at all. - job = uow.ingest_jobs.add(IngestJob(source_id=source.id, batch_name=name)) - - return self._run(job.id, source, name, batch_id) + # Enqueue then pick it straight back up. The two halves are spelled + # separately because a caller that cannot wait — the HTTP surface, a + # queue — needs the row before the work, and one composed call is how + # this one keeps having no second code path to get wrong. + job = self.enqueue(source_id, batch_id=batch_id, batch_name=batch_name) + return self.resume(job.id) def resume(self, job_id: UUID) -> IngestResult: """Run a failed job again, on the same row and into the same batch. @@ -198,8 +235,9 @@ def resume(self, job_id: UUID) -> IngestResult: would be a lie. The fatal ``error`` is cleared for the same reason. What may be resumed is whatever ``INGEST_TRANSITIONS`` says can reach - ``running``: a ``failed`` job, and a ``pending`` one, which a synchronous - run never leaves behind but a queued one would. A ``completed`` job + ``running``: a ``failed`` job, and a ``pending`` one — which is what + :meth:`enqueue` leaves, so this is also how a run is *started* by + whoever picked it up. A ``completed`` job cannot, and neither can one stuck at ``running`` — that is a process that died without reporting anything, so ingest the source again instead, which creates nothing and leaves the crashed row as the record it is. @@ -214,17 +252,46 @@ def resume(self, job_id: UUID) -> IngestResult: plus everything :meth:`ingest` raises. """ with self._workspace.unit_of_work() as uow: - job = self.require_job(uow, job_id) - source = self._sources.require_source(uow, job.source_id) - self._require_project(uow, source.project_id) - # The friendly pre-check, so a completed job is refused before the - # target batch is resolved. The real one is inside ``_run``, in the - # transaction that actually moves the row. - require_move(INGEST_TRANSITIONS, job.state, IngestState.RUNNING, _subject(job.id)) + job, source = self._resolve_for_run(uow, job_id) name = self._target_name(uow, source, job.batch_id, job.batch_name) return self._run(job.id, source, name, job.batch_id) + def resumable(self, job_id: UUID) -> IngestJob: + """The job, if :meth:`resume` would take it — otherwise refuse now. + + The same refusals :meth:`resume` makes before it reads anything, without + the reading. A caller that runs the work somewhere else needs them + *here*, on the calling thread: a launch that answered "accepted" and then + discovered in a worker that the job was already ``completed`` would give + a client no way to tell a redo from a no-op. + + It does not move the row. What it reports is that the move is legal at + this moment; ``_begin`` inside the run is still the one that makes it. + + Raises: + IngestJobNotFound: no such ingest job in this workspace. + InvalidTransition: the job is ``completed``, or stuck at ``running``. + SourceNotFound: the source has since been deleted. + """ + with self._workspace.unit_of_work() as uow: + job, _ = self._resolve_for_run(uow, job_id) + return job + + def _resolve_for_run(self, uow: UnitOfWork, job_id: UUID) -> tuple[IngestJob, Source]: + """The job and its source, once this workspace agrees it may run again. + + One spelling of the friendly pre-check, shared by the method that does + the work and the one that only asks. The *real* check is inside ``_run``, + in the transaction that actually moves the row — this one exists so a + refusal arrives before a target batch is resolved or a file is opened. + """ + job = self.require_job(uow, job_id) + source = self._sources.require_source(uow, job.source_id) + self._require_project(uow, source.project_id) + require_move(INGEST_TRANSITIONS, job.state, IngestState.RUNNING, _subject(job.id)) + return job, source + # --- the thumbnail cache ----------------------------------------------- def backfill_thumbnails(self, project_id: UUID) -> ThumbnailBackfill: diff --git a/src/visionset/server/dependencies.py b/src/visionset/server/dependencies.py index 77cbad46..e46fd72f 100644 --- a/src/visionset/server/dependencies.py +++ b/src/visionset/server/dependencies.py @@ -38,6 +38,7 @@ resolve_workspace_root as resolve_workspace_root, ) from visionset.server.errors import ERROR_RESPONSES +from visionset.server.runner import IngestRunner # ``WORKSPACE_ENV_VAR`` and ``resolve_workspace_root`` are re-exported above # rather than defined here — the redundant ``as`` aliases are the explicit @@ -148,6 +149,18 @@ def get_workspace(request: Request) -> WorkspaceService: return handle.get() +def get_ingest_runner(request: Request) -> IngestRunner: + """The background worker this application launches ingests on. + + Read off ``app.state`` and reached through a dependency rather than by + routes touching ``request.app`` themselves, for the reason + :func:`get_auth_provider` is its own dependency: this is the seam a test + replaces, and ``dependency_overrides`` only reaches what the graph resolves. + """ + runner: IngestRunner = request.app.state.ingest_runner + return runner + + def get_auth_provider( workspace: Annotated[WorkspaceService, Depends(get_workspace)], ) -> AuthProvider: @@ -185,6 +198,9 @@ def require_token( WorkspaceDep = Annotated[WorkspaceService, Depends(get_workspace)] """The workspace, for a route that needs to build a service over it.""" +RunnerDep = Annotated[IngestRunner, Depends(get_ingest_runner)] +"""The background worker, for a route that launches a run rather than doing it.""" + TokenDep = Annotated[str, Depends(require_token)] """The presented token, for the rare route that needs the credential itself. diff --git a/src/visionset/server/main.py b/src/visionset/server/main.py index 3a1cbdb4..b8067c38 100644 --- a/src/visionset/server/main.py +++ b/src/visionset/server/main.py @@ -12,6 +12,7 @@ from visionset.server.dependencies import WorkspaceHandle from visionset.server.errors import UNIVERSAL_ERROR_RESPONSES, install_error_handlers from visionset.server.routes import ROUTERS +from visionset.server.runner import IngestRunner DESCRIPTION = "REST surface of the VisionSet SDK. The committed openapi.json is the contract." @@ -40,14 +41,18 @@ async def health() -> dict[str, str]: @asynccontextmanager async def _lifespan(app: FastAPI) -> AsyncIterator[None]: - """Close the workspace on shutdown, if a request ever opened one. + """Stop the ingest worker and close the workspace, in that order. - Only the closing half. The handle itself is built in :func:`create_app`, - because ``TestClient(app)`` used without its context manager never runs - startup — and a handle that only existed after startup would simply be - missing there. + Only the closing half. Both objects are built in :func:`create_app`, because + ``TestClient(app)`` used without its context manager never runs startup — + and anything that only existed after startup would simply be missing there. + + The order is load-bearing: a run still in flight holds the workspace, so + closing it first would pull the store out from under a worker mid-write. """ yield + runner: IngestRunner = app.state.ingest_runner + runner.shutdown() handle: WorkspaceHandle = app.state.workspace_handle handle.close() @@ -87,6 +92,10 @@ def create_app() -> FastAPI: generate_unique_id_function=operation_id, ) app.state.workspace_handle = WorkspaceHandle() + # Beside the handle and for the same reason — one per application, so two + # apps in one pytest process never share a worker. Neither touches disk or + # starts a thread until a request asks it to. + app.state.ingest_runner = IngestRunner() install_error_handlers(app) app.include_router(router) for resource in ROUTERS: diff --git a/src/visionset/server/models.py b/src/visionset/server/models.py index d50a24e7..59025551 100644 --- a/src/visionset/server/models.py +++ b/src/visionset/server/models.py @@ -37,6 +37,8 @@ from __future__ import annotations +from datetime import datetime +from pathlib import Path from typing import Annotated, Literal, Self from uuid import UUID @@ -45,10 +47,19 @@ from visionset.kernel.domain import ( AnnotationSchema, + Asset, Attribute, GeometryType, + ImageFormat, + IngestFailure, + IngestFailureKind, + IngestJob, + IngestState, LabelClass, Project, + Source, + SourceKind, + VideoProvenance, ) # A gate is a query parameter and never a body field, so a client that gets a 409 @@ -256,3 +267,179 @@ class SchemaVersionCreate(BaseModel): model_config = ConfigDict(extra="forbid") classes: tuple[LabelClassBody, ...] = () + + +# --- sources ----------------------------------------------------------------- + + +# Flattened rather than nesting the domain's own ``VideoMetadata``, which would +# publish a kernel model under its kernel docstring — the module rule above. The +# two rates are the reason this type exists at all: ``fps`` is what the file was +# *shot* at and ``extraction_fps`` is what we chose to *cut* it at, and a client +# that confuses them decomposes at the wrong rate. See ``docs/sources.md``. +class VideoProvenanceOut(BaseModel): + """What a clip turned out to be, and the rate it is decomposed at.""" + + width: int + height: int + fps: float + duration_seconds: float + codec: str + extraction_fps: float + + @classmethod + def of(cls, provenance: VideoProvenance) -> Self: + return cls( + width=provenance.metadata.width, + height=provenance.metadata.height, + fps=provenance.metadata.fps, + duration_seconds=provenance.metadata.duration_seconds, + codec=provenance.metadata.codec, + extraction_fps=provenance.extraction_fps, + ) + + +# ``Source.path`` is deliberately absent, and this is the one omission worth +# stating twice. It is an absolute path on the server's own filesystem, inside +# the workspace's staging area — a client can do nothing with it, and publishing +# it hands every token holder the layout of the machine. ``name`` is the part a +# client recognises: the filename it uploaded. +class SourceOut(BaseModel): + """A registered origin: a folder of stills, or a clip.""" + + id: UUID + project_id: UUID + kind: SourceKind + name: str + registered_at: datetime + video: VideoProvenanceOut | None + + @classmethod + def of(cls, source: Source) -> Self: + return cls( + id=source.id, + project_id=source.project_id, + kind=source.kind, + name=Path(source.path).name, + registered_at=source.registered_at, + video=None if source.video is None else VideoProvenanceOut.of(source.video), + ) + + +class SourcePage(Page[SourceOut]): + """A page of sources.""" + + +# --- ingest ------------------------------------------------------------------ + + +class IngestFailureOut(BaseModel): + """One item a run could not read, and why.""" + + name: str + kind: IngestFailureKind + reason: str + + @classmethod + def of(cls, failure: IngestFailure) -> Self: + return cls(name=failure.name, kind=failure.kind, reason=failure.reason) + + +# The polling contract. ``processed``/``total``/``failures`` are written to the +# row as the run goes, so this says where a run *is* rather than where it ended; +# ``total`` is null for a clip, because ``VideoMetadata`` carries no frame count +# by design and a guess is worse than an honest absence. ``error`` is the fatal +# cause and is a different field from ``failures`` on purpose — one broken +# machine is not five thousand broken files. +class IngestJobOut(BaseModel): + """One run of one source, and how far it has got.""" + + id: UUID + source_id: UUID + state: IngestState + error: str | None + batch_id: UUID | None + batch_name: str | None + processed: int + total: int | None + failures: tuple[IngestFailureOut, ...] + + @classmethod + def of(cls, job: IngestJob) -> Self: + return cls( + id=job.id, + source_id=job.source_id, + state=job.state, + error=job.error, + batch_id=job.batch_id, + batch_name=job.batch_name, + processed=job.processed, + total=job.total, + failures=tuple(IngestFailureOut.of(failure) for failure in job.failures), + ) + + +class IngestJobPage(Page[IngestJobOut]): + """A page of ingest jobs.""" + + +# Targeting an existing draft batch by id is deliberately not here. Batches have +# no endpoints until #29, so there is nothing for a client to name — and leaving +# it out is what keeps this launch free of any refusal that would leave the +# caller a 202 pointing at a job row that was never written. +# +# And there is deliberately **no** ``_the_domain_accepts_it`` validator, which is +# the interesting half. ``LabelClassBody`` needs one because ``LabelClass`` +# refuses with a *pydantic* ``ValidationError``, which is neither a +# ``VisionSetError`` nor a ``RequestValidationError`` and so answers 500. A blank +# batch name refuses with ``InvalidName`` — a domain error, already in +# ``ERROR_RULES`` at 422 ``INVALID_NAME`` — so the kernel's own refusal arrives +# correctly on its own and a validator here would only restate it, less precisely. +class IngestStart(BaseModel): + """What launching a run needs, which is almost nothing.""" + + model_config = ConfigDict(extra="forbid") + + batch_name: str | None = None + + +# --- assets ------------------------------------------------------------------ + + +# ``Asset.uri`` is absent for the reason ``Source.path`` is: it is a server-side +# path, and for a frame it is that path plus ``#frame=N``. Reaching the bytes is +# #30's blob download, keyed by the hashes already on this model. +class AssetOut(BaseModel): + """One ingested item.""" + + id: UUID + project_id: UUID + modality: Literal["image"] + content_hash: str + width: int | None + height: int | None + format: ImageFormat | None + source_id: UUID | None + frame_index: int | None + frame_timestamp: float | None + thumbnail_hash: str | None + + @classmethod + def of(cls, asset: Asset) -> Self: + return cls( + id=asset.id, + project_id=asset.project_id, + modality=asset.modality, + content_hash=asset.content_hash, + width=asset.width, + height=asset.height, + format=asset.format, + source_id=asset.source_id, + frame_index=asset.frame_index, + frame_timestamp=asset.frame_timestamp, + thumbnail_hash=asset.thumbnail_hash, + ) + + +class AssetPage(Page[AssetOut]): + """A page of assets.""" diff --git a/src/visionset/server/routes/__init__.py b/src/visionset/server/routes/__init__.py index acec081e..a207f43d 100644 --- a/src/visionset/server/routes/__init__.py +++ b/src/visionset/server/routes/__init__.py @@ -18,8 +18,17 @@ from fastapi import APIRouter -from visionset.server.routes import projects, schemas - -ROUTERS: Final[tuple[APIRouter, ...]] = (projects.router, schemas.router) - -__all__ = ["ROUTERS", "projects", "schemas"] +from visionset.server.routes import batches, ingest, projects, schemas, sources + +# A module may contribute more than one router: ``sources`` has a collection +# under its owning project and a resource of its own, which is two prefixes. +ROUTERS: Final[tuple[APIRouter, ...]] = ( + projects.router, + schemas.router, + sources.project_router, + sources.router, + ingest.router, + batches.router, +) + +__all__ = ["ROUTERS", "batches", "ingest", "projects", "schemas", "sources"] diff --git a/src/visionset/server/routes/batches.py b/src/visionset/server/routes/batches.py new file mode 100644 index 00000000..8801dadf --- /dev/null +++ b/src/visionset/server/routes/batches.py @@ -0,0 +1,43 @@ +# usage: from visionset.server.routes import batches +"""Batches: one route today, and it is the one an ingest needs. + +An ingest puts what it gathered into a batch, so "what did that run produce?" +is answered here rather than on the job — the batch is where the membership +actually lives, and a second door onto the same rows would be a second thing to +keep in step. + +**#29 owns this module from here on**: batch listing, detail with per-state +counts, approval with a partition spec, and paging plus per-asset progress on +the listing below. The envelope is what makes that additive — `total` already +means *matching the query* rather than *in this page*, so `limit` and `offset` +arrive beside it without breaking a client that parsed this shape today. See +``docs/api.md``. + +Handlers are ``def``, not ``async def``, for the reason ``projects.py`` gives. +""" + +from __future__ import annotations + +from uuid import UUID + +from visionset.kernel.services import BatchService +from visionset.server.dependencies import WorkspaceDep, protected_router +from visionset.server.errors import documented +from visionset.server.models import AssetOut, AssetPage + +router = protected_router(prefix="/batches", tags=["batches"]) + + +@router.get("/{batch_id}/assets", responses=documented(404)) +def list_batch_assets(workspace: WorkspaceDep, batch_id: UUID) -> AssetPage: + """Everything in the batch, in membership order. + + The order is stored, so reading twice gives the same sequence and an ingest + into an existing batch appends rather than reshuffles. An empty batch is a + 200 with an empty list, never a 404. + + Bytes are not here: an asset is named by its `content_hash` and its + `thumbnail_hash`, and downloading either is a later capability. + """ + found = BatchService(workspace).assets(batch_id) + return AssetPage(items=[AssetOut.of(asset) for asset in found], total=len(found)) diff --git a/src/visionset/server/routes/ingest.py b/src/visionset/server/routes/ingest.py new file mode 100644 index 00000000..caae5eb4 --- /dev/null +++ b/src/visionset/server/routes/ingest.py @@ -0,0 +1,70 @@ +# usage: from visionset.server.routes import ingest +"""Ingest jobs: the polling half of the launch-and-poll contract. + +A job is addressed on its own rather than under the source that produced it, +because that is how a client reaches it: the launch handed back an id and a +`Location`, and nothing else about the run is needed to ask after it. Listing a +source's runs stays with the source, in ``routes/sources.py``. + +Handlers are ``def``, not ``async def``, for the reason ``projects.py`` gives. +""" + +from __future__ import annotations + +from uuid import UUID + +from fastapi import Response, status + +from visionset.kernel.services import IngestService +from visionset.server.dependencies import RunnerDep, WorkspaceDep, protected_router +from visionset.server.errors import documented +from visionset.server.models import IngestJobOut + +router = protected_router(prefix="/ingest-jobs", tags=["ingest"]) + + +@router.get("/{job_id}", responses=documented(404)) +def get_ingest_job(workspace: WorkspaceDep, job_id: UUID) -> IngestJobOut: + """Where a run is now. + + `processed` and `total` are written as the run goes, so this answers "where + is it" rather than "where did it end". `total` is null for a clip — a video's + frame count is a guess before extraction, so it is not reported. + + Terminal states are `completed` and `failed`. A `failed` job keeps its + counters exactly where they stopped, and `error` says why; unreadable + individual items are in `failures` and never fail a run on their own. + """ + return IngestJobOut.of(IngestService(workspace).get(job_id)) + + +@router.post( + "/{job_id}/resume", + status_code=status.HTTP_202_ACCEPTED, + responses=documented(404, 409), +) +def resume_ingest( + workspace: WorkspaceDep, + runner: RunnerDep, + response: Response, + job_id: UUID, +) -> IngestJobOut: + """Run a failed job again, on the same row and into the same batch. + + A redo, not a skip: the whole source is read again. That creates nothing it + created before — content is addressed by hash and assets are deduplicated — + so the cost is re-reading and the gain is that resume has no second code path. + + A `completed` job cannot be resumed, and neither can one stuck at `running`: + that is a process that died without reporting, so ingest the source again + instead, which creates nothing and leaves the stuck row as the record it is. + Both answer 409 `INVALID_TRANSITION`. + """ + ingest = IngestService(workspace) + # ``resumable``, not ``get``: a completed job must be 409 *here*, because a + # 202 followed by a refusal only the worker ever saw gives a client no way + # to tell a redo from a no-op. It reads the same table the run will. + job = ingest.resumable(job_id) + runner.submit(lambda: ingest.resume(job.id)) + response.headers["Location"] = f"/ingest-jobs/{job.id}" + return IngestJobOut.of(job) diff --git a/src/visionset/server/routes/sources.py b/src/visionset/server/routes/sources.py new file mode 100644 index 00000000..ac7d24d2 --- /dev/null +++ b/src/visionset/server/routes/sources.py @@ -0,0 +1,152 @@ +# usage: from visionset.server.routes import sources +"""Sources: offering a project some raw data, and launching a run over it. + +Two routers, because a source is addressable on its own. The collection hangs +off the project that owns it (`ProjectService` is the door to a project, and a +source belongs to exactly one); the resource does not, because what hangs off +*it* — its ingest jobs — would otherwise sit four path segments deep for no gain. + +**Registration is upload-only.** The kernel registers a source by path, so these +routes stage the bytes first (see ``server/uploads.py``) and register the staged +directory or file. There is no route that takes a server-side path: it would +hand every token holder an arbitrary-directory read, and the two surfaces that +legitimately hold real paths — the CLI and MCP — call the SDK in-process and +never come through here. It also has a quiet dividend: because the server just +wrote the file, `SourceService`'s ``FileNotFoundError`` and ``NotADirectoryError`` +are unreachable, and those are plain Python exceptions with no place in +``ERROR_RULES``. + +Handlers are ``def``, not ``async def``, for the reason ``projects.py`` gives. +Reading a spooled upload is blocking I/O too. +""" + +from __future__ import annotations + +from typing import Annotated +from uuid import UUID + +from fastapi import File, Form, Response, UploadFile, status + +from visionset.kernel.ports import DEFAULT_EXTRACTION_FPS +from visionset.kernel.services import IngestService, SourceService +from visionset.server.dependencies import RunnerDep, WorkspaceDep, protected_router +from visionset.server.errors import documented +from visionset.server.models import ( + IngestJobOut, + IngestJobPage, + IngestStart, + SourceOut, + SourcePage, +) +from visionset.server.uploads import stage + +project_router = protected_router(prefix="/projects/{project_id}/sources", tags=["sources"]) +router = protected_router(prefix="/sources", tags=["sources"]) + +#: The decomposition rate, as a multipart field. ``gt=0`` mirrors +#: ``VideoProvenance.extraction_fps``' own bound, which is what keeps +#: `SourceService`'s bare ``ValueError`` — outside the ``VisionSetError`` tree, +#: so a 500 — from ever being reachable over HTTP. +ExtractionFpsForm = Annotated[ + float, + Form(gt=0, description="Frames per second to cut the clip at. One per second by default."), +] + + +@project_router.post("/images", status_code=status.HTTP_201_CREATED, responses=documented(404)) +def register_image_source( + workspace: WorkspaceDep, + project_id: UUID, + files: Annotated[list[UploadFile], File(description="The images, as one multipart part each.")], +) -> SourceOut: + """Offer a project a folder of stills. + + The parts are staged as one directory and that directory becomes the source. + Uploading the same files again returns the **same** source rather than a + second one: staging is content-addressed, so identical bytes under identical + filenames land on the same path, and registration is idempotent on that path. + + Nothing is decoded here — what the files turn out to be is read at ingest, + and a file that is not an image is reported there rather than refused now. + """ + # ``capture_params`` is not on the wire. It is an opaque operator-supplied + # mapping, and threading a JSON object through a multipart form is a + # contract decision with no caller asking for it yet. + staged = stage(workspace.root, files) + return SourceOut.of(SourceService(workspace).register_images(project_id, staged.directory)) + + +@project_router.post("/video", status_code=status.HTTP_201_CREATED, responses=documented(404)) +def register_video_source( + workspace: WorkspaceDep, + project_id: UUID, + file: Annotated[UploadFile, File(description="The clip.")], + extraction_fps: ExtractionFpsForm = DEFAULT_EXTRACTION_FPS, +) -> SourceOut: + """Offer a project a clip, to be cut at `extraction_fps`. + + The clip is probed on the way in, so a file that is not a video, or one + whose bytes will not decode, is 422 here rather than a run that fails later. + + The rate is part of what the source *is*: the same clip registered at 1 fps + and again at 5 fps is two sources over one file, which is what makes "the + same source yields the same assets" mean anything. + """ + staged = stage(workspace.root, [file]) + source = SourceService(workspace).register_video( + project_id, staged.only, extraction_fps=extraction_fps + ) + return SourceOut.of(source) + + +@project_router.get("", responses=documented(404)) +def list_sources(workspace: WorkspaceDep, project_id: UUID) -> SourcePage: + """Every source of that project, in registration order.""" + found = SourceService(workspace).list(project_id) + return SourcePage(items=[SourceOut.of(source) for source in found], total=len(found)) + + +@router.get("/{source_id}", responses=documented(404)) +def get_source(workspace: WorkspaceDep, source_id: UUID) -> SourceOut: + """The source with that id.""" + return SourceOut.of(SourceService(workspace).get(source_id)) + + +@router.post( + "/{source_id}/ingest-jobs", + status_code=status.HTTP_202_ACCEPTED, + responses=documented(404), +) +def start_ingest( + workspace: WorkspaceDep, + runner: RunnerDep, + response: Response, + source_id: UUID, + body: IngestStart | None = None, +) -> IngestJobOut: + """Launch a run over the source and answer at once with the job to poll. + + **202, not 201**: the row exists, the work does not. Poll + `GET /ingest-jobs/{id}` — the `Location` header names it — and watch + `processed` climb until `state` is `completed` or `failed`. + + A run that could not even be recorded is refused here; everything that goes + wrong afterwards is reported *on the job*, which is the whole point of the + shape. Unreadable files land in `failures` and do not fail the run; a + missing ffmpeg does, in `error`. + """ + ingest = IngestService(workspace) + job = ingest.enqueue(source_id, batch_name=None if body is None else body.batch_name) + # ``resume``, not ``ingest``: the row is already there and ``pending`` is + # exactly what ``resume`` picks up. Doing the whole call in the worker would + # mean creating a second job. + runner.submit(lambda: ingest.resume(job.id)) + response.headers["Location"] = f"/ingest-jobs/{job.id}" + return IngestJobOut.of(job) + + +@router.get("/{source_id}/ingest-jobs", responses=documented(404)) +def list_ingest_jobs(workspace: WorkspaceDep, source_id: UUID) -> IngestJobPage: + """Every run of that source, in the order they were asked for.""" + found = IngestService(workspace).list(source_id) + return IngestJobPage(items=[IngestJobOut.of(job) for job in found], total=len(found)) diff --git a/src/visionset/server/runner.py b/src/visionset/server/runner.py new file mode 100644 index 00000000..1d305a8d --- /dev/null +++ b/src/visionset/server/runner.py @@ -0,0 +1,75 @@ +# usage: from visionset.server.runner import IngestRunner +"""Where a launched ingest actually runs. + +Ingest is the first operation this API exposes that outlives its request, and +the pattern set here is the one #29's jobs, #33's UI and #35's tools all reuse: +**the launch returns an id, the client polls that id.** For the id to be worth +handing back, the row has to exist before the work starts — which is what +`IngestService.enqueue` is for — and the work has to happen somewhere other than +the request that asked for it. This is that somewhere. + +**One worker, deliberately.** A `ThreadPoolExecutor` of size one serializes +every run against a single-writer SQLite store instead of racing them. What #80 +bought — WAL, a `busy_timeout`, `OperationalError` translated to `WorkspaceBusy` +— is then the safety net for the *reader*: a client polling `GET /ingest-jobs/…` +while the worker holds a write transaction reads through WAL rather than +blocking on it. Widening this pool is a decision about the store, not a tuning +knob, so it is a constructor argument with no route able to reach it. + +**A run that raises is logged here and reported on its row.** `IngestService` +has already marked the job ``failed`` and written its ``error`` by the time the +exception gets this far, so the client's answer is complete without this module; +the log is for whoever runs the server. Swallowing it here is what keeps it from +becoming an unretrieved-future warning at interpreter shutdown. +""" + +from __future__ import annotations + +import logging +from collections.abc import Callable +from concurrent.futures import Future, ThreadPoolExecutor +from typing import Final + +_logger: Final = logging.getLogger(__name__) + + +class IngestRunner: + """A single background worker, owned by one application.""" + + def __init__(self, workers: int = 1) -> None: + # Constructing an executor starts no threads — the first ``submit`` does + # — so building one in ``create_app()`` costs an import of the module + # nothing, which ``scripts/export_openapi.py`` depends on. + self._executor = ThreadPoolExecutor( + max_workers=workers, thread_name_prefix="visionset-ingest" + ) + + def submit(self, run: Callable[[], object]) -> Future[object]: + """Hand ``run`` to the worker and return immediately. + + The future is returned rather than discarded so a test can join on it + instead of polling with sleeps — the discipline + ``tests/kernel/test_concurrency.py`` set. Routes ignore it: what a client + waits on is the job row. + """ + return self._executor.submit(self._guarded, run) + + def shutdown(self) -> None: + """Wait for the running job and drop whatever was still queued. + + ``cancel_futures`` rather than draining: a queued run has not started, + so cancelling it leaves its job exactly where `enqueue` put it — + ``pending``, and resumable — while draining an arbitrary backlog would + hold shutdown open for as long as somebody kept uploading. + """ + self._executor.shutdown(wait=True, cancel_futures=True) + + @staticmethod + def _guarded(run: Callable[[], object]) -> object: + try: + return run() + except Exception: + # ``BaseException`` is deliberately not caught, the rule + # ``InProcessEventBus`` already follows. + _logger.exception("ingest run failed; see the job's error field") + return None diff --git a/src/visionset/server/uploads.py b/src/visionset/server/uploads.py new file mode 100644 index 00000000..33c6fdd9 --- /dev/null +++ b/src/visionset/server/uploads.py @@ -0,0 +1,160 @@ +# usage: from visionset.server.uploads import stage +"""Where an uploaded file lands before a source is registered over it. + +The kernel registers a source by **path**: `SourceService.register_images` takes +a directory and `register_video` takes a file, both of which have to exist on +this machine. An HTTP client has bytes. This module is the one place that bridges +the two, and it is deliberately a *server* concern — the CLI and MCP surfaces +already hold real paths and never come through here. + +**Uploads are staged content-addressed**, under ``/uploads//``, +where the digest names the whole part set: sha-256 over the sorted +``name:sha256`` lines. One rule for one clip and for fifty stills. The property +that buys is worth the arithmetic — the same bytes under the same filenames land +on the same path, so `SourceService`'s own ``(kind, path, extraction_fps)`` +idempotency answers a repeated upload with the *same* `Source` instead of a +second one over a second copy. + +**Nothing is buffered whole.** Starlette spools an `UploadFile` to disk past +1 MiB and hands over a plain synchronous file object, so a ``def`` handler reads +it here in chunks straight into the staging file. ``upload.read()`` would undo +that in one line and must not appear in this module. + +**Staged uploads are never deleted**, which is the posture blobs already have +(`BlobStore` has no ``delete``). A workspace's disk grows with what was offered +to it, not only with what was kept; ``docs/api.md`` says so out loud rather than +implying a cleanup nothing performs. +""" + +from __future__ import annotations + +import hashlib +import os +import shutil +from collections.abc import Sequence +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import Final +from uuid import uuid4 + +from fastapi import UploadFile + +#: Staging root, relative to the workspace directory. Server-owned: the kernel +#: neither writes nor reads it, and `WorkspaceService.open` simply tolerates it +#: the way it tolerates anything else beside the database and ``blobs/``. +UPLOADS_DIRNAME: Final = "uploads" + +#: What a part is called when the client sent no usable filename. +FALLBACK_NAME: Final = "upload" + +_CHUNK: Final = 1024 * 1024 + + +@dataclass(frozen=True, slots=True) +class StagedUpload: + """A staged part set: the directory it landed in, and what is in it.""" + + directory: Path + names: tuple[str, ...] + + @property + def only(self) -> Path: + """The single staged file, for a route that accepted exactly one.""" + return self.directory / self.names[0] + + +def safe_name(filename: str | None) -> str: + """The last component of a client-supplied filename, and nothing else. + + The path-traversal guard, and the only one needed: everything a client sends + is reduced to a bare name before it is joined to anything. Backslashes are + folded first because a Windows browser sends ``C:\\Users\\me\\clip.mp4`` in + that field. A name that survives as empty, as ``.`` or ``..``, or that + carries a NUL, is replaced rather than refused — a badly named part is not a + reason to reject an otherwise good upload. + """ + candidate = PurePosixPath((filename or "").replace("\\", "/")).name.strip() + if not candidate or candidate in {".", ".."} or "\x00" in candidate: + return FALLBACK_NAME + return candidate + + +def stage(root: Path, uploads: Sequence[UploadFile]) -> StagedUpload: + """Write every part under ``root`` and return where they landed. + + The parts go into a private ``.staging-`` directory first, because the + name of the final one is not known until every byte has been hashed. The + rename is what publishes them, so a half-written upload is never a directory + a source could be registered over. + """ + uploads_root = root / UPLOADS_DIRNAME + uploads_root.mkdir(parents=True, exist_ok=True) + staging = Path(uploads_root / f".staging-{uuid4().hex}") + staging.mkdir() + + try: + staged: list[tuple[str, str]] = [] + taken: set[str] = set() + for upload in uploads: + name = _unused(safe_name(upload.filename), taken) + staged.append((name, _write(staging / name, upload))) + + target = uploads_root / _set_digest(staged) + if target.exists(): + # Already staged by an earlier upload of the same bytes — identical + # content, so the winner is as good as ours. + shutil.rmtree(staging) + else: + try: + os.replace(staging, target) + except OSError: + # A concurrent upload of the same set won the rename between the + # check above and here. Same argument, same answer. + shutil.rmtree(staging, ignore_errors=True) + except BaseException: + shutil.rmtree(staging, ignore_errors=True) + raise + + return StagedUpload(directory=target, names=tuple(name for name, _ in staged)) + + +def _unused(name: str, taken: set[str]) -> str: + """``name``, or the next ``name-2``/``name-3`` that is not spoken for. + + Two parts may legitimately arrive under one filename, and a directory source + reads its files by name — collapsing them would silently drop one. The + suffix goes before the extension so the file still looks like what it is. + """ + candidate = name + stem, dot, suffix = name.partition(".") + attempt = 1 + while candidate in taken: + attempt += 1 + candidate = f"{stem}-{attempt}{dot}{suffix}" + taken.add(candidate) + return candidate + + +def _write(path: Path, upload: UploadFile) -> str: + """Stream one part to ``path``, returning the sha-256 of what was written.""" + digest = hashlib.sha256() + # Seek first for the reason `ImageProcessor`'s port docstring gives: a handle + # is read from wherever it sits, and this one has been looked at before. + upload.file.seek(0) + with path.open("wb") as out: + while chunk := upload.file.read(_CHUNK): + digest.update(chunk) + out.write(chunk) + return digest.hexdigest() + + +def _set_digest(staged: Sequence[tuple[str, str]]) -> str: + """One digest naming a whole part set — the staging directory's name. + + Sorted, so the order parts arrived in cannot fork one upload into two + directories. The name is inside the digest as well as the content: a file + renamed is a different thing to offer a project, and a directory source + reads its members by name. + """ + lines = "".join(f"{name}:{content}\n" for name, content in sorted(staged)) + return hashlib.sha256(lines.encode()).hexdigest() diff --git a/tests/kernel/test_batch_service.py b/tests/kernel/test_batch_service.py index a209cf9c..5ccd0e84 100644 --- a/tests/kernel/test_batch_service.py +++ b/tests/kernel/test_batch_service.py @@ -204,6 +204,31 @@ def test_removing_an_asset_the_batch_does_not_hold_is_a_no_op(tmp_path: Path) -> fixture.close() +def test_reading_a_batchs_assets_gives_them_in_membership_order(tmp_path: Path) -> None: + """The read behind "what did that ingest gather" — order is the stored position, + so a caller reading twice sees one sequence and `add_assets` appends.""" + fixture = Fixture(tmp_path) + batch = fixture.batches.create(fixture.project.id, "first", fixture.assets[:2]) + fixture.batches.add_assets(batch.id, fixture.assets[2:]) + + assert [asset.id for asset in fixture.batches.assets(batch.id)] == fixture.assets + fixture.close() + + +def test_an_empty_batch_has_no_assets_rather_than_no_answer(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + batch = fixture.batches.create(fixture.project.id, "empty") + assert fixture.batches.assets(batch.id) == [] + fixture.close() + + +def test_reading_the_assets_of_an_unknown_batch_is_refused(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + with pytest.raises(BatchNotFound): + fixture.batches.assets(uuid4()) + fixture.close() + + def test_an_asset_from_another_project_cannot_join_the_batch(tmp_path: Path) -> None: fixture = Fixture(tmp_path) stranger = Fixture(tmp_path, "other") diff --git a/tests/kernel/test_ingest_service.py b/tests/kernel/test_ingest_service.py index edb5857b..f38adf33 100644 --- a/tests/kernel/test_ingest_service.py +++ b/tests/kernel/test_ingest_service.py @@ -252,16 +252,18 @@ def freeze(self, batch_id: UUID) -> None: def job_in(self, state: IngestState) -> IngestJob: """A job in `state`, over a source of two images that is readable now. - `completed` and `failed` are walked to through real operations — a run - that works, and a run whose directory was taken away and put back. The - other two are written directly, and this is the only place in this file - that plants a state rather than reaching it: a synchronous run passes - through `pending` and `running` inside a single call and never leaves - one behind, so there is nothing to walk to. Leaving them out instead - would leave half the table unswept. + Three of the four are walked to through real operations — `pending` is + what `enqueue` leaves, `completed` is a run that works, and `failed` is a + run whose directory was taken away and put back. Only `running` is + written directly, and it is the one state no operation leaves behind: a + run that reaches it either finishes or fails inside the same call, and a + row stuck there is by definition a process that died. Leaving it out + would leave a quarter of the table unswept. """ write_images(self.stills, count=2) source = self.sources.register_images(self.project.id, self.stills) + if state is IngestState.PENDING: + return self.ingest.enqueue(source.id) if state is IngestState.COMPLETED: return self.ingest.get(self.ingest.ingest(source.id).job_id) if state is IngestState.FAILED: @@ -1071,6 +1073,114 @@ def test_a_failed_run_keeps_the_progress_it_had_made(tmp_path: Path) -> None: workspace.close() +# --- asking for a run without doing it ------------------------------------- + + +def test_enqueue_leaves_a_pending_job_and_reads_nothing(tmp_path: Path) -> None: + """The half a caller needs when the work happens somewhere else.""" + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=2) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + job = fixture.ingest.enqueue(source.id) + + assert job.state is IngestState.PENDING + assert (job.processed, job.total, job.failures) == (0, None, ()) + assert fixture.assets() == [] + assert fixture.batches.list(fixture.project.id) == [] + fixture.close() + + +def test_resume_is_how_an_enqueued_run_is_started(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=2) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + job = fixture.ingest.enqueue(source.id, batch_name="monday") + + result = fixture.ingest.resume(job.id) + + assert result.job_id == job.id + assert fixture.ingest.get(job.id).state is IngestState.COMPLETED + assert fixture.batches.get(result.batch_id).name == "monday" + assert len(result.created_asset_ids) == 2 + fixture.close() + + +def test_enqueue_refuses_before_it_writes_a_row(tmp_path: Path) -> None: + """The point of the split: a refused launch leaves nothing to poll.""" + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=2) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + with pytest.raises(SourceNotFound): + fixture.ingest.enqueue(uuid4()) + with pytest.raises(InvalidName, match="batch name"): + fixture.ingest.enqueue(source.id, batch_name=" ") + + assert fixture.ingest.list(source.id) == [] + fixture.close() + + +def test_enqueue_records_the_batch_the_run_is_headed_for(tmp_path: Path) -> None: + """`resume` reads it back, so it has to be on the row rather than in a caller.""" + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=2) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + target = fixture.batches.create(fixture.project.id, "waiting") + + job = fixture.ingest.enqueue(source.id, batch_id=target.id) + assert job.batch_id == target.id + + result = fixture.ingest.resume(job.id) + assert result.batch_id == target.id + assert len(fixture.batches.get(target.id).asset_ids) == 2 + fixture.close() + + +def test_enqueue_refuses_a_frozen_target_batch(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=2) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + batch = fixture.batches.create( + fixture.project.id, "frozen", fixture.ingest.ingest(source.id).asset_ids + ) + fixture.freeze(batch.id) + + with pytest.raises(BatchNotEditable): + fixture.ingest.enqueue(source.id, batch_id=batch.id) + fixture.close() + + +def test_resumable_reports_a_job_that_may_run_without_moving_it(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + job = fixture.job_in(IngestState.FAILED) + + assert fixture.ingest.resumable(job.id).id == job.id + assert fixture.ingest.get(job.id).state is IngestState.FAILED + fixture.close() + + +@pytest.mark.parametrize("state", [IngestState.COMPLETED, IngestState.RUNNING]) +def test_resumable_refuses_exactly_what_resume_refuses(tmp_path: Path, state: IngestState) -> None: + """One spelling of the pre-check, so a caller that runs the work elsewhere + gets the same refusal on its own thread.""" + fixture = Fixture(tmp_path) + job = fixture.job_in(state) + + with pytest.raises(InvalidTransition): + fixture.ingest.resumable(job.id) + with pytest.raises(InvalidTransition): + fixture.ingest.resume(job.id) + fixture.close() + + +def test_resumable_needs_a_job_that_exists(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + with pytest.raises(IngestJobNotFound): + fixture.ingest.resumable(uuid4()) + fixture.close() + + # --- resuming a failed run ------------------------------------------------ diff --git a/tests/server/_api.py b/tests/server/_api.py index 3d14a4a8..ffaab2a4 100644 --- a/tests/server/_api.py +++ b/tests/server/_api.py @@ -22,18 +22,23 @@ from visionset.kernel.services import TokenService, WorkspaceService from visionset.server.main import create_app +from visionset.server.runner import IngestRunner TOKEN_NAME: Final = "api-tests" -def served_app(root: Path) -> FastAPI: +def served_app(root: Path, *, runner: IngestRunner | None = None) -> FastAPI: """The shipped application, serving the workspace at ``root``. The handle is replaced rather than the environment patched, so the test says - which workspace it means instead of relying on process-wide state. + which workspace it means instead of relying on process-wide state. A + ``runner`` is replaced the same way and for the same reason; the one + ``create_app`` built has started no thread, so dropping it costs nothing. """ app = create_app() app.state.workspace_handle = handle_for(root) + if runner is not None: + app.state.ingest_runner = runner return app @@ -46,11 +51,14 @@ def api_workspace(root: Path) -> str: workspace.close() -def api_client(root: Path) -> TestClient: +def api_client(root: Path, *, runner: IngestRunner | None = None) -> TestClient: """A client for a fresh workspace at ``root``, authenticated on every request. - Use it as a context manager: the lifespan is what closes the workspace, and - a `visionset.db-wal` left behind would outlive the test's ``tmp_path``. + Use it as a context manager: the lifespan is what closes the workspace and + stops the ingest worker, and a `visionset.db-wal` left behind would outlive + the test's ``tmp_path``. """ secret = api_workspace(root) - return TestClient(served_app(root), headers={"Authorization": f"Bearer {secret}"}) + return TestClient( + served_app(root, runner=runner), headers={"Authorization": f"Bearer {secret}"} + ) diff --git a/tests/server/_runner.py b/tests/server/_runner.py new file mode 100644 index 00000000..63c00dc0 --- /dev/null +++ b/tests/server/_runner.py @@ -0,0 +1,63 @@ +"""Ingest runners a test can control, instead of polling with sleeps. + +`tests/kernel/test_concurrency.py` set the discipline these follow: sequence on +`threading.Event`, never on a sleep; join with a timeout and then assert the +thread is actually dead. A test that waits by sleeping is a test that is slow +when it passes and flaky when it does not. + +Plain functions in a private module, the `_probe.py` / `_api.py` precedent — +there is still no `conftest.py` anywhere. +""" + +from __future__ import annotations + +import threading +from collections.abc import Callable +from concurrent.futures import Future + +from visionset.server.runner import IngestRunner + +#: Long enough that a loaded CI runner does not trip it, short enough that a +#: genuine deadlock fails the suite rather than hanging it. +JOIN_TIMEOUT = 30.0 + + +class RecordingRunner(IngestRunner): + """The real runner, keeping every future so a test can join on the work.""" + + def __init__(self) -> None: + super().__init__() + self.futures: list[Future[object]] = [] + + def submit(self, run: Callable[[], object]) -> Future[object]: + future = super().submit(run) + self.futures.append(future) + return future + + def wait(self) -> None: + """Block until every submitted run has finished.""" + for future in self.futures: + future.result(timeout=JOIN_TIMEOUT) + + +class GatedRunner(RecordingRunner): + """A runner that parks before the work, so "launched" can be observed. + + `entered` is set once the worker has picked the job up and `release` is what + lets it proceed — which is how a test asserts the job row is already + readable while the run has not started. Nothing here touches the kernel: the + gate is around the submitted callable, not inside it. + """ + + def __init__(self) -> None: + super().__init__() + self.entered = threading.Event() + self.release = threading.Event() + + def submit(self, run: Callable[[], object]) -> Future[object]: + def gated() -> object: + self.entered.set() + assert self.release.wait(timeout=JOIN_TIMEOUT), "the test never released the worker" + return run() + + return super().submit(gated) diff --git a/tests/server/test_batches.py b/tests/server/test_batches.py new file mode 100644 index 00000000..7c732468 --- /dev/null +++ b/tests/server/test_batches.py @@ -0,0 +1,112 @@ +"""The one batch route #28 ships: what an ingest put in a batch. + +#29 owns the rest of this surface. What is pinned here is the envelope and the +404, because those are what its additions have to stay compatible with. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from pathlib import Path +from uuid import uuid4 + +import pytest +from fastapi.testclient import TestClient +from tests.fixtures.media import write_image +from tests.server._api import api_client +from tests.server._runner import RecordingRunner + + +@pytest.fixture() +def runner() -> RecordingRunner: + return RecordingRunner() + + +@pytest.fixture() +def client(tmp_path: Path, runner: RecordingRunner) -> Iterator[TestClient]: + with api_client(tmp_path / "ws", runner=runner) as made: + yield made + + +def png_part(tmp_path: Path, name: str, seed: int = 0) -> tuple[str, tuple[str, bytes, str]]: + """One multipart part carrying a generated image.""" + return ("files", (name, write_image(tmp_path / name, seed=seed).read_bytes(), "image/png")) + + +@pytest.fixture() +def ingested(client: TestClient, tmp_path: Path, runner: RecordingRunner) -> str: + """A batch id, reached the way a client reaches one: by ingesting into it.""" + project = client.post("/projects", json={"name": "road-signs"}).json()["id"] + parts = [png_part(tmp_path, f"{index}.png", seed=index) for index in range(3)] + source = client.post(f"/projects/{project}/sources/images", files=parts).json()["id"] + job = client.post(f"/sources/{source}/ingest-jobs").json() + runner.wait() + batch_id: str = client.get(f"/ingest-jobs/{job['id']}").json()["batch_id"] + return batch_id + + +def test_a_batchs_assets_answer_with_the_envelope(client: TestClient, ingested: str) -> None: + response = client.get(f"/batches/{ingested}/assets") + + assert response.status_code == 200 + body = response.json() + assert body["total"] == 3 + assert len(body["items"]) == 3 + + +def test_membership_order_is_stable(client: TestClient, ingested: str) -> None: + """Stored order, so reading twice gives the same sequence — what paging will page.""" + first = client.get(f"/batches/{ingested}/assets").json()["items"] + second = client.get(f"/batches/{ingested}/assets").json()["items"] + + assert [asset["id"] for asset in first] == [asset["id"] for asset in second] + + +def test_an_asset_carries_its_hashes_but_not_its_path(client: TestClient, ingested: str) -> None: + """`uri` is a server-side path; reaching the bytes is #30's download by hash.""" + asset = client.get(f"/batches/{ingested}/assets").json()["items"][0] + + assert "uri" not in asset + assert len(asset["content_hash"]) == 64 + assert len(asset["thumbnail_hash"]) == 64 + assert asset["format"] == "png" + + +def test_a_batch_an_ingest_could_not_fill_is_an_empty_page_not_a_404( + client: TestClient, runner: RecordingRunner +) -> None: + """A run whose every item was unreadable still makes a batch. It is just empty.""" + project = client.post("/projects", json={"name": "empty"}).json()["id"] + source = client.post( + f"/projects/{project}/sources/images", + files=[("files", ("notes.txt", b"not an image", "text/plain"))], + ).json()["id"] + job = client.post(f"/sources/{source}/ingest-jobs").json() + runner.wait() + batch_id = client.get(f"/ingest-jobs/{job['id']}").json()["batch_id"] + + response = client.get(f"/batches/{batch_id}/assets") + + assert response.status_code == 200 + assert response.json() == {"items": [], "total": 0} + + +def test_an_unknown_batch_is_404(client: TestClient) -> None: + response = client.get(f"/batches/{uuid4()}/assets") + + assert response.status_code == 404 + assert response.json()["code"] == "BATCH_NOT_FOUND" + + +def test_a_malformed_batch_id_is_422_not_404(client: TestClient) -> None: + response = client.get("/batches/not-a-uuid/assets") + + assert response.status_code == 422 + assert response.json()["code"] == "VALIDATION_ERROR" + + +def test_the_batch_route_refuses_a_request_with_no_token(client: TestClient) -> None: + response = client.get(f"/batches/{uuid4()}/assets", headers={"Authorization": ""}) + + assert response.status_code == 401 + assert response.json()["code"] == "UNAUTHORIZED" diff --git a/tests/server/test_ingest.py b/tests/server/test_ingest.py new file mode 100644 index 00000000..41fdae2f --- /dev/null +++ b/tests/server/test_ingest.py @@ -0,0 +1,343 @@ +"""Launching a run and polling it — the contract every long operation reuses. + +The acceptance walk of #28 is one test here: upload a clip, register it at 5 fps, +launch, wait, read the assets. It is deliberately the shape a real client has — +nothing reaches past the API for an answer the API is supposed to give. + +**Nothing in this module sleeps.** Waiting is `RecordingRunner.wait()`, which +joins the worker's future, and sequencing is `GatedRunner`'s events — the +discipline `tests/kernel/test_concurrency.py` set. A test that polls with sleeps +is slow when it passes and flaky when it does not. +""" + +from __future__ import annotations + +import threading +from collections.abc import Iterator +from pathlib import Path +from typing import Any +from uuid import UUID, uuid4 + +import pytest +from fastapi.testclient import TestClient +from tests.fixtures.media import write_image, write_video +from tests.server._api import api_client +from tests.server._runner import JOIN_TIMEOUT, GatedRunner, RecordingRunner + +from visionset.kernel.services import WorkspaceService + +# Above `testsrc`'s resolution floor — see `test_sources.py`. +CLIP_SIZE = (160, 120) + +#: 2 seconds at 10 fps cut at 5 fps. The generator's defaults make this exact, +#: which is what lets the walk assert a count rather than a range. +EXTRACTION_FPS = 5 +EXPECTED_FRAMES = 10 + + +@pytest.fixture() +def runner() -> RecordingRunner: + return RecordingRunner() + + +@pytest.fixture() +def client(tmp_path: Path, runner: RecordingRunner) -> Iterator[TestClient]: + with api_client(tmp_path / "ws", runner=runner) as made: + yield made + + +@pytest.fixture() +def project(client: TestClient) -> str: + response = client.post("/projects", json={"name": "road-signs"}) + assert response.status_code == 201, response.text + project_id: str = response.json()["id"] + return project_id + + +def registered_clip(client: TestClient, project: str, tmp_path: Path) -> str: + clip = write_video(tmp_path / "made" / "drive.mp4", size=CLIP_SIZE).path + response = client.post( + f"/projects/{project}/sources/video", + files={"file": (clip.name, clip.read_bytes(), "video/mp4")}, + data={"extraction_fps": EXTRACTION_FPS}, + ) + assert response.status_code == 201, response.text + source_id: str = response.json()["id"] + return source_id + + +def png_part( + tmp_path: Path, name: str = "a.png", seed: int = 0 +) -> tuple[str, tuple[str, bytes, str]]: + """One multipart part carrying a generated image.""" + return ("files", (name, write_image(tmp_path / name, seed=seed).read_bytes(), "image/png")) + + +def registered_images(client: TestClient, project: str, *parts: Any) -> str: + response = client.post(f"/projects/{project}/sources/images", files=list(parts)) + assert response.status_code == 201, response.text + source_id: str = response.json()["id"] + return source_id + + +def launch(client: TestClient, source: str, **body: Any) -> Any: + return client.post(f"/sources/{source}/ingest-jobs", json=body or None) + + +# --- the acceptance walk ----------------------------------------------------- + + +def test_a_clip_uploaded_and_ingested_at_five_fps_lists_its_assets( + client: TestClient, project: str, tmp_path: Path, runner: RecordingRunner +) -> None: + source = registered_clip(client, project, tmp_path) + + started = launch(client, source) + assert started.status_code == 202, started.text + job = started.json() + assert job["state"] == "pending" + assert started.headers["Location"] == f"/ingest-jobs/{job['id']}" + + runner.wait() + + polled = client.get(f"/ingest-jobs/{job['id']}").json() + assert polled["state"] == "completed" + assert polled["processed"] == EXPECTED_FRAMES + # NULL for a clip: `VideoMetadata` carries no frame count by design. + assert polled["total"] is None + assert polled["failures"] == [] + assert polled["batch_id"] is not None + + assets = client.get(f"/batches/{polled['batch_id']}/assets") + assert assets.status_code == 200 + body = assets.json() + assert body["total"] == EXPECTED_FRAMES + assert [asset["frame_index"] for asset in body["items"]] == list(range(EXPECTED_FRAMES)) + assert all(asset["source_id"] == source for asset in body["items"]) + + +# --- launching --------------------------------------------------------------- + + +def test_a_launch_answers_before_the_worker_has_picked_the_job_up( + tmp_path: Path, project: str +) -> None: + """The whole promise of the 202: the row is pollable while the work has not begun.""" + gated = GatedRunner() + with api_client(tmp_path / "gated", runner=gated) as client: + made = client.post("/projects", json={"name": "gated"}).json()["id"] + source = registered_images(client, made, png_part(tmp_path)) + + job = launch(client, source).json() + assert gated.entered.wait(timeout=JOIN_TIMEOUT) + + parked = client.get(f"/ingest-jobs/{job['id']}") + assert parked.status_code == 200 + assert parked.json()["state"] == "pending" + + gated.release.set() + gated.wait() + + assert client.get(f"/ingest-jobs/{job['id']}").json()["state"] == "completed" + + +def test_a_launch_names_the_batch_it_was_asked_to( + client: TestClient, project: str, tmp_path: Path, runner: RecordingRunner +) -> None: + source = registered_images(client, project, png_part(tmp_path)) + + job = launch(client, source, batch_name="monday").json() + runner.wait() + + assert client.get(f"/ingest-jobs/{job['id']}").json()["batch_name"] == "monday" + + +def test_a_blank_batch_name_is_the_domains_own_422_not_a_500( + client: TestClient, project: str, tmp_path: Path, runner: RecordingRunner +) -> None: + """`InvalidName` is a mapped domain error, so no wire-model validator restates it. + + The contrast with `LabelClassBody` is the point: *that* one needs a parsing-time + validator because the domain refuses with a pydantic `ValidationError`, which + reaches the catch-all handler as a 500. This one refuses with a `VisionSetError` + that `ERROR_RULES` already places, and it arrives before the job row is written. + """ + source = registered_images(client, project, png_part(tmp_path)) + + response = launch(client, source, batch_name=" ") + + assert response.status_code == 422 + assert response.json()["code"] == "INVALID_NAME" + assert runner.futures == [] + + +def test_launching_over_an_unknown_source_is_404_and_starts_nothing( + client: TestClient, runner: RecordingRunner +) -> None: + """Refused on the calling thread, so a 202 never points at a job row nobody wrote.""" + response = launch(client, str(uuid4())) + + assert response.status_code == 404 + assert response.json()["code"] == "SOURCE_NOT_FOUND" + assert runner.futures == [] + + +# --- what a run reports ------------------------------------------------------ + + +def test_an_unreadable_item_is_reported_and_does_not_fail_the_run( + client: TestClient, project: str, tmp_path: Path, runner: RecordingRunner +) -> None: + """Failure splits by remedy: operator noise is a line in the report, not a dead run.""" + source = registered_images( + client, + project, + png_part(tmp_path, seed=1), + ("files", ("notes.txt", b"not an image", "text/plain")), + ) + + job = launch(client, source).json() + runner.wait() + + polled = client.get(f"/ingest-jobs/{job['id']}").json() + assert polled["state"] == "completed" + assert polled["processed"] == 2 + assert polled["total"] == 2 + assert [(Path(f["name"]).name, f["kind"]) for f in polled["failures"]] == [ + ("notes.txt", "unsupported") + ] + # A different field from `failures`, and empty here: one broken machine is + # not five thousand broken files, and neither is one unreadable file a run. + assert polled["error"] is None + + assets = client.get(f"/batches/{polled['batch_id']}/assets").json() + assert assets["total"] == 1 + + +def test_listing_the_runs_of_a_source( + client: TestClient, project: str, tmp_path: Path, runner: RecordingRunner +) -> None: + source = registered_images(client, project, png_part(tmp_path)) + first = launch(client, source).json() + runner.wait() + second = launch(client, source).json() + runner.wait() + + body = client.get(f"/sources/{source}/ingest-jobs").json() + + assert body["total"] == 2 + assert {job["id"] for job in body["items"]} == {first["id"], second["id"]} + + +def test_re_ingesting_a_source_creates_nothing( + client: TestClient, project: str, tmp_path: Path, runner: RecordingRunner +) -> None: + """Content is addressed by hash, so a second run reports the same items as already held.""" + source = registered_images(client, project, png_part(tmp_path)) + first = launch(client, source).json() + runner.wait() + second = launch(client, source).json() + runner.wait() + + first_batch = client.get(f"/ingest-jobs/{first['id']}").json()["batch_id"] + second_batch = client.get(f"/ingest-jobs/{second['id']}").json()["batch_id"] + + assert first_batch != second_batch + left = client.get(f"/batches/{first_batch}/assets").json()["items"] + right = client.get(f"/batches/{second_batch}/assets").json()["items"] + assert [a["id"] for a in left] == [a["id"] for a in right] + + +# --- polling and resuming ---------------------------------------------------- + + +def test_reading_an_unknown_job_is_404(client: TestClient) -> None: + response = client.get(f"/ingest-jobs/{uuid4()}") + + assert response.status_code == 404 + assert response.json()["code"] == "INGEST_JOB_NOT_FOUND" + + +def test_resuming_a_completed_run_is_409_rather_than_a_silent_no_op( + client: TestClient, project: str, tmp_path: Path, runner: RecordingRunner +) -> None: + """Refused on the calling thread: a 202 here would leave a client unable to tell + a redo from a job that did nothing.""" + source = registered_images(client, project, png_part(tmp_path)) + job = launch(client, source).json() + runner.wait() + before = len(runner.futures) + + response = client.post(f"/ingest-jobs/{job['id']}/resume") + + assert response.status_code == 409 + assert response.json()["code"] == "INVALID_TRANSITION" + assert len(runner.futures) == before + + +def test_resuming_an_unknown_job_is_404(client: TestClient) -> None: + response = client.post(f"/ingest-jobs/{uuid4()}/resume") + + assert response.status_code == 404 + assert response.json()["code"] == "INGEST_JOB_NOT_FOUND" + + +def test_polling_is_answered_while_another_writer_holds_the_workspace( + client: TestClient, project: str, tmp_path: Path, runner: RecordingRunner +) -> None: + """#80's payoff, at the surface it was landed for. + + A second `WorkspaceService` over one file is two engines with no shared cache + — what two *processes* look like to SQLite. It writes and parks; the request + thread must still be answered, and with the last committed state rather than + the writer's uncommitted one. + """ + source = registered_images(client, project, png_part(tmp_path)) + job = launch(client, source).json() + runner.wait() + + writing = threading.Event() + release = threading.Event() + other = WorkspaceService.open(tmp_path / "ws") + + def hold_the_write_lock() -> None: + with other.unit_of_work() as uow: + held = uow.ingest_jobs.get(UUID(job["id"])) + assert held is not None + uow.ingest_jobs.update(held.model_copy(update={"total": 99})) + writing.set() + assert release.wait(timeout=JOIN_TIMEOUT) + + writer = threading.Thread(target=hold_the_write_lock, name="lock-holder") + writer.start() + try: + assert writing.wait(timeout=JOIN_TIMEOUT) + during = client.get(f"/ingest-jobs/{job['id']}") + assert during.status_code == 200 + assert during.json()["total"] != 99 + finally: + release.set() + writer.join(timeout=JOIN_TIMEOUT) + other.close() + + assert not writer.is_alive() + assert client.get(f"/ingest-jobs/{job['id']}").json()["total"] == 99 + + +# --- the guard --------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("method", "path"), + [ + ("GET", "/ingest-jobs/00000000-0000-0000-0000-000000000000"), + ("POST", "/ingest-jobs/00000000-0000-0000-0000-000000000000/resume"), + ], +) +def test_every_ingest_route_refuses_a_request_with_no_token( + client: TestClient, method: str, path: str +) -> None: + response = client.request(method, path, headers={"Authorization": ""}) + + assert response.status_code == 401 + assert response.json()["code"] == "UNAUTHORIZED" diff --git a/tests/server/test_openapi_contract.py b/tests/server/test_openapi_contract.py index d327c36f..a0a44290 100644 --- a/tests/server/test_openapi_contract.py +++ b/tests/server/test_openapi_contract.py @@ -74,7 +74,13 @@ def test_the_page_envelope_is_named_for_its_item_type() -> None: """ schemas = app.openapi()["components"]["schemas"] - assert {"ProjectPage", "SchemaVersionPage"} <= set(schemas) + assert { + "AssetPage", + "IngestJobPage", + "ProjectPage", + "SchemaVersionPage", + "SourcePage", + } <= set(schemas) assert not [name for name in schemas if name.startswith("Page")] diff --git a/tests/server/test_sources.py b/tests/server/test_sources.py new file mode 100644 index 00000000..21137953 --- /dev/null +++ b/tests/server/test_sources.py @@ -0,0 +1,259 @@ +"""Registering a source over HTTP: upload, list, read, refuse. + +`test_projects.py`'s shape — a two-line fixture over `_api.py`, assertions at +the wire (status and `code`), and a closing guard sweep. What the kernel does +with a source is `tests/kernel/test_source_service.py`; nothing here restates it. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from pathlib import Path +from typing import Any +from uuid import uuid4 + +import pytest +from fastapi.testclient import TestClient +from tests.fixtures.media import write_image, write_video +from tests.server._api import api_client + +# Above `testsrc`'s resolution floor: below roughly 96x72 its per-frame movement +# falls under the scaler and consecutive frames come out byte-identical, which +# ingest then deduplicates. See `docs/examples.md`. +CLIP_SIZE = (160, 120) + + +@pytest.fixture() +def client(tmp_path: Path) -> Iterator[TestClient]: + with api_client(tmp_path / "ws") as made: + yield made + + +@pytest.fixture() +def project(client: TestClient) -> str: + response = client.post("/projects", json={"name": "road-signs"}) + assert response.status_code == 201, response.text + project_id: str = response.json()["id"] + return project_id + + +def image_part(tmp_path: Path, name: str, seed: int = 0) -> tuple[str, tuple[str, bytes, str]]: + path = write_image(tmp_path / "made" / name, seed=seed) + return ("files", (name, path.read_bytes(), "image/png")) + + +def post_images(client: TestClient, project: str, *parts: Any) -> Any: + return client.post(f"/projects/{project}/sources/images", files=list(parts)) + + +def post_video(client: TestClient, project: str, clip: Path, **form: Any) -> Any: + return client.post( + f"/projects/{project}/sources/video", + files={"file": (clip.name, clip.read_bytes(), "video/mp4")}, + data=form, + ) + + +@pytest.fixture() +def clip(tmp_path: Path) -> Path: + """A real clip. `write_video` is what requires ffmpeg, so image tests stay free of it.""" + return write_video(tmp_path / "made" / "drive.mp4", size=CLIP_SIZE).path + + +# --- registering stills ------------------------------------------------------ + + +def test_uploading_images_registers_a_directory_source( + client: TestClient, project: str, tmp_path: Path +) -> None: + response = post_images(client, project, image_part(tmp_path, "a.png", 1)) + + assert response.status_code == 201, response.text + body = response.json() + assert body["kind"] == "image_directory" + assert body["project_id"] == project + assert body["video"] is None + + +def test_nothing_is_decoded_at_registration( + client: TestClient, project: str, tmp_path: Path +) -> None: + """A file that is not an image registers fine and is reported at ingest instead.""" + (tmp_path / "made").mkdir(parents=True, exist_ok=True) + (tmp_path / "made" / "notes.txt").write_bytes(b"not an image") + + response = post_images(client, project, ("files", ("notes.txt", b"not an image", "text/plain"))) + + assert response.status_code == 201, response.text + + +def test_uploading_the_same_images_again_returns_the_same_source( + client: TestClient, project: str, tmp_path: Path +) -> None: + """Staging is content-addressed and registration is idempotent on the path.""" + first = post_images(client, project, image_part(tmp_path, "a.png", 1)) + second = post_images(client, project, image_part(tmp_path, "a.png", 1)) + + assert first.json()["id"] == second.json()["id"] + + +def test_a_traversing_filename_is_reduced_to_its_last_component( + client: TestClient, project: str, tmp_path: Path +) -> None: + path = write_image(tmp_path / "made" / "a.png", seed=1) + response = post_images( + client, project, ("files", ("../../escaped.png", path.read_bytes(), "image/png")) + ) + + assert response.status_code == 201, response.text + assert not (tmp_path / "escaped.png").exists() + + +def test_uploading_images_to_an_unknown_project_is_404(client: TestClient, tmp_path: Path) -> None: + response = post_images(client, str(uuid4()), image_part(tmp_path, "a.png")) + + assert response.status_code == 404 + assert response.json()["code"] == "PROJECT_NOT_FOUND" + + +# --- registering a clip ------------------------------------------------------ + + +def test_uploading_a_clip_registers_a_video_source( + client: TestClient, project: str, clip: Path +) -> None: + response = post_video(client, project, clip, extraction_fps=5) + + assert response.status_code == 201, response.text + body = response.json() + assert body["kind"] == "video" + assert body["name"] == "drive.mp4" + assert body["video"]["extraction_fps"] == 5 + # The rate the file was shot at, which is not the rate we cut it at. + assert body["video"]["fps"] == 10 + assert (body["video"]["width"], body["video"]["height"]) == CLIP_SIZE + + +def test_a_clip_registered_at_two_rates_is_two_sources( + client: TestClient, project: str, clip: Path +) -> None: + """The rate is part of what the source is — `docs/sources.md`.""" + slow = post_video(client, project, clip, extraction_fps=1) + fast = post_video(client, project, clip, extraction_fps=5) + + assert slow.json()["id"] != fast.json()["id"] + + +def test_the_default_rate_is_one_frame_per_second( + client: TestClient, project: str, clip: Path +) -> None: + response = post_video(client, project, clip) + + assert response.json()["video"]["extraction_fps"] == 1.0 + + +def test_a_non_positive_rate_is_422_before_anything_is_written( + client: TestClient, project: str, clip: Path +) -> None: + """`gt=0` on the form field, so the kernel's bare `ValueError` is unreachable.""" + response = post_video(client, project, clip, extraction_fps=0) + + assert response.status_code == 422 + assert response.json()["code"] == "VALIDATION_ERROR" + + +def test_uploading_something_that_is_not_a_video_is_422(client: TestClient, project: str) -> None: + response = client.post( + f"/projects/{project}/sources/video", + files={"file": ("notes.txt", b"not a video", "text/plain")}, + ) + + assert response.status_code == 422 + assert response.json()["code"] == "UNSUPPORTED_MEDIA" + + +# --- reading ----------------------------------------------------------------- + + +def test_listing_a_project_with_no_sources_is_an_empty_page( + client: TestClient, project: str +) -> None: + response = client.get(f"/projects/{project}/sources") + + assert response.status_code == 200 + assert response.json() == {"items": [], "total": 0} + + +def test_listing_returns_every_source_of_that_project( + client: TestClient, project: str, tmp_path: Path +) -> None: + post_images(client, project, image_part(tmp_path, "a.png", 1)) + post_images(client, project, image_part(tmp_path, "b.png", 2)) + + body = client.get(f"/projects/{project}/sources").json() + + assert body["total"] == 2 + assert len(body["items"]) == 2 + + +def test_reading_a_source_by_id(client: TestClient, project: str, tmp_path: Path) -> None: + created = post_images(client, project, image_part(tmp_path, "a.png", 1)).json() + + response = client.get(f"/sources/{created['id']}") + + assert response.status_code == 200 + assert response.json() == created + + +def test_a_source_never_publishes_its_path( + client: TestClient, project: str, tmp_path: Path +) -> None: + """It is a server-side path inside the workspace, and no client can use one.""" + body = post_images(client, project, image_part(tmp_path, "a.png", 1)).json() + + assert "path" not in body + assert set(body) == {"id", "project_id", "kind", "name", "registered_at", "video"} + + +def test_reading_an_unknown_source_is_404(client: TestClient) -> None: + response = client.get(f"/sources/{uuid4()}") + + assert response.status_code == 404 + assert response.json()["code"] == "SOURCE_NOT_FOUND" + + +def test_a_malformed_source_id_is_422_not_404(client: TestClient) -> None: + response = client.get("/sources/not-a-uuid") + + assert response.status_code == 422 + assert response.json()["code"] == "VALIDATION_ERROR" + + +def test_listing_sources_of_an_unknown_project_is_404(client: TestClient) -> None: + response = client.get(f"/projects/{uuid4()}/sources") + + assert response.status_code == 404 + assert response.json()["code"] == "PROJECT_NOT_FOUND" + + +# --- the guard --------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("method", "path"), + [ + ("POST", "/projects/{project}/sources/images"), + ("POST", "/projects/{project}/sources/video"), + ("GET", "/projects/{project}/sources"), + ("GET", "/sources/00000000-0000-0000-0000-000000000000"), + ("POST", "/sources/00000000-0000-0000-0000-000000000000/ingest-jobs"), + ("GET", "/sources/00000000-0000-0000-0000-000000000000/ingest-jobs"), + ], +) +def test_every_source_route_refuses_a_request_with_no_token( + client: TestClient, project: str, method: str, path: str +) -> None: + response = client.request(method, path.format(project=project), headers={"Authorization": ""}) + + assert response.status_code == 401 + assert response.json()["code"] == "UNAUTHORIZED" diff --git a/tests/server/test_uploads.py b/tests/server/test_uploads.py new file mode 100644 index 00000000..4733818d --- /dev/null +++ b/tests/server/test_uploads.py @@ -0,0 +1,120 @@ +"""The upload staging rules, without HTTP. + +`test_wire_models.py`'s shape: these are pure functions with invariants worth +pinning on their own, and asserting them through a request would test the +routing instead. What a route does with the result is `test_sources.py`. +""" + +from __future__ import annotations + +import hashlib +from io import BytesIO +from pathlib import Path + +import pytest +from fastapi import UploadFile + +from visionset.server.uploads import FALLBACK_NAME, UPLOADS_DIRNAME, safe_name, stage + + +def part(name: str | None, content: bytes = b"bytes") -> UploadFile: + return UploadFile(file=BytesIO(content), filename=name) + + +# --- naming ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + ("sent", "expected"), + [ + ("clip.mp4", "clip.mp4"), + ("../../etc/passwd", "passwd"), + ("/absolute/photo.png", "photo.png"), + (r"C:\Users\me\photo.png", "photo.png"), + (" spaced.png ", "spaced.png"), + ], +) +def test_a_filename_is_reduced_to_its_last_component(sent: str, expected: str) -> None: + assert safe_name(sent) == expected + + +@pytest.mark.parametrize("sent", [None, "", " ", ".", "..", "../", "evil\x00.png"]) +def test_a_filename_that_names_nothing_usable_falls_back(sent: str | None) -> None: + """Replaced, not refused: a badly named part is not a reason to lose a good upload.""" + assert safe_name(sent) == FALLBACK_NAME + + +# --- staging ----------------------------------------------------------------- + + +def test_a_staged_part_lands_under_uploads_with_its_bytes(tmp_path: Path) -> None: + staged = stage(tmp_path, [part("clip.mp4", b"video bytes")]) + + assert staged.directory.parent == tmp_path / UPLOADS_DIRNAME + assert staged.names == ("clip.mp4",) + assert staged.only.read_bytes() == b"video bytes" + + +def test_the_same_bytes_under_the_same_name_stage_to_the_same_directory(tmp_path: Path) -> None: + """What makes a repeated upload return the same `Source` rather than a second one.""" + first = stage(tmp_path, [part("clip.mp4", b"video bytes")]) + second = stage(tmp_path, [part("clip.mp4", b"video bytes")]) + + assert first.directory == second.directory + assert len(list((tmp_path / UPLOADS_DIRNAME).iterdir())) == 1 + + +def test_different_bytes_stage_apart(tmp_path: Path) -> None: + first = stage(tmp_path, [part("clip.mp4", b"one")]) + second = stage(tmp_path, [part("clip.mp4", b"two")]) + + assert first.directory != second.directory + + +def test_the_same_bytes_under_a_different_name_stage_apart(tmp_path: Path) -> None: + """The name is inside the digest: a file renamed is a different thing to offer.""" + first = stage(tmp_path, [part("a.png", b"same")]) + second = stage(tmp_path, [part("b.png", b"same")]) + + assert first.directory != second.directory + + +def test_the_order_parts_arrive_in_does_not_fork_the_directory(tmp_path: Path) -> None: + forward = stage(tmp_path, [part("a.png", b"one"), part("b.png", b"two")]) + backward = stage(tmp_path, [part("b.png", b"two"), part("a.png", b"one")]) + + assert forward.directory == backward.directory + assert forward.names == ("a.png", "b.png") + + +def test_two_parts_under_one_filename_both_survive(tmp_path: Path) -> None: + """A directory source reads its files by name, so collapsing them would drop one.""" + staged = stage(tmp_path, [part("photo.png", b"one"), part("photo.png", b"two")]) + + assert staged.names == ("photo.png", "photo-2.png") + assert (staged.directory / "photo.png").read_bytes() == b"one" + assert (staged.directory / "photo-2.png").read_bytes() == b"two" + + +def test_a_traversing_filename_cannot_escape_the_staging_directory(tmp_path: Path) -> None: + staged = stage(tmp_path, [part("../../escaped.png", b"nope")]) + + assert staged.names == ("escaped.png",) + assert not (tmp_path.parent / "escaped.png").exists() + + +def test_nothing_is_left_behind_under_a_staging_name(tmp_path: Path) -> None: + """The private directory is a rename away from being the published one.""" + stage(tmp_path, [part("a.png")]) + + assert not [ + entry for entry in (tmp_path / UPLOADS_DIRNAME).iterdir() if entry.name.startswith(".") + ] + + +def test_the_directory_is_named_for_the_whole_part_set(tmp_path: Path) -> None: + """Pinned rather than described, so the addressing scheme cannot drift silently.""" + staged = stage(tmp_path, [part("a.png", b"one")]) + + content = hashlib.sha256(b"one").hexdigest() + assert staged.directory.name == hashlib.sha256(f"a.png:{content}\n".encode()).hexdigest() diff --git a/uv.lock b/uv.lock index 3ce7c39c..7f98998a 100644 --- a/uv.lock +++ b/uv.lock @@ -1239,6 +1239,7 @@ dependencies = [ { name = "mcp" }, { name = "pillow" }, { name = "pydantic" }, + { name = "python-multipart" }, { name = "sqlalchemy" }, { name = "typer" }, { name = "uvicorn" }, @@ -1259,6 +1260,7 @@ requires-dist = [ { name = "mcp", specifier = ">=1.2" }, { name = "pillow", specifier = ">=11.0" }, { name = "pydantic", specifier = ">=2.7" }, + { name = "python-multipart", specifier = ">=0.0.9" }, { name = "sqlalchemy", specifier = ">=2.0" }, { name = "typer", specifier = ">=0.12" }, { name = "uvicorn", specifier = ">=0.30" },