From 68600e7b3111a94f18e264f68f07a190cf95ed75 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 1 Sep 2026 13:28:13 -0300 Subject: [PATCH] Add docs --- docs/model-flavors.md | 156 +++++++++++++ docs/model-import.md | 417 ++++++++++++++++++++++++++++++++++ docs/opc-communication.md | 170 ++++++++++++++ docs/sientia-bundle-format.md | 100 ++++++++ 4 files changed, 843 insertions(+) create mode 100644 docs/model-flavors.md create mode 100644 docs/model-import.md create mode 100644 docs/opc-communication.md create mode 100644 docs/sientia-bundle-format.md diff --git a/docs/model-flavors.md b/docs/model-flavors.md new file mode 100644 index 0000000..9efd64c --- /dev/null +++ b/docs/model-flavors.md @@ -0,0 +1,156 @@ +# 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()` — the loader fetches what it needs | +| `pyfunc` | yes | yes | `mlflow.pyfunc.load_model()`; the wrapper path downloads the artifact instead and unwraps `._model_impl.python_model` | +| `joblib` | yes | yes | downloads the same artifact and `joblib.load`s its `/model.pkl` — **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 three values govern reading and writing. + +`pytorch` was accepted until this change and is now rejected like any other unknown value: `torch` is +not installed in the runtime, so `mlflow.pytorch.load_model` could never have loaded anything there. + +**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. + +## Same artifact, three readers + +The flavor changes the reader, never the path. All three resolve the same two artifact directories of +the Production run: + +``` +/artifacts/ +├── data_model/ ← transform_flavor loads this one +│ ├── MLmodel ← parsed by sklearn/pyfunc; ignored by joblib +│ ├── model.pkl ← what joblib.load reads +│ └── code/ ← put on sys.path by every flavor, when present +├── prediction_model/ ← predict_flavor loads this one +└── model_card/, *.csv +``` + +The run's `artifacts/` root holds **no** `.pkl` — every model pickle lives one level down, inside its +artifact directory. Predict resolves that directory through the registry URI +(`models://production`, which MLflow points at the same `prediction_model`); transform builds +the run artifact URI directly with `get_model_uri`, since the run is already resolved. Observability +is the same for the three: `MODEL_READ_*` on load, `MODEL_WRITE_*` on `log_model`, under the base +`operation_type` — no per-flavor metric, operation type or label. + +## `code/` on `sys.path`, for every flavor + +An artifact may ship the modules its own classes live in, under `/code/`. Without that +directory on `sys.path`, unpickling raises `ModuleNotFoundError` — the artifact is intact, the class +is simply not importable. + +MLflow's loaders do add it, but **from the manifest**: `_add_code_from_conf_to_system_path` acts only +when the `MLmodel` flavor config declares a `code` entry. An artifact carrying `code/` on disk +without that entry — code logged with `log_artifacts`, a re-log that dropped the config, anything a +`joblib` retrain wrote — loads with no module path at all. So this runtime reads the **directory**, +which no manifest can misdescribe: + +- `prepend_artifact_code_dir` inserts `/code` at the front of `sys.path`, **only when that + directory exists**, and only once (an entry already present is not duplicated). +- `download_artifact_code_dir` is what feeds it on the download paths, and it is deliberately cheap: + the remote listing (`list_artifacts(run_id, '')`) decides first, so a run without `code` + downloads **nothing**, and a run with it downloads only `/code` — a handful of `.py` + files, never the model's weights. +- `download_model` runs that step first, for every flavor and both of its branches, *before* any + model is loaded — the insert has to be in place before the loader unpickles. +- The models themselves are not pre-downloaded: `mlflow.sklearn.load_model` and + `mlflow.pyfunc.load_model` are handed the model URI and fetch what they need. `joblib` downloads + its artifact because it must read the pickle, and adds the entry again inside `_load_model_pickle` + through the same helper, so the flavor works when called directly too. +- When the manifest *does* declare `code`, MLflow inserts the same directory a second time. Harmless. + +The entry is never removed: a loaded model may import lazily, long after the load returned. + +The directory `download_artifact_code_dir` downloaded into is also what `download_model` returns +next to the model — `tuple[Any, str | None]`, the second element being the artifact directory the +retrain path re-logs `code/` from, or `None` when the artifact has no `code/`. It falls back to +whatever the loader downloaded on its own, which is a real artifact directory for `joblib` and for +the wrapper branch. `load_predict_model` and `load_transform_model` stay the **single** place where +the flavor is decided: `download_model` has no `joblib` branch, it delegates by `model_type`. Its +only remaining branch is `load_wrapper`, which is derived from `flavor == 'pyfunc'` but is otherwise +orthogonal to the flavor. + +> `sys.path` is process-global: two models embedding a package with the same name (`utils/`) on the +> same worker resolve to whichever entered first. Mitigation, if it bites: one worker per model. + +## `sklearn` + +Default. A standard MLflow sklearn artifact: `mlflow.sklearn.log_model` writes `MLmodel` + +`model.pkl`, `mlflow.sklearn.load_model` reads it, and every outside consumer (MLflow UI, MLflow +serving) recognizes it as a model. + +## `pyfunc` + +For models logged as `mlflow.pyfunc`, including the wrapper models this pipeline produces. Retrain +writes through the wrapper's own `store_model(artifact_path=, code_path=…)`. Two read +paths, and the second is **orthogonal to the flavor**: + +- Plain: `mlflow.pyfunc.load_model()`, called as `predict(context, model_input)`. +- Wrapper (`download_model(load_wrapper=True)`): downloads to `./tmp/artifacts//` + via `dowload_artifacts`, loads as pyfunc, then unwraps `._model_impl.python_model`. Inside such an + artifact the real pickle is **nested** — `/artifacts/stacking_model.pkl` for predict, + `/artifacts/training_transformer.pkl` for transform (`PREDICTION_COMPRESSED_PATH` / + `TRANSFORMED_COMPRESSED_PATH`). That nesting is why the `joblib` listing does not recurse. + +## `joblib` + +For models whose `.pkl` is in the compressed form `joblib.dump` writes, which MLflow's `pickle.load` +does not read — the artifact is intact, but no flavor loader opens it. + +- **Read**: `client.download_artifacts(run_id, )`, then + `joblib.load(/model.pkl)`. `model.pkl` is `MODEL_PICKLE_NAME`, the same constant + `log_model` writes; another `.pkl` directly inside the artifact is a fallback, in name order. +- **Write**: `joblib.dump(model, '/model.pkl', compress=3)`, plus a re-log of the source + artifact's `code/` to `/code`. +- **`code/` goes on `sys.path`** before the load, when present — see the section above; the joblib + flavor is where that started, and it is now shared by all three. +- **No `MLmodel` is written**, so only this runtime reads the artifact back — + `mlflow.sklearn.load_model` and `models://production` both fail on it. A model retrained in + joblib stays joblib. +- **The pickled object must expose `predict(data)`**, plus `fit(data)` if retrained. A pyfunc + `PythonModel` wrapper expecting `predict(context, model_input)` raises `TypeError`. + +## Switching a model to `joblib` + +The runtime does not migrate documents; editing `models` is an operator step. + +1. **Inventory before the deploy** — run predict/transform and note which models raise. There is no + discovery afterwards: the failure names the loader, not the fix. +2. **Edit the document** — `predict_flavor` and/or `transform_flavor` to `'joblib'`. Imported models + arrive ready: `import_model` writes it itself, see [`model-import.md`](model-import.md) § 12. +3. **Check the first retrain** — the new run must carry `prediction_model/model.pkl` and, when the + source had one, `prediction_model/code/`. +4. **Rollback is not just a revert** — artifacts a joblib retrain already wrote stop being loadable, + so it also means promoting the previous version back and reverting the documents' flavors. + +## Known gap: how the run id is resolved + +`get_model_run_id` takes `source.split('/')[2]` of the registered version's `source`. **Not confirmed +against the real tracking server**: no scheme reproduced locally puts the run id there (a file store +yields `''`, S3 the bucket, `mlflow-artifacts:/` the experiment id), yet predict works in production. +Every flavor that resolves a run goes through this line — confirm it with a **read-only** query +before relying on it. diff --git a/docs/model-import.md b/docs/model-import.md new file mode 100644 index 0000000..50a76d8 --- /dev/null +++ b/docs/model-import.md @@ -0,0 +1,417 @@ +# 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` | `/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//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://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 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. What a failure looks like in the logs + +Every failure — a row that cannot be claimed, a gate that rejected the bundle, an MLflow that refused +the connection, a verdict that could not be written — goes through the one channel in +`laborious/utils/models/import_errors.py`, which produces exactly two `logfmt` lines: the technical +one, whose message +carries the traceback (or the failure chain, root cause first, for anything that crossed an activity +boundary), and the user-facing sentence the notification logs. **Both carry the same fields**, so one +query returns the pair: + +| Field | What it holds | +|---|---| +| `import_run_id` | the record's primary key — the id a user quotes to support | +| `import_workflow_id` | the Temporal workflow id | +| `import_step` | the step that failed, or `not_claimable` / `status_write` | +| `import_code` | the code of § 8, the same one written to the row | +| `error_type` | the failing exception's own type — `MlflowException`, not the `ApplicationError` wrapping it | +| `error_summary` | one bounded line of what it said | +| `import_gate` | the gate that rejected the bundle, when one did | +| `import_verdict_not_recorded` | the verdict a failed terminal write could not store | + +The workflow metadata every other line already carries (`pod_id`, `runtime`, `model_id`, `model_name`, +`workflow_name`, `schedule_name`) is kept alongside them. + +A failure at any pipeline step is reported from the workflow's `except`, before cleanup and before the +terminal write, so the log reads in the order things happened and the line's step can never disagree +with the row's. That report never changes the outcome: its own failure is swallowed, and the failure it +was reporting is re-raised untouched. The Temporal SDK still prints its own traceback for each failed +attempt — that is the per-attempt record, and it is not structured; the channel's line is the one to +query. + +## 16. 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). diff --git a/docs/opc-communication.md b/docs/opc-communication.md new file mode 100644 index 0000000..15d78e7 --- /dev/null +++ b/docs/opc-communication.md @@ -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) diff --git a/docs/sientia-bundle-format.md b/docs/sientia-bundle-format.md new file mode 100644 index 0000000..209cb82 --- /dev/null +++ b/docs/sientia-bundle-format.md @@ -0,0 +1,100 @@ +# The `.sientia` bundle format + +A `.sientia` file is one exported MLflow model: its whole artifact tree, its run parameters and its +origin metadata, zipped and then encrypted under a password the exporter chose. + +**It is written by the Streamlit platform, `sientia-projects-templates`** (`app/src/operations/ +model_export/`). That application owns the export — its screen, its flow, its code — and documents +it. This page is the short version: the parameters both sides have to agree on, and what this +repository's reader accepts. + +## How it is generated + +The producer downloads the run's artifact tree whole, copies `run.data.params` **verbatim** (with +`target_variable` among them, or the import fails at gate 6), writes `metadata.json` first into a +`ZIP_DEFLATED` archive with the artifacts following as `artifacts/` and no directory +entries, encrypts that zip, and offers it as `{model_name}_v{model_version}.sientia`. + +`metadata.json` has exactly two keys: `parameters` — the run's parameters verbatim — and `metadata`, +six fields: `model_name`, `experiment_name`, `run_id`, `model_version`, `model_project`, +`export_timestamp` (UTC ISO 8601). + +## Encryption parameters + +The reader derives the key with the values the header carries, but refuses anything above the +producer's own cost — a header declaring more is a resource-exhaustion attempt, refused before any +memory is allocated. + +| | Value | +|---|---| +| KDF | Argon2id (`crypto_pwhash_ALG_ARGON2ID13`) | +| memlimit | `268435456` (256 MiB) — a **ceiling** on the reader's side | +| opslimit | `3` — likewise a ceiling | +| parallelism | `1` (reserved, not configurable) | +| salt | 16 bytes, fresh per bundle | +| key | 32 bytes | +| AEAD | XChaCha20-Poly1305 `secretstream`, 64 KiB plaintext chunks, last tagged `FINAL` | +| digest | SHA-256 of the **pushed chunks only** — on disk, `sha256(file[164:])`, past the 140-byte header and the 24-byte stream header. Hashing from 140 rejects every valid bundle | +| password | normalised `NFC`, encoded UTF-8 — no trimming, no case folding | + +The export-side password rule (12+ characters, mixed case, digit, symbol) is enforced there only: by +the time a bundle exists, the password is whatever it was, and the reader cannot re-check it. + +## Shape of the file + +`HEADER_STRUCT` is `>8sBBB16sQQB32s64s`. Big-endian, no alignment padding, so the offsets are exact: + +``` +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 +``` + +`format_version` is checked for equality, not for a range: an unknown version is refused at gate 2 +rather than parsed optimistically. `reserved_signature` must be all zero today, so a signed bundle +will be a new format version and not a silent change of meaning. + +`laborious/utils/bundle/format.py` is the normative layout on this side, and the golden fixture is +what keeps it and the producer from drifting apart. + +## How it is read + +`BundleReader.open()` runs seven ordered gates, each mapped to an import step and to one message a +person can read: + +| Gate | Step | Refuses | +|---|---|---| +| 1 | `download` | object too large, wrong prefix or suffix, digest ≠ the one the uploader sent | +| 2 | `download` | header that is not this format or not this version | +| 3 | `decryption` | wrong password, or altered bytes — the AEAD cannot tell them apart, and the message says so | +| 4 | `archive_inspection` | zip bomb, absolute or `..` entries, symlinks — read from the central directory, before extracting | +| 5 | `extraction` | anything that escapes the fresh work directory, or overwrites a file | +| 6 | `structure_validation` | missing `metadata.json` fields, missing `target_variable`, missing `prediction_model`/`data_model` | +| 7 | `content_policy` | a file that is not a known model file by name or suffix | + +Gates 1 and 2 share the `download` step on purpose: neither may borrow the `decryption` sentence, +which implicates the password. A gate-6 rejection naming `target_variable` means the origin run never +logged it — the fix is a re-export, on the producer's side. + +Error codes, sentences and the ceilings the gates read from configuration: +[`model-import.md`](model-import.md). + +## Fixtures + +- `tests/fixtures/bundle/` — a real bundle emitted by the producer, with its password and producer + commit recorded in that directory's `README.md`. It pays the producer's real Argon2id cost, so it + is used where the real bytes matter and not in every test. +- `tests/helpers/bundle_factory.py` — the test-only writer: same header at the same offsets, cheap + KDF parameters, and a knob for every hostile variant the gates above refuse.