suaveplan/openspec/specs/api-plugin-import/spec.md

<!-- markdownlint-disable MD024 -->

# Capability Spec: api-plugin-import

**Package**: `@suaveplan/api-plugin-import`
**Status**: SHIPPED

> Every requirement below carries a `**Verified by:**` line pointing at the
> `it(...)` titles in the package's test suite.

## Requirements

### Requirement: Stable IMPORT.* error registry

The package SHALL register exactly nine stable error codes under the `"IMPORT"` namespace via `defineCodes("IMPORT", […])`, exporting `IMPORT_CODES` and `ImportError extends SuaveplanError<ImportErrorCode>`. Every code SHALL be reachable from a real production code path; deletion to satisfy a coverage gate is forbidden.

#### Scenario: registry declares the nine canonical codes

- **WHEN** a caller imports `IMPORT_CODES`
- **THEN** the literal keys equal `{NOT_FOUND, FORBIDDEN, VALIDATION, FILE_TOO_LARGE, UNSUPPORTED_FORMAT, PROCESSING_FAILED, STORAGE_ERROR, TIMEOUT, INVALID_OPTIONS}` and every value is namespaced `"IMPORT.<KEY>"`

#### Scenario: every code is thrown from a production path

- **WHEN** the test suite drives the full feature surface
- **THEN** every key in `IMPORT_CODES` appears as `error.code` on at least one captured throw

**Verified by:** `src/error-codes/error-codes.test.ts::"declares all nine IMPORT codes"` AND `src/error-codes/error-codes.test.ts::"every IMPORT code is reachable from a production path"`

---

### Requirement: CSV parsing — quoted, custom-delimiter, header-aware

The package SHALL parse CSV input into `Record<string, string>[]` rows respecting `"`-quoted fields (including escaped `""` quotes and embedded newlines), custom delimiters, optional headers (defaulting to `column_N` placeholders when absent), and SHALL expose a streaming variant (`streamParseCsv`) that surfaces per-row validation errors without aborting the run.

#### Scenario: header-first CSV with quoted comma

- **WHEN** input is `name,city\n"Doe, Jane",NYC`
- **THEN** the result is `[{ name: "Doe, Jane", city: "NYC" }]`

#### Scenario: streaming parser collects validation errors

- **WHEN** `streamParseCsv` is invoked with `validate` returning `"bad"` on row 1
- **THEN** the result reports `errors.length === 1` with `row: 1, message: "bad"` and `validRows` excludes that row

**Verified by:** `src/parsers/csv.test.ts::"parses quoted fields with embedded commas"` AND `src/parsers/csv.test.ts::"streamParseCsv surfaces per-row validate errors"`

---

### Requirement: JSON parser rejects non-arrays

The package SHALL accept a JSON-array input string and return its elements as rows; non-array input SHALL throw with a message starting `JSON data must be an array`.

#### Scenario: valid JSON array

- **WHEN** input is `[{"a":1}]`
- **THEN** `parseJson(input)` returns `[{ a: 1 }]`

#### Scenario: invalid non-array input

- **WHEN** input is `{"a":1}`
- **THEN** `parseJson(input)` throws an `Error` whose message contains `JSON data must be an array`

**Verified by:** `src/parsers/json.test.ts::"returns array elements as rows"` AND `src/parsers/json.test.ts::"throws when input is not a JSON array"`

---

### Requirement: XLSX (TSV-style) parser

The package SHALL provide `parseXlsx` that treats `\t`-separated lines as rows with optional header inference; binary `.xlsx` decoding is explicitly out of scope and SHALL be documented as a limitation.

#### Scenario: TSV input with header row

- **WHEN** input is `a\tb\n1\t2`
- **THEN** `parseXlsx(input)` returns `{ headers: ["a","b"], rows: [{a:"1",b:"2"}], totalRows: 1 }`

**Verified by:** `src/parsers/xlsx.test.ts::"parses TSV with headers"` AND `src/parsers/xlsx.test.ts::"synthesises column_N headers when hasHeaders is false"`

---

### Requirement: Transform pipeline — coerce, validate, compute

The package SHALL provide a declarative `transformRow` / `transformRows` pipeline that applies `TransformMapping[]` (source→target with optional type coercion, default value, required flag, transform fn), `ComputedField[]` (target derived from named source columns), and `FieldValidation[]` (`required` / `min` / `max` / `pattern` / `custom`). `coerceType` SHALL succeed/fail explicitly for `number` / `boolean` / `date` / pass-through for `string`. Failed transforms SHALL push a `TransformError` and skip the field, never throw.

#### Scenario: required field missing produces TransformError

- **WHEN** `transformRow({}, 0, { mappings: [], validations: [{ field: "x", rule: "required" }] })` runs against an empty `result`
- **THEN** the result is `{ success: false, errors: [{ row: 0, field: "x", message: '"x" is required', value: undefined }] }`

#### Scenario: type coercion failure marks row failed

- **WHEN** `transformRow({ x: "abc" }, 0, { mappings: [{ sourceColumn: "x", targetField: "x", type: "number" }] })`
- **THEN** `errors` contains one entry with `message: "Cannot coerce to number"` and `result.x` is absent

#### Scenario: computed field aggregates source columns

- **WHEN** `transformRow({ a: 1, b: 2 }, 0, { mappings: [], computedFields: [{ targetField: "sum", sourceColumns: ["a","b"], compute: v => Number(v.a) + Number(v.b) }] })`
- **THEN** `row.sum === 3`

**Verified by:** `src/transform/pipeline.test.ts::"validateField required produces error for missing value"` AND `src/transform/pipeline.test.ts::"coerceType number returns ok:false for non-numeric input"` AND `src/transform/pipeline.test.ts::"computed field is evaluated from named source columns"`

---

### Requirement: ImportProgressTracker — rows/sec, ETA, percent

The package SHALL ship an `ImportProgressTracker` that records single + batch successes/failures, computes `percentComplete` (rounded int), `rowsPerSecond` (2dp), and `estimatedRemainingMs`, fans snapshots to registered listeners, and SHALL expose an `isComplete()` predicate that returns `true` iff `processedRows >= totalRows`.

#### Scenario: batch success updates totals and notifies listeners

- **WHEN** `tracker.recordBatchSuccess(5)` runs after a listener is registered
- **THEN** the listener receives a snapshot with `successCount: 5` and `processedRows: 5`

#### Scenario: zero totalRows yields zero percent

- **WHEN** `new ImportProgressTracker("j", 0).snapshot()` runs
- **THEN** `percentComplete === 0` and `rowsPerSecond` is finite (`0` or a small positive number)

**Verified by:** `src/progress/tracker.test.ts::"recordBatchSuccess fans a snapshot with the cumulative counts"` AND `src/progress/tracker.test.ts::"snapshot guards against zero totalRows"`

---

### Requirement: Async ImportStoreAdapter + MemoryImportStore

The package SHALL declare an async `ImportStoreAdapter` interface with `create` / `get` / `update` / `delete` / `list` and SHALL ship a `MemoryImportStore` implementing it. `update` on a missing id SHALL throw with `Job "<id>" not found`. `list` SHALL filter by `status` and respect `limit` / `offset`.

#### Scenario: round-trip create + get

- **WHEN** a record is created and then fetched by id
- **THEN** the fetched value equals the created one

#### Scenario: update on missing id throws

- **WHEN** `update("missing", {...})` runs on an empty store
- **THEN** it rejects with an error whose message matches `Job "missing" not found`

**Verified by:** `src/store/memory.test.ts::"round-trips a record via create and get"` AND `src/store/memory.test.ts::"throws when updating a missing id"`

---

### Requirement: ImportService — preview, startImport, owner-scoped jobs

The package SHALL ship a canonical `ImportService extends EventEmitter<ImportEventMap>` with `preview(format, data)`, `startImport(config, { ownerId })`, `getJob(id, ownerId)`, `getResults(id, ownerId)`, `listJobs(ownerId)`, `deleteJob(id, ownerId)`. Foreign-owner reads SHALL throw `IMPORT.NOT_FOUND` (anti-enumeration; never `IMPORT.FORBIDDEN` on lookup misses). Format validation SHALL throw `IMPORT.UNSUPPORTED_FORMAT`; parse failures SHALL throw `IMPORT.PROCESSING_FAILED`; missing mappings SHALL throw `IMPORT.VALIDATION`. The service SHALL emit `import.started`, `import.progress`, `import.completed`, `import.failed`, `import.row_error`, `import:row:processed` per the documented `ImportEventMap`. Successful runs SHALL increment `import.jobs_total{outcome:"success"}` and `import.rows_total` OTel counters.

#### Scenario: preview returns column auto-mapping

- **WHEN** `preview("csv", "a,b\n1,2")` runs
- **THEN** the returned `mappings` has `sourceColumn === targetField` for every detected column

#### Scenario: dry-run does not persist results

- **WHEN** `startImport({ format, data, mappings, dryRun: true }, { ownerId })` succeeds
- **THEN** `getResults(jobId, ownerId)` returns `[]`

#### Scenario: foreign-owner lookup returns 404 not 403

- **WHEN** `getJob(jobId, "other-user")` runs on a job owned by `"alice"`
- **THEN** it throws `ImportError` with `code === IMPORT_CODES.NOT_FOUND`

#### Scenario: parse failure surfaces as IMPORT.PROCESSING_FAILED

- **WHEN** `startImport({ format: "json", data: "not-json", mappings: [{ sourceColumn: "x", targetField: "x" }] })` runs
- **THEN** it throws `ImportError` with `code === IMPORT_CODES.PROCESSING_FAILED`

**Verified by:** `src/service/service.test.ts::"preview returns identity column mappings"` AND `src/service/service.test.ts::"dryRun does not persist imported rows"` AND `src/service/service.test.ts::"foreign owner lookup returns NOT_FOUND"` AND `src/service/service.test.ts::"parse failure throws PROCESSING_FAILED"`

---

### Requirement: 5 Hono routes, auth-gated, owner-scoped

The package SHALL ship `createImportRoutes({ service, instruments })` returning an `OpenAPIHono` with five routes — `POST /preview`, `POST /`, `GET /`, `GET /{id}`, `DELETE /{id}`. Every route SHALL be wrapped in `requireAuth()` from `@suaveplan/api-plugin-auth`. Job-scoped routes SHALL read `c.get("user").id` as `ownerId`. Lookup misses on jobs owned by a different user SHALL return HTTP 404 with RFC 7807 problem+json (anti-enumeration).

#### Scenario: anonymous request to POST / returns 401

- **WHEN** a request reaches `POST /` with `c.get("user") === null`
- **THEN** the response is HTTP 401 with `content-type: application/problem+json`

#### Scenario: foreign-owner GET /{id} returns 404

- **WHEN** alice owns job `j1` and bob calls `GET /j1`
- **THEN** the response is HTTP 404 with problem+json

**Verified by:** `src/routes/build-routes.test.ts::"POST / returns 401 when caller is anonymous"` AND `src/routes/build-routes.test.ts::"GET /{id} returns 404 when the job is owned by a different user"`

---

### Requirement: RFC 7807 problem+json mapping

The package SHALL provide `respondProblemJson(c, err)` that renders any `SuaveplanError` as RFC 7807 `application/problem+json`, reading `error.context.statusCode` when present and defaulting to 500. Unknown throwables SHALL render as a generic 500.

#### Scenario: SuaveplanError with statusCode 404

- **WHEN** the error carries `context.statusCode = 404`
- **THEN** the response status is 404 and the body is JSON with the `code` and `message` fields

#### Scenario: unknown throwable yields 500

- **WHEN** a plain `string` is thrown
- **THEN** the response status is 500 with `content-type: application/problem+json`

**Verified by:** `src/routes/problem-json.test.ts::"renders SuaveplanError with the context statusCode"` AND `src/routes/problem-json.test.ts::"renders unknown throwables as a generic 500"`

---

### Requirement: OTel instruments bundle

The package SHALL export `getImportInstruments()` returning a cached bundle of four OpenTelemetry instruments — `import.jobs_total` (Counter, labelled by `outcome`), `import.rows_total` (Counter, labelled by `outcome`), `import.row_errors_total` (Counter), `import.duration_ms` (Histogram). `resetImportInstruments()` SHALL drop the cache so subsequent calls rebind against the current global `MeterProvider`. Production code paths SHALL emit on every `startImport`.

#### Scenario: cached bundle returns same reference

- **WHEN** `getImportInstruments()` is called twice without reset
- **THEN** the second call returns the same object

#### Scenario: emit after install collects metrics

- **WHEN** a fresh `MeterProvider` is installed, `resetImportInstruments()` runs, and a counter is incremented
- **THEN** the in-memory exporter contains the metric

**Verified by:** `src/otel/instruments.test.ts::"getImportInstruments returns a cached bundle"` AND `src/otel/instruments.test.ts::"resetImportInstruments rebinds against a fresh MeterProvider"`

---

### Requirement: createImportPlugin descriptor

The package SHALL export `createImportPlugin(options?)` returning a typed `PluginDescriptor<ImportPluginConfig>` whose `name: "import"`, `version: "0.1.0"`, `optionalDependencies: ["auth", "events", "database"]`, and `tags: ["import", "csv", "json", "xlsx"]`. Invalid options SHALL throw `ImportError(IMPORT.INVALID_OPTIONS)` with the `ZodError` as `cause`. `onInit` SHALL register the `ImportService` instance on `ServiceRegistry["importService"]`; `onReady` SHALL mount routes under the configured `routePrefix`; `onClose` SHALL clear all listeners on the service.

#### Scenario: bad options throws INVALID_OPTIONS

- **WHEN** `createImportPlugin({ routePrefix: "" })` runs
- **THEN** it throws `ImportError` with `code === IMPORT_CODES.INVALID_OPTIONS` and `cause instanceof ZodError`

#### Scenario: onInit registers the service slot

- **WHEN** `descriptor.onInit(ctx)` runs
- **THEN** `ctx.getService("importService")` is defined and is an `ImportService` instance

#### Scenario: onReady mounts routes under routePrefix

- **WHEN** `descriptor.onReady(ctx)` runs with `routePrefix: "/imports"`
- **THEN** the host `app.routes` contains an entry whose path starts with `/imports/`

**Verified by:** `src/plugin/plugin.test.ts::"throws INVALID_OPTIONS on bad config"` AND `src/plugin/plugin.test.ts::"onInit registers the importService slot"` AND `src/plugin/plugin.test.ts::"onReady mounts routes under the configured routePrefix"` AND `src/plugin/plugin.spec-conformance.test.ts::"descriptor carries the canonical metadata"`

---

### Requirement: Subpath exports per feature module

The package SHALL declare a subpath export in `package.json#exports` for every feature module so consumers can tree-shake. The package root SHALL re-export public symbols via explicit named re-exports; feature-module barrels SHALL NOT use `export *`.

#### Scenario: subpath export count matches feature count

- **WHEN** the `exports` map is enumerated
- **THEN** it contains entries for `./error-codes/error-codes`, `./parsers/csv`, `./parsers/json`, `./parsers/xlsx`, `./transform/pipeline`, `./progress/tracker`, `./store/memory`, `./service/service`, `./routes/problem-json`, `./routes/build-routes`, `./otel/instruments`, `./schemas/options`, `./schemas/schemas`, `./plugin/plugin`, `./types/types`

#### Scenario: feature barrels use explicit named re-exports

- **WHEN** the file tree is scanned for `export *` in `src/<feature>/index.ts`
- **THEN** no matches exist (the conformance script enforces this)

**Verified by:** `src/plugin/plugin.spec-conformance.test.ts::"package exposes a subpath export for every feature module"` AND `src/plugin/plugin.spec-conformance.test.ts::"every feature barrel uses explicit named re-exports"`

---

### Requirement: SurrealImportStore — durable ImportStoreAdapter

The package SHALL ship `SurrealImportStore implements ImportStoreAdapter` at `src/store/surreal/surreal-import-store.ts`, providing the same `create` / `get` / `update` / `delete` / `list` surface as `MemoryImportStore` against a `SurrealAdapter` backend from `@suaveplan/surrealdb`. All SurrealQL placeholders SHALL be positional (`$p0`, `$p1`, …) per the canonical pattern in `@suaveplan/api-plugin-events`. The constructor SHALL accept a `SurrealImportStoreConfig` validated through Zod at the boundary; an invalid config SHALL throw `ImportError` with `code === IMPORT_CODES.INVALID_OPTIONS` and `cause instanceof ZodError`. The store SHALL accept a custom `table` name (default `"import_job"`). `create` SHALL upsert the record by stringified id; `update` on a missing id SHALL throw an error whose message contains `Job "<id>" not found` (parity with `MemoryImportStore`). Adapter-layer failures SHALL be wrapped as `ImportError(IMPORT_CODES.STORAGE_ERROR)` with the underlying error as `cause`. The store SHALL increment `import.store.writes_total{outcome:"success"|"error"}` on every `create` / `update` / `delete`, and `import.store.reads_total{outcome:"success"|"error"}` on every `get` / `list`.

#### Scenario: create emits a CREATE SET statement with positional params

- **WHEN** `surrealStore.create(record)` runs with `record.id = "j1"`
- **THEN** the issued SurQL begins with `CREATE import_job:j1 SET` and the parameter array contains the record's `ownerId`, `status`, numeric counters, and timestamp strings in the order they appear in the `SET` clause

#### Scenario: get returns undefined when no row exists

- **WHEN** `surrealStore.get("missing")` runs against an empty backend
- **THEN** the call resolves to `undefined` and `import.store.reads_total{outcome:"success"}` is incremented (a miss is still a successful read)

#### Scenario: update on missing id throws Job "<id>" not found

- **WHEN** `surrealStore.update("missing", { status: "completed" })` runs on a store with no record `"missing"`
- **THEN** the call rejects with an `Error` whose message contains `Job "missing" not found`

#### Scenario: list filters by status, ownerId, limit, offset

- **WHEN** `surrealStore.list({ status: "completed", ownerId: "alice", limit: 10, offset: 5 })` runs
- **THEN** the SurQL contains `WHERE status = $p0 AND owner_id = $p1`, `LIMIT $p2`, `START $p3` and the parameter array is `["completed", "alice", 10, 5]`

#### Scenario: adapter failure surfaces as STORAGE_ERROR

- **WHEN** the underlying `SurrealAdapter.query` throws
- **THEN** the store's call rejects with an `ImportError` whose `code === IMPORT_CODES.STORAGE_ERROR` and `cause` is the original throwable

#### Scenario: invalid config throws INVALID_OPTIONS

- **WHEN** `new SurrealImportStore({ db: adapter, table: "" })` runs
- **THEN** it throws `ImportError` with `code === IMPORT_CODES.INVALID_OPTIONS` and `cause instanceof ZodError`

**Verified by:** `src/store/surreal/surreal-import-store.test.ts::"create issues a CREATE SET statement with positional params"` AND `src/store/surreal/surreal-import-store.test.ts::"get returns undefined when no row exists"` AND `src/store/surreal/surreal-import-store.test.ts::"update on missing id throws Job not found"` AND `src/store/surreal/surreal-import-store.test.ts::"list filters by status, ownerId, limit, offset"` AND `src/store/surreal/surreal-import-store.test.ts::"adapter failure surfaces as STORAGE_ERROR on read"` AND `src/store/surreal/surreal-import-store.test.ts::"invalid config throws INVALID_OPTIONS"`

---

### Requirement: SurrealDB schema migration

The package SHALL ship `surql/tables/import_job.surql` defining the `import_job` table SCHEMAFULL with fields `owner_id (string)`, `status (string)` constrained to the `ImportJobStatus` enum, `total_rows (int)`, `processed_rows (int)`, `success_count (int)`, `error_count (int)`, `errors (array)`, `metadata (option<object>)`, `created_at (datetime)`, `updated_at (datetime)`, `completed_at (option<datetime>)`, plus indexes on `owner_id`, `status`, and `created_at`. The migration SHALL use the canonical `DEFINE TABLE OVERWRITE` / `DEFINE FIELD OVERWRITE` form mirroring `packages/api/api-plugin-events/surql/tables/event.surql`. Hosts MAY apply the migration via `@suaveplan/api-plugin-database`'s migration runner or directly via `adapter.query(readFileSync(...))`.

#### Scenario: migration declares the canonical schema

- **WHEN** the integration test reads and applies `surql/tables/import_job.surql`
- **THEN** the SurrealDB instance subsequently accepts a `CREATE import_job:j1 SET owner_id = 'alice', status = 'created', …` statement and rejects a `status = 'bogus'` write

**Verified by:** `tests/integration/surreal-import-store.test.ts::"migration enables canonical writes and rejects invalid status"`

---

### Requirement: Integration round-trip against a real SurrealDB instance

The package SHALL ship `tests/integration/surreal-import-store.test.ts` using `SurrealDBFixture` from `@suaveplan/testing-integration-surrealdb`. The test SHALL apply the `.surql` migration once in `beforeAll`, then `create` a record, `get` it back, `update` it, `list` with a `status` filter, and `delete` it — all against the real container. The integration test SHALL also assert that `update("missing", …)` rejects with the `Job "missing" not found` message and that the `errors` array round-trips intact.

#### Scenario: full round-trip against testcontainer

- **WHEN** the integration test runs against a started `SurrealDBFixture`
- **THEN** each store operation completes successfully and the final `get` after `delete` returns `undefined`

**Verified by:** `tests/integration/surreal-import-store.test.ts::"round-trips a record through the real SurrealDB instance"` AND `tests/integration/surreal-import-store.test.ts::"update on missing id surfaces the canonical message against a real instance"`

---

### Requirement: OTel store counters

The package SHALL extend `getImportInstruments()` to expose two additional counters — `import.store.writes_total` and `import.store.reads_total` — both labelled by `outcome`. The counter pair SHALL be cached on the same bundle as the existing four instruments and SHALL rebind through `resetImportInstruments()`. `SurrealImportStore` SHALL emit on every adapter call; `MemoryImportStore` is out of scope for this requirement (it is an in-process test/dev adapter without observable backend health).

#### Scenario: bundle exposes the two new counters

- **WHEN** a caller invokes `getImportInstruments()`
- **THEN** the returned object exposes `storeWritesTotal` and `storeReadsTotal` Counter instruments in addition to the existing four

**Verified by:** `src/otel/instruments.test.ts::"bundle exposes the store write and read counters"`

---

### Requirement: Subpath export for the SurrealDB store

The package SHALL declare the subpath export `./store/surreal/surreal-import-store` in `package.json#exports` so consumers can tree-shake the SurrealDB store without pulling in the in-memory implementation. The `src/store/surreal/index.ts` barrel SHALL use explicit named re-exports (no `export *`).

#### Scenario: subpath export exists and the barrel uses named exports

- **WHEN** the package's `exports` map is enumerated and the new barrel is read
- **THEN** the map contains `./store/surreal/surreal-import-store` and the barrel contains no `export *` lines

**Verified by:** `src/plugin/plugin.spec-conformance.test.ts::"package exposes a subpath export for the SurrealDB store"`

---

### Requirement: SSE progress push — `GET /imports/:id/progress`

The package SHALL expose `GET /imports/:id/progress` returning
`Content-Type: text/event-stream; charset=utf-8`, mounted under the existing
`requireAuth()` middleware. Frames SHALL be driven directly off the existing
`ImportService.bus` events (`import.progress`, `import.completed`, `import.failed`);
NO polling loop over the store is permitted. Every frame SHALL carry the canonical
`ImportProgress` JSON payload. Three SSE event kinds SHALL be emitted: `progress`
(every `import.progress` tick), `complete` (terminal, on `import.completed`),
and `failed` (terminal, on `import.failed`). The stream SHALL close cleanly after
a terminal event. The handler SHALL respect the request's `AbortSignal` (client
disconnect) by detaching the bus listener and incrementing
`import.sse.disconnected{reason: "client"}`. Foreign-owner or missing-job lookups
SHALL return HTTP 404 RFC 7807 problem+json with `code === IMPORT.NOT_FOUND`
(anti-enumeration parity with `GET /:id`). Anonymous requests SHALL return HTTP
401 problem+json via the existing auth middleware.

The package SHALL extend `getImportInstruments()` with three additional counters:
`import.sse.connected` (Counter), `import.sse.events_sent` (Counter, labelled by
`kind ∈ "progress" | "complete" | "failed"`), `import.sse.disconnected`
(Counter, labelled by `reason ∈ "complete" | "failed" | "client"`). The counters
SHALL be cached on the same bundle as the existing six instruments and SHALL
rebind through `resetImportInstruments()`.

#### Scenario: opens a text/event-stream and emits an initial progress frame

- **WHEN** an authenticated caller `GET /imports/:id/progress` for a job they own
- **THEN** the response is HTTP 200 with `content-type: text/event-stream; charset=utf-8`
  and the first frame is `event: progress\ndata: <ImportProgress JSON>\n\n`

#### Scenario: emits a progress frame for every bus progress event

- **WHEN** the service emits `import.progress` while the client is connected
- **THEN** the stream writes a frame whose first non-id line is `event: progress`
  and whose `data:` line decodes to an `ImportProgress` object with the matching
  `processedRows`

#### Scenario: emits a complete frame and closes on import.completed

- **WHEN** the service emits `import.completed`
- **THEN** the stream writes a frame whose first non-id line is `event: complete`
  and then closes; subsequent reads from the stream return `done: true`

#### Scenario: emits a failed frame and closes on import.failed

- **WHEN** the service emits `import.failed`
- **THEN** the stream writes a frame whose first non-id line is `event: failed`
  and then closes

#### Scenario: returns 404 problem+json for a foreign-owner job

- **WHEN** alice owns job `j1` and bob requests `GET /imports/j1/progress`
- **THEN** the response is HTTP 404 with `content-type: application/problem+json`
  and a body whose `code === "IMPORT.NOT_FOUND"`

#### Scenario: returns 401 problem+json when the caller is anonymous

- **WHEN** an anonymous caller requests `GET /imports/:id/progress`
- **THEN** the response is HTTP 401 with `content-type: application/problem+json`

#### Scenario: detaches the listener and increments disconnect counter on client abort

- **WHEN** the client aborts the request before a terminal event
- **THEN** the handler unsubscribes from the bus and increments
  `import.sse.disconnected{reason:"client"}`

#### Scenario: increments import.sse counters on connect, send, and disconnect

- **WHEN** a connection opens, two progress events fire, and a terminal `complete`
  event closes the stream
- **THEN** `import.sse.connected` increments by 1, `import.sse.events_sent{kind}`
  increments by 3 (2× `progress` + 1× `complete`), and
  `import.sse.disconnected{reason:"complete"}` increments by 1

**Verified by:**
`src/routes/progress-sse/progress-sse.test.ts::"opens a text/event-stream and emits an initial progress frame"`
AND `src/routes/progress-sse/progress-sse.test.ts::"emits a progress frame for every bus progress event"`
AND `src/routes/progress-sse/progress-sse.test.ts::"emits a complete frame and closes on import.completed"`
AND `src/routes/progress-sse/progress-sse.test.ts::"emits a failed frame and closes on import.failed"`
AND `src/routes/progress-sse/progress-sse.test.ts::"returns 404 problem+json for a foreign-owner job"`
AND `src/routes/progress-sse/progress-sse.test.ts::"returns 401 problem+json when the caller is anonymous"`
AND `src/routes/progress-sse/progress-sse.test.ts::"detaches the listener and increments disconnect counter on client abort"`
AND `src/routes/progress-sse/progress-sse.test.ts::"increments import.sse counters on connect, send, and disconnect"`
AND `src/otel/instruments.test.ts::"bundle exposes the SSE counters"`

<!-- markdownlint-enable MD024 -->