gitea_admin/sientia-dataops-laborious_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add docs

Browse Source
This commit is contained in:
vitor-aignosi
2026-08-26 16:15:14 -03:00
commit eba812310d
4 changed files with 864 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
docs/model-flavors.md Normal file
View File

@@ -0,0 +1,112 @@
# Model flavors
The *flavor* is how this runtime is told to deserialize a model. It is **configuration, never
discovery**: the value comes from the model document in the `models` collection and the code never
reads the artifact's `MLmodel` manifest to guess it.
```json
{
"model_config": {
"transform_flavor": "sklearn",
"predict_flavor": "joblib",
"retention_minutes": 60,
"target": "SE"
}
}
```
`predict_flavor` drives `load_predict_model` (artifact `prediction_model`); `transform_flavor` drives
`load_transform_model` (artifact `data_model`). Both default to `sklearn` when the key is absent.
## Accepted values
| Flavor | Read | Write (retrain) | How it loads |
|---|---|---|---|
| `sklearn` | yes | yes | `mlflow.sklearn.load_model(<uri>)` |
| `pyfunc` | yes | yes | `mlflow.pyfunc.load_model(<uri>)`; the wrapper path also unwraps `._model_impl.python_model` |
| `pytorch` | yes | yes | `mlflow.pytorch.load_model(<uri>)` |
| `joblib` | yes | yes | downloads the artifact and `joblib.load`s its pickle — **no MLflow flavor loader involved** |
Anything else raises `ValueError` with `INVALID_FLAVOR_MESSAGE`, **before** any artifact is resolved
or downloaded. The same message and the same four values govern reading and writing.
**There is no fallback between flavors.** A failing loader propagates: one attempt, one cause. A
model configured with the wrong flavor fails visibly instead of being rescued silently.
## The `joblib` flavor
Some models the pipeline produces do not load through an MLflow flavor loader even though the
artifact is intact: the `.pkl` is in the compressed form `joblib.dump` writes, which MLflow's
`pickle.load` does not read. Declaring `joblib` is how such a model is served.
**Reading.** Resolve the Production run, `client.download_artifacts(run_id, <artifact>)`, take the
**first top-level `.pkl` in name order**, and `joblib.load` it. The listing does not recurse, so a
nested wrapper artifact (`artifacts/*.pkl`) is never deserialized in place of the model. No
`MLmodel` is parsed and no `models:/<name>/production` URI is built.
**The artifact's `code/` goes on `sys.path` first.** A model whose class lives inside the artifact
would otherwise raise `ModuleNotFoundError` at unpickle time. Before `joblib.load`, the flavor
prepends `<artifact>/code` to `sys.path` when that directory exists — the directory itself, matching
MLflow's own convention where `code/utils/…` imports as the package `utils`. The entry is not
duplicated and is **not** removed after the load, because the deserialized object may import
lazily. A missing `code/` is not an error, and no extra download is performed: `download_artifacts`
already brings `code/` along.
> `sys.path` is process-global. Two different models that embed a package with the same name
> (`utils/`, typically) and are served by the same worker resolve to whichever entered first — the
> second silently runs the first one's code. This is pre-existing behaviour of MLflow's own pyfunc
> path with the same artifact layout; the mitigation, if it ever bites, is one worker per model.
**Writing / retraining.** `joblib` is a write flavor too, so a joblib model retrains. `log_model`
writes `joblib.dump(model, 'model.pkl', compress=3)` under the artifact path and re-logs the source
artifact's `code/` to `<artifact>/code`, so the retrained model stays deserializable. Registration
and promotion to `Production` are unchanged.
**The written artifact has no `MLmodel` manifest.** Only this runtime's `joblib` flavor reads it:
`mlflow.sklearn.load_model` and `models:/<name>/production` both fail on it. A model retrained in
joblib **stays** joblib — `predict_flavor` needs no edit after a retrain, and no consumer outside
this runtime (the MLflow UI, MLflow serving, another service) recognizes that artifact as a model.
**What `joblib` is for.** A pickled object exposing `predict(data)` — and `fit(data)` if it is
retrained. It is **not** for a bare pyfunc `PythonModel` wrapper: `get_prediction_data` and
`get_cached_operation` call `predict(data)` for every non-pyfunc flavor, so a wrapper expecting
`predict(context, model_input)` raises `TypeError` at predict and retrain time.
**Observability.** A joblib load is an ordinary load: `MODEL_READ_LAG` / `MODEL_READ_COUNT` on
success and `MODEL_READ_ERROR_COUNT` on failure, under the base `operation_type`
(`load_predict_model` / `load_transform_model`). A joblib write emits the usual `MODEL_WRITE_*`
under `log_model`. No suffixed operation type, no new metric, no new label.
## Switching a model to `joblib`
The runtime does not migrate documents. Editing the `models` collection is an operator step, and the
order matters:
1. **Inventory first, before the deploy.** Collect the models that fail in their configured flavor
loader — run predict/transform and note which ones raise. There is no automatic discovery
afterwards: once deployed, such a model simply fails, and the failure names the flavor loader,
not the fix.
2. **Edit the document.** Set `model_config.predict_flavor: 'joblib'` and/or
`model_config.transform_flavor: 'joblib'` for each model in the inventory. Nothing else changes;
models already on a working flavor need no edit and behave identically.
3. **Imported models arrive ready.** The `import_model` workflow writes
`predict_flavor: 'joblib'` itself — see [`model-import.md`](model-import.md) § 12.
4. **Check the first retrain.** In MLflow, confirm the new run carries
`prediction_model/model.pkl` **and** `prediction_model/code/` when the source model had a
`code/` directory, and that the promoted version loads on the next predict.
5. **Rolling back is not just a revert.** Reverting the code leaves any artifact a joblib retrain
already wrote in MLflow: that Production version stops being loadable, so the rollback also means
promoting the previous version back in the registry, and setting the documents' flavors back —
manual, in reverse order.
## Known gap: how the run id is resolved
`get_model_run_id` resolves the Production run with `source.split('/')[2]` on the registered
version's `source`. **This has not been confirmed against the real tracking server.** No artifact
scheme reproduced locally puts the run id in that position (a file store yields `''`, S3 yields the
bucket, `mlflow-artifacts:/` yields the experiment id), yet predict works in production — so the
real `source` has a shape none of the local probes produced.
This is not specific to `joblib`: every flavor that resolves a run goes through the same line, and
the joblib read path depends on it. Confirm it with a **read-only** query against the tracking
server before relying on it, ideally before the next deploy.

386
docs/model-import.md Normal file
View File

@@ -0,0 +1,386 @@
# Importing a model (`import_model`)
The `import_model` workflow turns an encrypted `.sientia` object in MinIO into a registered MLflow
model plus one document in the MongoDB `models` collection — or into a row that says, in a sentence a
person can read, why it did not.
This document is the contract for the two sides that talk to it: the **frontend**, which uploads the
object and builds the request, and the **BFF**, which owns the import log row. Everything here is
enforced by code in this repository; where a statement is about a column or a constraint, the
authority is the migration named in §7.
Format of the file itself: [`sientia-bundle-format.md`](sientia-bundle-format.md).
---
## 1. What the caller does, in order
1. **Upload** the `.sientia` object to the import bucket, under the configured prefix (default
`imported_models/`). The key must end in `.sientia`.
2. **Compute `expected_digest`** — the SHA-256, lowercase hex, of the **exact bytes that were
uploaded**. Not of a re-read, not of a re-serialisation: of what went over the wire.
3. **Encrypt the bundle password** into a `password_envelope` (§3).
4. **Insert the import log row** (§5) and take its `id`.
5. **Start the workflow** on `import_model-{RUNTIME}-queue` with the input of §2.
The workflow claims the row, opens the bundle, provisions MLflow, writes the model document, cleans
up and records one terminal verdict. It never inserts a row, never promotes a version and never
activates a model.
## 2. The workflow input
```json
{
"import_run_id": 41,
"object_key": "imported_models/sales_forecast_v3.sientia",
"expected_digest": "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08",
"password_envelope": "Base64OfNonceCiphertextTag==",
"bucket": "sientia"
}
```
| Field | Required | Meaning |
|---|---|---|
| `import_run_id` | yes | primary key of the `public.experiment_run` row the BFF inserted. Must be a positive integer |
| `object_key` | yes | the uploaded object's key. Must match the row's `file_name` when the row carries one |
| `expected_digest` | **yes** | SHA-256 of the uploaded bytes, 64 lowercase hex characters. There is no path that skips the comparison: a missing value fails at `received`, a mismatching one at `download`, before the header is parsed |
| `password_envelope` | yes | the bundle password, encrypted (§3). A plaintext `password` field is **refused** |
| `bucket` | no | defaults to `IMPORT_BUNDLE_BUCKET` |
**Fields that are refused rather than ignored**: `project_id`, `project_name`,
`provisioning_target`, `schema`, `table_name`, `status_table`, `password`. An input carrying any of
them fails at `received`, so no caller can believe it recorded something the import does not record
(see §10).
## 3. The password envelope
AES-256-GCM, and the ciphertext layout is what the WebCrypto API returns natively:
```
envelope = base64( nonce(12 bytes) || ciphertext || tag(16 bytes) )
```
`crypto.subtle.encrypt({name: 'AES-GCM', iv: nonce}, key, plaintext)` returns `ciphertext || tag`
already concatenated, so the browser side is: generate 12 random bytes, encrypt, concatenate the
nonce in front, base64 the result.
The key is a 32-byte AES-256 key shared with the worker through `IMPORT_PASSWORD_KEY` (base64). The
worker decrypts the envelope **inside the activity that uses the password**, so no plaintext password
is ever an activity input, an activity result or a workflow variable — and therefore never appears in
the Temporal event history. A test asserts that for every failure path.
A failure to open the envelope is recorded as a `decryption` failure, distinct in the logs from a
wrong *bundle* password but sharing its step, because the step says where the import stopped, not who
is at fault.
## 4. Environment
| Variable | Default | What it is |
|---|---|---|
| `IMPORT_MODEL_ENABLED` | unset | registers the fifth (import) worker. Truthy values: `1`, `true`, `yes`, `on` |
| `IMPORT_PASSWORD_KEY` | — | base64 of the 32-byte AES-256-GCM key that wraps bundle passwords |
| `IMPORT_BUNDLE_BUCKET` | `sientia` | bucket holding the uploaded bundles |
| `IMPORT_BUNDLE_PREFIX` | `imported_models/` | key prefix the uploads must sit under |
| `IMPORT_BUNDLE_RETENTION_DAYS` | `7` | lifecycle expiry the importer applies to that prefix |
| `IMPORT_WORK_DIR` | `<tempdir>/sientia-import` | worker scratch directory |
| `IMPORT_MAX_OBJECT_BYTES` | 1 GiB | gate 1 |
| `IMPORT_MAX_ARCHIVE_ENTRIES` | `5000` | gate 4 |
| `IMPORT_MAX_UNCOMPRESSED_BYTES` | 4 GiB | gate 4 |
| `IMPORT_MAX_COMPRESSION_RATIO` | `200` | gate 4 |
| `IMPORT_MODELS_COLLECTION` | `models` | the MongoDB collection that *is* the model listing |
| `IMPORT_STATUS_DB_NAME` | `sientia-core-mlops-bff` | the BFF database holding the import log |
| `IMPORT_STATUS_DB_HOST` | `POSTGRES_HOST` | override: the log is on another server |
| `IMPORT_STATUS_DB_PORT` | `POSTGRES_PORT` | override |
| `IMPORT_STATUS_DB_USER` | `POSTGRES_USER` | override |
| `IMPORT_STATUS_DB_PASSWORD` | `POSTGRES_PASSWORD` | override |
The import log is a **second connection but the same server**: host, port, user and password come
from `POSTGRES_*`, and only the database name is import configuration, because the log lives in the
BFF's database rather than in the one Laborious holds for predictions. The import writes to that
database and no other, and to one table in it. The four `IMPORT_STATUS_DB_*` connection variables
are overrides for a deployment whose log genuinely sits elsewhere — the e2e suite, which runs the
two databases as two containers, is exactly that case.
A missing or malformed `IMPORT_PASSWORD_KEY` is an import failure at the decryption step, not a boot
failure: the worker starts either way.
## 5. The import log row — the BFF's half
The row is `public.experiment_run` in `sientia-core-mlops-bff`. **The workflow never inserts it.**
The inserting side writes:
| Column | Value |
|---|---|
| `run_type` | `'IMPORT'` |
| `status` | `'PENDING'` |
| `username` | whoever asked for the import |
| `bucket_name` | the bucket the object was uploaded to |
| `file_name` | the object key |
| `request_data` | the request payload. If it carries a digest under `expected_digest`, `digest`, `sha256` or `file_digest`, the workflow checks it against the input's |
Then it passes the returned `id` as `import_run_id`.
The workflow's first act is to **claim** that row:
```sql
UPDATE public.experiment_run
SET status = 'ORCHESTRATOR_WAITING_PROC', updated_at = :now,
orchestrator_response_data = :detail
WHERE id = :import_run_id
AND run_type = 'IMPORT'
AND (status = 'PENDING'
OR (status = 'ORCHESTRATOR_WAITING_PROC'
AND orchestrator_response_data -> 'import' ->> 'workflow_id' = :workflow_id))
```
The second disjunct makes a re-execution of the same workflow idempotent instead of a conflict. Every
later write keys on `id` alone, because ownership was settled here.
**Columns the workflow never writes**: `username`, `request_data`, `bucket_name`, `file_name`,
`run_type`, `created_at`. **Columns it does write**: `status`, `error_message`, `updated_at`,
`orchestrator_response_data`, and — as soon as the bundle is open, before anything exists in MLflow —
`experiment_name` (the bundle's own, verbatim) and `run_name` (`import-{model_name}-v{model_version}`,
truncated inside the name segment to fit 50 characters). Those two are **overwritten** by the
workflow; whatever the inserting side put there is replaced.
### When the row cannot be claimed
If the claim matches nothing — no such id, a `run_type='TRAIN'` row, or a row already running under
another workflow — the import **writes nothing anywhere**, because the row belongs to someone else.
Instead it:
- fails the Temporal run with error type `ImportLogRowNotClaimableError` and code
`IMPORT_LOG_ROW_NOT_CLAIMABLE`, non-retryable;
- increments the import error metric and fires a notification;
- logs the id and the observed state (which of the three causes it was).
So a row left `PENDING` with no `workflow_id` in its detail JSON means the workflow never owned it.
**Closing that row out belongs to whichever side inserted it** — by timeout, by a sweeper, by a person
— however that side chooses. The importer deliberately does not guess.
## 6. Status vocabulary
| Status | Written when | New in `V12`? |
|---|---|---|
| `ORCHESTRATOR_WAITING_PROC` | claimed, and while running | no — **reused** from the training flow |
| `ORCHESTRATOR_VALIDATION_ERROR` | failure at `received` … `content_policy` — the user can act on it | no — **reused** |
| `IMPORT_SUCCESS` | every step succeeded | yes |
| `IMPORT_ERROR` | failure at `experiment_creation` … `model_document`, or unclassified — only support can act on it | yes |
The two reused values describe a *stage*, not a flow: what says the row is an import is `run_type`,
not the status. There is deliberately no `IMPORT_RUNNING`; it would carry no information `run_type`
does not already carry, and it would widen both the pairing `CHECK` and the frontend's switch for
nothing.
`error_message` is the user-facing sentence, taken from the catalog in §8. On success it is written as
`NULL`, never `''` — the column carries a `LENGTH(error_message) >= 3` check.
## 7. The migration this depends on
`sientia-dataops-database-migrations`,
`postgres/sientia-core-mlops-bff/migrations/V12__add_import_statuses_to_experiment_run.sql`:
- adds `IMPORT_SUCCESS` and `IMPORT_ERROR` to `chk_experiment_run_status`;
- adds `chk_experiment_run_terminal_status_run_type`, pairing only the **terminal** values with their
`run_type` (`IMPORT_*` on `'IMPORT'`, `TRAINING_*` on `'TRAIN'`) and leaving the shared
`ORCHESTRATOR_*` values out of the rule;
- **adds no columns at all.**
`V12` must be merged and applied before `IMPORT_MODEL_ENABLED` is set anywhere. Without it, an import
provisions the model and then cannot record its verdict: the claim still succeeds (no new columns are
needed for it), the old `CHECK` refuses `IMPORT_SUCCESS`, and the failure surfaces as
`IMPORT_STATUS_NOT_RECORDED` on the first attempt — no retry wait — leaving the row in
`ORCHESTRATOR_WAITING_PROC` with a fully provisioned model behind it.
## 8. `orchestrator_response_data`, and the error catalog
There is no column for the error code and none for the workflow identifier (decision D21). Both live
at fixed keys in the JSON column the workflow already owns, and the whole `import` object is rewritten
on every write, so the identifier the claim wrote is never dropped:
```json
{
"import": {
"workflow_id": "import_model-41-01HW…",
"code": "IMPORT_BUNDLE_INCOMPLETE",
"step": "structure_validation",
"gate": 6,
"source": { "model_name": "sales_forecast", "experiment_name": "…", "…": "…" },
"cleanup_failed": false
}
}
```
- `workflow_id` — what a user quotes to support, and what the claim's idempotency disjunct reads.
- `code` — the frontend's i18n key. **Absent on success.**
- `step` — where the import stopped (the values of §9).
- `gate` — which of the seven gates rejected it, when one did.
- `source` — the bundle's own metadata block, once it is known.
- `cleanup_failed` — temporary files were left on the worker. Present on a **successful** import too:
it never changes the status, because the user's model was imported.
### Code → reason
`error_message` is exactly the sentence in this table; nothing is interpolated into it, and the
exception's text — paths, table names, exception classes, tracebacks — reaches only the logs.
| Code | Step | Sentence written to `error_message` |
|---|---|---|
| `IMPORT_REQUEST_INVALID` | `received` | The import could not be started because the request was incomplete or contradictory. Nothing was created. Submit the import again. |
| `IMPORT_FILE_UNREADABLE` | `download` | The uploaded file could not be read, or it is not the file that was sent. Upload it again. |
| `IMPORT_BUNDLE_CANNOT_BE_OPENED` | `decryption` | This file could not be opened: the password is wrong, or the file is damaged or was altered. Confirm the password with whoever exported the model, then try again. |
| `IMPORT_BUNDLE_REJECTED` | `archive_inspection` | The file's contents did not pass the platform's safety checks, so it was not opened. Export the model again from the origin platform. |
| `IMPORT_BUNDLE_UNPACK_FAILED` | `extraction` | The file could not be unpacked safely and nothing from it was kept. Export the model again from the origin platform. |
| `IMPORT_BUNDLE_INCOMPLETE` | `structure_validation` | This file is not a complete model export — part of what the platform needs is missing from it. Export the model again from the origin platform. |
| `IMPORT_BUNDLE_UNEXPECTED_CONTENT` | `content_policy` | The file contains something a model export should not contain, so it was rejected. Export the model again from the origin platform. |
| `IMPORT_STORAGE_PREPARATION_FAILED` | `experiment_creation` | The platform could not prepare a place to keep this model. The import stopped and no model was created. Contact support with this import's identifier. |
| `IMPORT_MODEL_FILES_NOT_STORED` | `artifact_upload` | The model's files could not be stored on this platform. The import stopped and the model is not available. Contact support with this import's identifier. |
| `IMPORT_MODEL_NOT_REGISTERED` | `registration` | The model's files were stored, but the model itself could not be registered, so it cannot be used. Contact support with this import's identifier. |
| `IMPORT_MODEL_SETTINGS_NOT_SAVED` | `model_document` | The model was registered, but the settings that tell the platform how to run it could not be saved. Contact support with this import's identifier. |
| `IMPORT_UNEXPECTED_ERROR` | — | The import stopped for an unexpected reason and no model was created. Contact support with this import's identifier. |
Three codes never reach a row's `error_message`, because there is no row to write or no failure to
report to a user. They exist in the logs, the metric and the notification only:
| Code | When |
|---|---|
| `IMPORT_LOG_ROW_NOT_CLAIMABLE` | the row is not this workflow's (§5) |
| `IMPORT_STATUS_NOT_RECORDED` | the verdict itself could not be written; the log line carries the verdict it could not record |
| `IMPORT_CLEANUP_INCOMPLETE` | temporary files were left behind; also flagged as `cleanup_failed` in the detail |
## 9. Steps
`received`, `download`, `decryption`, `archive_inspection`, `extraction`, `structure_validation`,
`content_policy`, `experiment_creation`, `artifact_upload`, `registration`, `model_document`,
`cleanup`. The first seven are the gates (see the format document); a failure in any of them is a
`ORCHESTRATOR_VALIDATION_ERROR`. The next four are provisioning; a failure there is an
`IMPORT_ERROR`. `cleanup` is never a reported failure.
## 10. The import knows nothing about projects
The model listing **is** the MongoDB `models` document. No migrated relational database in this
platform holds a project table or a model table, nothing in this repository reads a project-to-model
link, and the workflow input carries no project target — which is why naming one is refused rather
than accepted and ignored. Whatever the import screen does with projects is the frontend's business.
The bundle's own `metadata.model_project` is logged as an MLflow parameter (`source.model_project`),
as a record of where the model came from. It is not a link.
## 11. What MLflow ends up holding
| Thing | Value |
|---|---|
| Experiment | `metadata.experiment_name` from the bundle, created if absent, reused if present |
| Run | always **new**, created with the `run_name` already written on the record |
| Artifacts | the bundle's whole `artifacts/` tree, with `prediction_model` and `data_model` at the run's artifact root |
| Parameters | the bundle's `parameters`, verbatim, plus the origin metadata as `source.*` (including `source.run_id`, which is never dereferenced) |
| Registered version | created on demand; source is the run's **resolved artifact URI** with `/prediction_model` appended, e.g. `mlflow-artifacts:/17/<run_id>/artifacts/prediction_model` |
| Stage | **untouched** |
The source shape matters and is easy to get wrong: the legacy loader extracts the run id as
`source.split('/')[2]`, which is the run id in the resolved artifact URI but the *artifact path* in a
literal `runs:/<run_id>/prediction_model`. Registering with the unresolved URI breaks loading
silently — nothing raises, the loader just returns the wrong string. This is the same value
`mlflow.register_model` stores for the training flow, because it resolves the `runs:/` URI before
writing it; `MlflowClient.create_model_version` does not, so the import resolves it itself.
## 12. The model document
Written **last**, into the collection named by `IMPORT_MODELS_COLLECTION` (default `models`):
```json
{
"id": "31",
"name": "sales_forecast",
"active": false,
"model_config": {
"transform_flavor": "sklearn",
"predict_flavor": "joblib",
"retention_minutes": 60,
"target": "SE"
}
}
```
- `id` is `str(max(int(id) for id in collection) + 1)` — computed in Python, from the ids that are
numbers. Never a `count`, never a lexicographic sort (which would answer `"9" > "30"`). Ids that
are not decimal numbers are ignored for the maximum and counted in the log.
- `name` is `metadata.model_name` from the bundle, verbatim.
- `active` is `false`, always. No code path in the import writes or updates a document with
`active: true`.
- `target` comes from `parameters.target_variable`; nothing else supplies it.
- The `models` collection carries a **unique sparse index on `id`**, created by the MongoDB baseline
migration. A duplicate-key insert is reported as a `model_document` failure — no retry, no id
recomputation.
**A name already in the collection is left exactly as it is.** The import writes nothing, modifies
nothing, logs the skip and succeeds: `active`, `target` and the flavors of a model already on the
platform are operator-owned.
## 13. After a successful import — the manual steps
A `COMPLETED` import is **not** a servable model. Stated plainly:
- the registered version is **unstaged**, and `load_predict_model` only sees a version once it is in
stage `Production`;
- the model document is **inactive** (`active: false`).
So an operator must promote the version to `Production` and set `active: true`. This workflow never
promotes and never activates — that runbook is QTZPOC-21's.
One more thing to know before serving an imported model: `predict_flavor: 'joblib'` is what the
artifact actually is, and it loads, predicts **and** retrains — `log_model` writes that flavor too,
as `model.pkl` + `code/`. The artifact it writes carries no `MLmodel` manifest, so an imported model
stays readable only by this runtime, before and after a retrain. Full contract:
[`model-flavors.md`](model-flavors.md).
## 14. Retry, and what does not retry
Only the status writes retry, and only because `UPDATE ... SET <fixed values> WHERE id` leaves the row
the same however many times it runs:
| Call | Attempts | Why |
|---|---|---|
| claim, terminal write | 10, 1 s → 30 s backoff (≈151 s envelope) | the verdict must land; the database being briefly unavailable is exactly what a retry fixes |
| progress writes, names | 3, capped at 2 s | a hint nobody reads — the terminal write rewrites it — must not stall a working import |
| everything else | 1 | a repeated registry call or Mongo insert can write twice |
`ImportLogRowNotClaimableError` and `ImportStatusRejectedError` are non-retryable: the first would
read the same rows and reach the same conclusion, and the second means the statement itself was
refused (a `CHECK` violation, an undefined column, a value too long), which a second attempt cannot
change.
There is no rollback, no compensation, no resume and no reconciliation. A failed import stays failed;
trying again is a new upload and a new row.
## 15. How this contract is verified
Two suites, and they prove different things.
**Unit** — `pytest tests/`, part of the standard gate (`validate --project-name=laborious`). Covers
the bundle reader against committed golden bytes, every gate, the failure catalog, the status SQL and
the workflow's ordering, all against fakes.
**End to end** — `pytest e2e -m import_e2e`, opt-in and outside `testpaths`. Runs the real workflow on
a real worker against real MinIO, a real Postgres holding this document's `public.experiment_run`
(with and without `V12`), a real MongoDB with the unique index on `models.id`, and a real
`mlflow server`. Nothing on the import path is mocked. This is what proves the claims a fake cannot:
that the `CHECK` constraints of § 6 accept every status written, that the registry source of § 11
resolves through the existing loader, that the `V12` failure shape of § 7 is what it says, and that no
plaintext password reaches the Temporal event history.
Requires Docker plus one one-time setup per machine:
```bash
echo "ryuk.disabled=true" >> ~/.testcontainers.properties
```
Without it every e2e test fails in fixture setup with a Docker mount error — see
[`README.md` § End-to-End Tests](../README.md#end-to-end-tests-e2e) for why and what it costs.
Scenario catalogue: [`e2e/scenarios.md`](../e2e/scenarios.md) § 4.
Requirement-to-test mapping: `openspec/changes/import-model-e2e-suite/traceability.md`.
Doing one import by hand, standing in for the frontend that does not exist yet:
[`scripts/import_model_manual_run.py`](../scripts/import_model_manual_run.py) — see
[`README.md` § Running One Model Import by Hand](../README.md#running-one-model-import-by-hand).

170
docs/opc-communication.md Normal file
View File

@@ -0,0 +1,170 @@
# OPC UA communication (Laborious)
Laborious exports predictions to OPC UA servers through `OpcRepository` ([`laborious/utils/repository/opc_repository.py`](../laborious/utils/repository/opc_repository.py)) and the Temporal activity layer in [`laborious/activities/opc.py`](../laborious/activities/opc.py).
Implementation plan for session/channel recovery on Tier-1 `Bad*` errors: [`.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md`](../.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md).
## Architecture
```text
Worker (long-lived)
└── OpcRepository per OPC server id (from OPC_CONFIG / env)
├── connect / disconnect / validate_connection (read-only)
├── _connect_locked / _reconnect_locked (under _connection_lock)
├── write_data (single attempt per call)
└── background reconnect on Tier-1 Bad*, protocol closed, or stale session
Temporal activity write_opc_data
└── OPC.manage_output_tags → write_data per tag (sequential per activity)
```
One worker process holds one `OpcRepository` instance per configured server. Multiple Temporal activities can call `write_data` concurrently on the same repository.
## Connection lifecycle
| Phase | Behavior |
|-------|----------|
| Startup | `init_opc()` creates repositories and calls `connect()` → `_connect_locked()` |
| Steady state | `validate_connection()` is read-only (`protocol.state` only); `_session_ready` is checked in `write_data` |
| Tier-1 Bad* / protocol closed | `_start_reconnect` → `_run_reconnect` → `_reconnect_locked()` (respects `reconnection_interval`) |
| Write | `write_data()` checks reconnect task, `_session_ready`, validates protocol, then one `get_node` + `write_value` |
| Shutdown | `close()` disconnects all repositories |
### Session and channel timeouts
Requested session and secure-channel lifetime: **10 minutes** (`OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS` in `opc_repository.py`). The server may revise these values; negotiated values are logged after connect and exposed as `opc_session_revised_timeout_milliseconds`.
### Reconnection interval
`OPC_RECONNECTION_INTERVAL` is in **seconds** (default `120`). It gates **background** reconnect after Tier-1 `Bad*`, closed protocol, or stale session (`last_reconnection_time` is updated only in `_reconnect_locked()`). It limits load on the OPC server when many workflows fail at once.
## Concurrency: connection lock and session readiness
To allow **multiple concurrent writes** when the session is healthy, but **block all writes** while the connection is being torn down or re-established:
| Primitive | Role |
|-----------|------|
| `_connection_lock` (`asyncio.Lock`) | Held for the entire `disconnect` → `connect` path. Only one connection-maintenance task at a time. |
| `_session_ready` (`asyncio.Event`) | Set when a session is ready for writes; cleared before reconnect starts and set again after a successful connect. |
**Connection methods (caller holds `_connection_lock` for `_*_locked` helpers):**
| Method | Role |
|--------|------|
| `_create_client()` | Create asyncua `Client` + optional `set_security`; raises if `client` already exists |
| `_open_session()` | `client.connect()` + metrics; raises if session already open or client missing |
| `_connect_locked()` | `_create_client()` (when needed) + `_open_session()`; raises if already connected |
| `_disconnect_locked()` | Teardown session and clear `client` |
| `_reconnect_locked()` | `_disconnect_locked()` + `_connect_locked()`; sets `last_reconnection_time` |
Public `connect()` / `disconnect()` acquire the lock and call `_connect_locked()` / `_disconnect_locked()`.
**Write path (`write_data`):**
1. If a reconnect task is **in flight** → **fail immediately** (`opc_error_kind=reconnect_in_progress`).
2. If `_session_ready` is cleared and no task is running → schedule reconnect (`SessionNotReady`); fail with `connection_lost` or `reconnect_in_progress` if a task started.
3. `validate_connection()` checks `protocol.state` only (read-only). If closed → schedule reconnect (`ProtocolClosed`) and fail with `opc_error_kind=connection_lost`.
4. Single `get_node` + `write_value` (no retry). Tier-1 `Bad*` on write also schedules reconnect.
**Reconnect path (`_run_reconnect`):**
1. `_start_reconnect` clears `_session_ready` and schedules the task when the interval allows and `_allow_reconnect` is true.
2. `async with _connection_lock:` → `_reconnect_locked()`.
3. `_session_ready` is set on successful `_open_session()`.
4. `disconnect()` sets `_allow_reconnect=False` so shutdown does not respawn sessions.
A second `_connect_locked()` while a session is already open raises `OpcSessionAlreadyConnectedError` (disconnect first).
**asyncua note:** Concurrent `write_value` on the same session is only safe if the stack tolerates it. If production shows issues, serialize writes with an optional `asyncio.Semaphore(1)` while keeping the connection lock semantics above.
**Future threads:** replace `asyncio.Lock` / `Event` with `threading` primitives or route all OPC I/O through one dedicated loop.
## Reconnect triggers
Background reconnect is scheduled when:
- `validate_connection()` sees a closed or missing protocol (`ProtocolClosed`).
- `_session_ready` is clear after a failed reconnect (`SessionNotReady`).
- A write raises a Tier-1 `UaStatusCodeError` in `RECONNECTABLE_OPC_BAD_NAMES`.
For Tier-1 `Bad*` when the server invalidates the session (e.g. `BadSessionIdInvalid`) but the client still sees transport as open, `write_data` fails once, records the OPC status in metrics, and **schedules** reconnect if:
- The exception is a `UaStatusCodeError` whose name is in `RECONNECTABLE_OPC_BAD_NAMES` (see plan), and
- `reconnection_interval` has elapsed since `last_reconnection_time`, and
- No reconnect task is already running.
There is **no write retry**: the failed export is not sent again in the same activity.
## Prediction confidence and PostgreSQL comments
| `prediction_confidence` | Meaning |
|-------------------------|---------|
| (unchanged) | Successful OPC export |
| **12** | Generic OPC write failure (`OPC_WRITTING_ERROR_CONFIDENCE`) |
| **14** | Tier-1 session/channel `Bad*` on export (`OPC_SESSION_BAD_CONFIDENCE`) |
| **14** | Write while reconnect in progress (`OPC_SESSION_BAD_CONFIDENCE`, comment `OPC UA reconnect in progress`) |
| **13** | PI Web API write failure (separate path) |
Session/channel errors use a stable comment for counting:
```text
OPC UA session/channel error: BadSessionIdInvalid
```
Reconnect-in-progress exports use:
```text
OPC UA reconnect in progress
```
Example SQL:
```sql
SELECT count(*) FROM predictions WHERE prediction_confidence = 14;
SELECT count(*) FROM predictions WHERE comments LIKE 'OPC UA session/channel error:%';
```
## Prometheus metrics (`opc_*`)
Defined in [`laborious/metrics.py`](../laborious/metrics.py). Do not rename in production without a dashboard migration.
| Metric | Purpose |
|--------|---------|
| `opc_connections_initiated_total` | Connection attempts |
| `opc_connections_failed_total` | Failed connects |
| `opc_connection_status` | Gauge 1=connected, 0=disconnected |
| `opc_session_created_total` | Session established after connect |
| `opc_session_closed_total` | Disconnect initiated |
| `opc_session_revised_timeout_milliseconds` | Negotiated session timeout (ms) |
| `opc_write_attempts_total` | Per write; label `result` = `OK` or exception name |
| `opc_write_inter_arrival_over_session_timeout_total` | Successful writes spaced longer than revised session timeout |
Legacy activity metrics: `laborious_prediction_opc_writing_count`, `laborious_prediction_opc_writing_response_time_monitor`.
## Environment variables
| Variable | Default | Description |
|----------|---------|-------------|
| `OPC_CONFIG` | — | JSON map of server configs (overrides single-server env) |
| `OPC_ID` | `1` | Server id |
| `OPC_URL` | `opc.tcp://localhost:4840` | Endpoint |
| `OPC_SERVER_NAME` | `default_server` | Label for metrics/logs |
| `OPC_SERVER_URI` | same as URL | Application URI / cert SAN |
| `OPC_CERT_PATH` | — | Client certificate (secure mode) |
| `OPC_PRIVATE_KEY_PATH` | — | Client private key |
| `OPC_SERVER_CERT_PATH` | — | Server certificate |
| `OPC_RECONNECTION_INTERVAL` | `120` | Minimum seconds between reconnects |
## Operations checklist
- Correlate `BadSessionIdInvalid` in `opc_write_attempts_total` with `opc_session_closed_total` / `opc_session_created_total` (reconnect may finish after the row is stored with confidence 14).
- Use confidence **14** and comment prefix for session invalidation rates; use **12** for other OPC failures.
- Respect `OPC_RECONNECTION_INTERVAL` under parallel load; bursts of confidence 14 are expected until the next successful cycle.
## Related tests
- Unit: [`tests/laborious/utils/repository/test_opc_repository.py`](../tests/laborious/utils/repository/test_opc_repository.py)
- Unit: [`tests/laborious/activities/test_opc.py`](../tests/laborious/activities/test_opc.py)
- E2E (mock OPC): [`e2e/test_predictions_batch_format_export.py`](../e2e/test_predictions_batch_format_export.py)
- E2E (in-process asyncua server + real `OpcRepository`): [`e2e/test_opc_real_server.py`](../e2e/test_opc_real_server.py) — scenarios 3.1.2, 3.2.2, 3.2.4, 3.2.5
- Scenarios: [`e2e/scenarios.md`](../e2e/scenarios.md)

196
docs/sientia-bundle-format.md Normal file
View File

@@ -0,0 +1,196 @@
# The `.sientia` bundle format
A `.sientia` file is one exported MLflow model: its whole artifact tree, its run parameters and six
fields of origin metadata, zipped and then encrypted under a password the exporter chose. It is
written by the Streamlit platform (`sientia-projects-templates`) and read by this repository's
importer. Two programs in two repositories agree on these bytes, so the constants below are
duplicated by necessity — the drift is held in place by tests on both sides, not by a shared library.
This document covers both directions: **§1–§3 how a bundle is generated**, so the producer can be
audited against it, and **§4–§7 how it is read**, so a rejection can be explained.
Normative implementations:
| Side | Where |
|---|---|
| Writer | `sientia-projects-templates`, `app/src/operations/model_export/{crypto,bundle,exporter}.py` |
| Reader | this repository, `laborious/utils/bundle/{format,reader}.py` |
| Test writer | this repository, `tests/helpers/bundle_factory.py` — mirrors the producer field for field, for building hostile variants |
| Golden fixture | `tests/fixtures/bundle/` — a real producer artefact, with its password and producer commit recorded in that directory's `README.md` |
---
## 1. Generation, step by step
Entry point: `exporter.export_model_bundle(*, client, run_id, model_name, experiment_name,
model_version, model_project, password) -> bytes`.
1. **Password strength** — `crypto.validate_password_strength`: at least 12 characters, one
uppercase, one lowercase, one digit, one symbol. **Export-side only.** The importer cannot
re-check it and does not try: by the time a bundle exists, the password is whatever it was.
2. **Exportability** — `bundle.validate_run_is_exportable`: the run's top-level artifacts must
include **both** `data_model` and `prediction_model`, checked with `client.list_artifacts(run_id)`
before anything is downloaded.
3. **Whole-tree download** — `client.download_artifacts(run_id, '', staging)` into a
`tempfile.mkdtemp(prefix='sientia_export_')` directory. The entire artifact tree, not a subset.
4. **Run parameters** — `dict(run.data.params)`, copied **verbatim**. Whatever the run logged is
what the bundle carries, including `target_variable` (see §3).
5. **Metadata document** — `bundle.build_metadata_json`, exactly two top-level keys:
- `parameters`: the run parameters, verbatim;
- `metadata`: `model_name`, `experiment_name`, `run_id`, `model_version`, `model_project` and
`export_timestamp = datetime.now(UTC).isoformat()`. Six fields, no others.
6. **Zip** — `bundle.build_zip`, `zipfile.ZIP_DEFLATED`. `metadata.json` is written **first**, with
`json.dumps(...)` and no indent. Every file under the staging directory follows as
`artifacts/<relpath>`, with `\` replaced by `/`. **No directory entries are stored**, so a reader
must derive directories from file paths.
7. **Encryption** — `crypto.encrypt_archive`: a fresh 16-byte `os.urandom` salt, Argon2id key
derivation, then a libsodium `secretstream` push in 64 KiB plaintext chunks, the last chunk tagged
`FINAL` and the rest `MESSAGE`.
8. **Download name** — `{model_name}_v{model_version}.sientia`. The staging directory is removed in a
`finally`, and the password is popped from the Streamlit session state.
Producer pins: `pynacl==1.5.0`, `mlflow==2.10.1`.
## 2. Constants
Copied from `crypto.py` unless noted. The reader's names are the ones in
`laborious/utils/bundle/format.py`.
| Producer | Value | Reader |
|---|---|---|
| `MAGIC` | `b'SIENTIA1'` | `MAGIC` |
| `_HEADER_STRUCT` | `struct.Struct('>8sBBB16sQQB32s64s')` | `HEADER_STRUCT` |
| `HEADER_STRUCT_SIZE` | `140` | `HEADER_SIZE` |
| `FORMAT_VERSION` | `1` | `FORMAT_VERSION` |
| `AEAD_ID_XCHACHA20POLY1305_SECRETSTREAM` | `1` | same name |
| `KDF_ID_ARGON2ID` | `1` | same name |
| `SALT_SIZE` | `16` | `SALT_SIZE` |
| `KDF_MEMLIMIT_BYTES` | `268435456` (256 MiB) | `MAX_KDF_MEMLIMIT` — a **ceiling** on the reader's side |
| `KDF_OPSLIMIT` | `3` | `MAX_KDF_OPSLIMIT` — likewise a ceiling |
| `KDF_PARALLELISM_RESERVED` | `1` | `PARALLELISM` |
| `DIGEST_SIZE` | `32` | `DIGEST_SIZE` |
| `RESERVED_SIGNATURE_SIZE` | `64` | `RESERVED_SIGNATURE_SIZE` |
| `CHUNK_SIZE` | `65536` (64 KiB) | `CHUNK_SIZE` |
| KDF algorithm | `crypto_pwhash_ALG_ARGON2ID13` | `KDF_ALG` |
| Key length | `crypto_secretstream_xchacha20poly1305_KEYBYTES` (32) | `KEY_SIZE` |
Fixed by libsodium rather than by the producer, but needed to frame the stream:
- `crypto_secretstream_xchacha20poly1305_HEADERBYTES == 24` → `STREAM_HEADER_SIZE`
- `crypto_secretstream_xchacha20poly1305_ABYTES == 17` → `ABYTES`
- pre-upload prefix `140 + 24 + (65536 + 17) = 65717` → `PREFIX_SIZE`, the smallest number of leading
bytes from which a header and one full chunk can be checked
**Password normalisation.** `_derive_key` applies `unicodedata.normalize('NFC', password)` and
encodes UTF-8 — no trimming, no case folding, no padding. The reader normalises identically, so a
password typed in decomposed form opens a bundle written from the composed form.
## 3. `target_variable`, and where it does not come from
Gate 6 requires `parameters.target_variable`: it is the only source of `model_config.target` in the
model document, and every workflow that serves a model indexes that key directly. On the producer
side:
| Export path | Logs `target_variable`? | Evidence |
|---|---|---|
| Time-series training template | yes | `app/pages/template_01.py` L1212-1214 |
| Model manager, "Save Experiment" | yes, if the session carried it | `app/pages/model_manager.py` L791-796 re-logs `st.session_state.parameters` verbatim |
| Model ensemble | same condition | `app/pages/model_ensemble.py` L695-700 |
| Pipeline / timeseries template | **no** | `app/pages/timeseries_template.py` L1761-1762, L1811-1812 log `run_metadata` only |
| AutoML | **no** | `automl_logging_operations.py` L454-491; the target appears as `"target"` inside the model card JSON, never as a run parameter |
So a gate-6 rejection naming `target_variable` means the origin run came from the pipeline or AutoML
path. The remedy is to re-export from a run trained by the time-series template, which is what the
importer's `IMPORT_BUNDLE_INCOMPLETE` sentence already asks for.
---
## 4. The file on disk
```
offset width field constraint the reader enforces
------ ----- -------------------- ------------------------------------------------------
0 8 magic == b'SIENTIA1'
8 1 format_version == 1
9 1 aead_id == 1 (XChaCha20-Poly1305 secretstream)
10 1 kdf_id == 1 (Argon2id)
11 16 salt passed to the KDF as-is
27 8 kdf_memlimit_bytes <= 268435456
35 8 kdf_opslimit <= 3
43 1 kdf_parallelism == 1
44 32 ciphertext_digest == sha256(file[164:])
76 64 reserved_signature all zero
------ -----
140 24 secretstream header libsodium HEADERBYTES
164 ... chunk stream 64 KiB plaintext chunks, each +17 bytes of AEAD tag
```
All integers are big-endian; the format string starts with `>`, so there is no alignment padding and
the offsets above are exact.
**The digest boundary.** The producer computes `hashlib.sha256(ciphertext)` where `ciphertext` is the
concatenation of the *pushed chunks only*, and then writes `header + stream_header + ciphertext`. On
disk, therefore, the digest covers everything after offset **164**, not after 140:
> `ciphertext_digest == sha256(file[HEADER_SIZE + STREAM_HEADER_SIZE:])`
A reader that hashes from 140 will reject every valid bundle. The fixture test that recomputes both
values and asserts the first matches while the second does not is what pins this, and is the reason
the golden fixture exists.
**Version policy.** `format_version` is checked for equality, not for a range: a version this reader
does not know is refused at gate 2 rather than parsed optimistically. `reserved_signature` is
reserved for a future signature and must be all zero today, so a signed bundle is a new format
version and not a silent change of meaning.
## 5. The seven gates
`BundleReader.open()` runs them in this order, in one function, so the order cannot be changed by a
caller. Each raises `BundleGateError(gate=n)`, which the activity layer maps to an import step.
| Gate | Step recorded | What it checks | Written to disk |
|---|---|---|---|
| 1 | `download` | object size ceiling, key prefix, `.sientia` suffix — from `stat_object`, before the body is fetched; then the SHA-256 of the downloaded bytes against the digest the uploader computed | the downloaded object only |
| 2 | `download` | the whole header table of §4, including the chunk-stream digest | nothing |
| 3 | `decryption` | Argon2id derivation with the header's parameters, then `secretstream` pull. The AEAD is the integrity check: a wrong password and an altered file fail identically, and the message says so | the decrypted zip, in the work directory |
| 4 | `archive_inspection` | central directory only: entry count, total uncompressed size, compression ratio, absolute names, `..` components, symlink and non-regular modes | nothing |
| 5 | `extraction` | extraction into a **fresh** directory (a pre-existing one is a failure), every member's `realpath` contained in it, no overwrite of a pre-existing file; a rejection removes its own partial tree | the extracted tree |
| 6 | `structure_validation` | `metadata.json` + `artifacts/` layout, exactly two top-level keys, all six metadata fields present and non-blank, `experiment_name` between 3 and 50 characters, `parameters.target_variable` present and non-blank, and `prediction_model`/`data_model` each holding `MLmodel` and `model.pkl` | nothing |
| 7 | `content_policy` | every file is either a known name (`MLmodel`, `metadata.json`, `model.pkl`, `conda.yaml`, `python_env.yaml`, `requirements.txt`, `model_card.json`, `model_card.svg`) or a known suffix (`.csv .json .yaml .yml .txt .pkl .svg .md`) | nothing |
Gates 1 and 2 share the `download` step deliberately: an oversized object, a wrong magic, an
unsupported version and a digest that disagrees all say *these are not the bytes that were sent*, and
none of them may borrow the `decryption` sentence, which implicates the password.
The `experiment_name` bound in gate 6 is not cosmetic: the import record's `experiment_name` column is
`VARCHAR(50)`, and the name is written onto the record before provisioning starts. Enforcing the
column's limit at the gate is what keeps it from surfacing three activities later as a constraint
violation.
## 6. Ceilings and where they come from
Gate 1 and gate 4 read their limits from `build_import_config()`, so a deployment can tighten them:
| Variable | Default | Gate |
|---|---|---|
| `IMPORT_MAX_OBJECT_BYTES` | 1 GiB | 1 |
| `IMPORT_MAX_ARCHIVE_ENTRIES` | 5000 | 4 |
| `IMPORT_MAX_UNCOMPRESSED_BYTES` | 4 GiB | 4 |
| `IMPORT_MAX_COMPRESSION_RATIO` | 200 | 4 |
| `IMPORT_BUNDLE_PREFIX` | `imported_models/` | 1 |
The KDF ceilings are **not** configurable: they are the producer's own cost parameters, and a header
declaring more is a resource-exhaustion attempt, refused before any memory is allocated.
## 7. Fixtures
- `tests/fixtures/bundle/golden_model_v3.sientia` — emitted by the producer's own code in its venv,
with the password and producer commit recorded in `tests/fixtures/bundle/README.md`. Regenerate it
with `tests/fixtures/bundle/regenerate.py`. It pays the producer's real Argon2id cost (256 MiB,
3 passes), so it is used where the real thing matters and not in every test.
- `tests/helpers/bundle_factory.py` — the test-only writer. Same header fields at the same offsets,
but 8 MiB and 1 pass, and with a knob for every hostile variant: wrong magic, unknown version, KDF
parameters above the ceiling, digest mismatch, tampered ciphertext, truncated final chunk, non-zero
reserved signature, zip bomb, `..` entry, absolute entry, symlink, missing `data_model`, missing
`MLmodel`, missing `model.pkl`, malformed metadata, missing `target_variable`, extra `.sh` file. A
test asserts its header bytes are structurally identical to the golden fixture's.