QTZPOC-20

QTZPOC-20: Implement model import workflow for .sientia bundles

Introduces the import_model workflow and its associated activities to enable importing .sientia bundles into MLflow and MongoDB. This includes:

*   New core workflow and activity logic for handling bundle download, decryption, validation, MLflow provisioning (experiments, runs, registered versions), and MongoDB model document creation.
*   Comprehensive E2E and unit test suites, including a dedicated test harness and a manual run script for development.
*   Updated .env.example and README.md with new configurations and usage instructions.
*   Ensures correct MLflow model source resolution for legacy loader compatibility and robust MongoDB document management.
*   Refactors existing utility scripts into a scripts/ directory.

tests/laborious/activities/test_activities.py

@@ -156,6 +156,121 @@ def test___init__(
     )
 
 
+BASE_CONFIGS = {
+    'postgres_config': {
+        'host': 'localhost',
+        'port': 5432,
+        'user': 'postgres',
+        'password': 'postgres',
+        'dbname': 'postgres',
+        'min_connections': 1,
+        'max_connections': 10,
+    },
+    'minio_config': {
+        'endpoint_url': 'localhost:9000',
+        'access_key': 'minio',
+        'secret_key': 'minio123',
+        'default_bucket': 'test',
+        'retention_hours': 24,
+        'secure': False,
+    },
+    'mlflow_config': {'host': 'localhost', 'port': 5000, 'username': 'm', 'password': 'm'},
+    'opc_config': {},
+    'pi_web_api_config': {'base_url': 'https://pi', 'auth_type': 'bearer', 'auth_token': 't'},
+}
+
+IMPORT_CONFIG = {'bucket': 'sientia', 'prefix': 'imported_models/'}
+IMPORT_STATUS_CONFIG = {
+    'host': 'bff-db',
+    'port': 5432,
+    'dbname': 'sientia-core-mlops-bff',
+    'user': 'u',
+    'password': 'p',
+    'schema': 'public',
+}
+
+
+@patch('laborious.activities.activities.ModelImport.__init__')
+@patch('laborious.activities.activities.Storage.__init__')
+@patch('laborious.activities.activities.MLFlow.__init__')
+@patch('laborious.activities.activities.OPC.__init__')
+@patch('laborious.activities.activities.Gates.__init__')
+@patch('laborious.activities.activities.ModelMetrics.__init__')
+@patch('laborious.activities.activities.API.__init__')
+@patch('laborious.activities.activities.MongoDBRepository')
+@patch('laborious.activities.activities.MinioRepository')
+@patch('laborious.activities.activities.MetricsController')
+def test_the_import_activities_share_the_worker_s_mlflow_repository(  # noqa: PLR0913
+    _mock_metrics_controller,
+    mock_minio_repository,
+    mock_mongo_repository,
+    _mock_api_init,
+    _mock_model_metrics_init,
+    _mock_gates_init,
+    _mock_opc_init,
+    mock_mlflow_init,
+    _mock_storage_init,
+    mock_model_import_init,
+):
+    """Setting the MLflow tracking URI is a process-wide side effect, so two repository instances
+    would be two chances to disagree about where MLflow is."""
+    repository = MagicMock(name='model_monitoring_repository')
+    mock_mlflow_init.side_effect = lambda instance, **_: setattr(
+        instance, 'model_monitoring_repository', repository
+    )
+
+    Activities(
+        **BASE_CONFIGS,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        mongo_config={'connection_string': 'mongodb://mongo', 'database_name': 'sientia'},
+        import_config=IMPORT_CONFIG,
+        import_status_config=IMPORT_STATUS_CONFIG,
+    )
+
+    passed = mock_model_import_init.call_args.kwargs
+    assert passed['mlflow_repository'] is repository
+    assert passed['minio_repository'] is mock_minio_repository.return_value
+    assert passed['mongo_repository'] is mock_mongo_repository.return_value
+    assert passed['import_config'] is IMPORT_CONFIG
+    assert passed['status_config'] is IMPORT_STATUS_CONFIG
+
+
+@patch('laborious.activities.activities.ModelImport.__init__')
+@patch('laborious.activities.activities.Storage.__init__')
+@patch('laborious.activities.activities.MLFlow.__init__')
+@patch('laborious.activities.activities.OPC.__init__')
+@patch('laborious.activities.activities.Gates.__init__')
+@patch('laborious.activities.activities.ModelMetrics.__init__')
+@patch('laborious.activities.activities.API.__init__')
+@patch('laborious.activities.activities.MongoDBRepository')
+@patch('laborious.activities.activities.MinioRepository')
+@patch('laborious.activities.activities.MetricsController')
+def test_without_a_mongo_config_the_import_gets_no_repository(  # noqa: PLR0913
+    _mock_metrics_controller,
+    _mock_minio_repository,
+    mock_mongo_repository,
+    _mock_api_init,
+    _mock_model_metrics_init,
+    _mock_gates_init,
+    _mock_opc_init,
+    _mock_mlflow_init,
+    _mock_storage_init,
+    mock_model_import_init,
+):
+    """It refuses rather than guessing: the notification handler keeps its own client either way."""
+    Activities(
+        **BASE_CONFIGS,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        import_config=IMPORT_CONFIG,
+        import_status_config=IMPORT_STATUS_CONFIG,
+    )
+
+    assert mock_model_import_init.call_args.kwargs['mongo_repository'] is None
+    mock_mongo_repository.assert_not_called()
+
+
 @mark.asyncio
 @patch('laborious.activities.activities.Storage')
 @patch('laborious.activities.activities.MLFlow')

tests/laborious/activities/test_model_import_bundle.py

@@ -0,0 +1,565 @@
+"""Downloading the object and opening the bundle: the two activities that touch the secret.
+
+Three things are asserted here:
+
+- **the digest is enforced against the bytes that arrived**, before the header is parsed, with no
+  path that skips the comparison — a mismatch means the object behind that key is not the file the
+  user chose, and no amount of well-formedness makes it the right file;
+- **the envelope is opened inside the activity that uses the password**, so a plaintext password is
+  never an activity input, an activity result, or a value the workflow holds;
+- **nothing observable carries the secret** — for every failure path, the password, the envelope key
+  and the derived key are absent from the log, the notification, the metric labels, the exception and
+  the return value.
+
+The bundles here are real: the shared factory writes them field for field like the producer, at the
+test cost parameters, and the production reader opens them with real Argon2id and real
+`secretstream`.
+"""
+
+import hashlib
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from pytest import fixture, mark, raises
+
+from laborious import metrics
+from laborious.activities.model_import import ModelImport
+from laborious.activities.model_import_errors import ImportGateRejectedError
+from laborious.utils.bundle.steps import ImportStep
+from tests.helpers import bundle_factory as factory
+from tests.helpers.status_engine import FakeStatusEngine
+
+# Distinctive enough that a substring search for it in a log line cannot match by accident.
+BUNDLE_PASSWORD = 'Zk9-bundle-secret-Passw0rd'
+ENVELOPE_KEY = bytes(range(100, 132))
+OTHER_ENVELOPE_KEY = bytes(range(200, 232))
+
+METADATA = {
+    'model_id': '77',
+    'model_name': 'factory_model',
+    'workflow_name': 'import_model',
+    'schedule_name': '-',
+}
+WORKFLOW_ID = 'import_model-77-abc'
+OBJECT_KEY = 'imported_models/factory_model_v2.sientia'
+
+
+def import_config(work_dir: Path, **overrides) -> dict:
+    return {
+        'password_key': factory.envelope_key_b64(ENVELOPE_KEY),
+        'bucket': 'sientia',
+        'prefix': 'imported_models/',
+        'work_dir': str(work_dir),
+        'max_object_bytes': 8 * 1024 * 1024,
+        'max_entries': 5000,
+        'max_uncompressed_bytes': 4 * 1024 * 1024,
+        'max_compression_ratio': 200.0,
+        'models_collection': 'models',
+        'retention_days': 7,
+        **overrides,
+    }
+
+
+@fixture
+def minio_client() -> MagicMock:
+    client = MagicMock()
+    client.stat_object.return_value = MagicMock(size=1024)
+    return client
+
+
+@fixture
+def import_activity(tmp_path: Path, minio_client: MagicMock) -> ModelImport:
+    repository = MagicMock()
+    repository.minio_client = minio_client
+    with patch('laborious.activities.model_import.create_engine', return_value=FakeStatusEngine()):
+        activity = ModelImport(
+            status_config={
+                'host': 'h',
+                'port': 5432,
+                'dbname': 'sientia-core-mlops-bff',
+                'user': 'u',
+                'password': 'p',
+                'schema': 'public',
+            },
+            import_config=import_config(tmp_path),
+            minio_repository=repository,
+            mlflow_repository=MagicMock(),
+            mongo_repository=AsyncMock(),
+            logger=MagicMock(),
+            notification_handler=MagicMock(),
+            metrics_controller=AsyncMock(),
+        )
+    activity.emit_metric = AsyncMock()
+    activity.send_notification_async = AsyncMock()
+    return activity
+
+
+def serve_object(minio_client: MagicMock, payload: bytes) -> None:
+    """Make `fget_object` write `payload` where the activity asked for it."""
+
+    def write(_bucket, _key, destination):
+        Path(destination).write_bytes(payload)
+
+    minio_client.stat_object.return_value = MagicMock(size=len(payload))
+    minio_client.fget_object.side_effect = write
+
+
+def download_input(digest: str, **overrides) -> dict:
+    return {
+        'metadata': METADATA,
+        'import_run_id': 77,
+        'workflow_id': WORKFLOW_ID,
+        'object_key': OBJECT_KEY,
+        'expected_digest': digest,
+        **overrides,
+    }
+
+
+def open_input(path: str, envelope: str | None = None, **overrides) -> dict:
+    return {
+        'metadata': METADATA,
+        'import_run_id': 77,
+        'workflow_id': WORKFLOW_ID,
+        'path': path,
+        'password_envelope': envelope
+        if envelope is not None
+        else factory.build_envelope(BUNDLE_PASSWORD, ENVELOPE_KEY),
+        **overrides,
+    }
+
+
+def observable_text(activity: ModelImport) -> str:
+    """Everything this activity said out loud: logs, notifications and metric labels."""
+    logger = activity.logger
+    parts = [
+        str(call) for call in logger.custom_info.call_args_list + logger.custom_error.call_args_list
+    ]
+    parts += [str(call) for call in activity.send_notification_async.await_args_list]
+    parts += [str(call) for call in activity.emit_metric.await_args_list]
+    return '\n'.join(parts)
+
+
+# ------------------------------------------------------------------------- the download
+
+
+@mark.asyncio
+async def test_a_matching_digest_yields_the_downloaded_file(
+    import_activity: ModelImport, minio_client: MagicMock, tmp_path: Path
+):
+    payload = factory.make_bundle(password=BUNDLE_PASSWORD)
+    serve_object(minio_client, payload)
+    digest = hashlib.sha256(payload).hexdigest()
+
+    result = await import_activity.download_import_bundle(download_input(digest))
+
+    assert Path(result['path']).read_bytes() == payload
+    assert result['digest'] == digest
+    assert result['size'] == len(payload)
+    assert Path(result['path']).parent == tmp_path / WORKFLOW_ID
+
+
+@mark.asyncio
+async def test_the_configured_bucket_is_used_when_the_request_names_none(
+    import_activity: ModelImport, minio_client: MagicMock
+):
+    payload = factory.make_bundle(password=BUNDLE_PASSWORD)
+    serve_object(minio_client, payload)
+
+    await import_activity.download_import_bundle(
+        download_input(hashlib.sha256(payload).hexdigest())
+    )
+
+    assert minio_client.stat_object.call_args.args[0] == 'sientia'
+
+
+@mark.asyncio
+async def test_the_request_may_name_another_bucket(
+    import_activity: ModelImport, minio_client: MagicMock
+):
+    payload = factory.make_bundle(password=BUNDLE_PASSWORD)
+    serve_object(minio_client, payload)
+
+    await import_activity.download_import_bundle(
+        download_input(hashlib.sha256(payload).hexdigest(), bucket='other-bucket')
+    )
+
+    assert minio_client.stat_object.call_args.args[0] == 'other-bucket'
+
+
+@mark.asyncio
+async def test_a_digest_from_another_file_fails_before_the_header_is_parsed(
+    import_activity: ModelImport, minio_client: MagicMock
+):
+    payload = factory.make_bundle(password=BUNDLE_PASSWORD)
+    other = factory.make_bundle(password=BUNDLE_PASSWORD, salt=b'\x01' * 16)
+    serve_object(minio_client, payload)
+
+    with raises(ImportGateRejectedError) as failure:
+        await import_activity.download_import_bundle(
+            download_input(hashlib.sha256(other).hexdigest())
+        )
+
+    assert failure.value.details[0]['step'] == ImportStep.DOWNLOAD.value
+    assert failure.value.details[0]['gate'] == 1
+
+
+@mark.asyncio
+async def test_a_file_that_does_not_match_is_not_left_on_the_worker(
+    import_activity: ModelImport, minio_client: MagicMock, tmp_path: Path
+):
+    serve_object(minio_client, factory.make_bundle(password=BUNDLE_PASSWORD))
+
+    with raises(ImportGateRejectedError):
+        await import_activity.download_import_bundle(download_input('0' * 64))
+
+    assert not (tmp_path / WORKFLOW_ID / 'bundle.sientia').exists()
+
+
+@mark.asyncio
+async def test_the_digest_comparison_is_case_insensitive_on_the_input(
+    import_activity: ModelImport, minio_client: MagicMock
+):
+    payload = factory.make_bundle(password=BUNDLE_PASSWORD)
+    serve_object(minio_client, payload)
+
+    result = await import_activity.download_import_bundle(
+        download_input(hashlib.sha256(payload).hexdigest().upper())
+    )
+
+    assert result['digest'] == hashlib.sha256(payload).hexdigest()
+
+
+@mark.asyncio
+async def test_an_object_over_the_ceiling_is_refused_before_its_body_is_fetched(
+    import_activity: ModelImport, minio_client: MagicMock
+):
+    minio_client.stat_object.return_value = MagicMock(size=9 * 1024 * 1024)
+
+    with raises(ImportGateRejectedError) as failure:
+        await import_activity.download_import_bundle(download_input('a' * 64))
+
+    assert failure.value.details[0]['step'] == ImportStep.DOWNLOAD.value
+    minio_client.fget_object.assert_not_called()
+
+
+@mark.asyncio
+@mark.parametrize(
+    'object_key',
+    ['other_prefix/model.sientia', 'imported_models/model.zip'],
+)
+async def test_a_key_of_the_wrong_shape_is_refused(
+    import_activity: ModelImport, minio_client: MagicMock, object_key: str
+):
+    with raises(ImportGateRejectedError):
+        await import_activity.download_import_bundle(
+            download_input('a' * 64, object_key=object_key)
+        )
+
+    minio_client.fget_object.assert_not_called()
+
+
+@mark.asyncio
+async def test_an_unreadable_object_is_a_download_failure(
+    import_activity: ModelImport, minio_client: MagicMock
+):
+    minio_client.stat_object.side_effect = OSError('no such object')
+
+    with raises(ImportGateRejectedError) as failure:
+        await import_activity.download_import_bundle(download_input('a' * 64))
+
+    assert failure.value.details[0]['step'] == ImportStep.DOWNLOAD.value
+
+
+@mark.asyncio
+async def test_a_failed_transfer_is_a_download_failure(
+    import_activity: ModelImport, minio_client: MagicMock
+):
+    minio_client.fget_object.side_effect = OSError('connection reset')
+
+    with raises(ImportGateRejectedError) as failure:
+        await import_activity.download_import_bundle(download_input('a' * 64))
+
+    assert failure.value.details[0]['step'] == ImportStep.DOWNLOAD.value
+
+
+@mark.asyncio
+async def test_a_missing_minio_repository_is_a_configuration_error(tmp_path: Path):
+    with patch('laborious.activities.model_import.create_engine', return_value=FakeStatusEngine()):
+        activity = ModelImport(
+            status_config={
+                'host': 'h',
+                'port': 5432,
+                'dbname': 'd',
+                'user': 'u',
+                'password': 'p',
+                'schema': 'public',
+            },
+            import_config=import_config(tmp_path),
+            logger=MagicMock(),
+            notification_handler=MagicMock(),
+            metrics_controller=AsyncMock(),
+        )
+
+    with raises(ValueError, match='Minio repository not initialized'):
+        await activity.download_import_bundle(download_input('a' * 64))
+
+
+# ----------------------------------------------------------------------- opening the bundle
+
+
+@mark.asyncio
+async def test_a_well_formed_bundle_opens_and_returns_nothing_secret(
+    import_activity: ModelImport, tmp_path: Path
+):
+    path = tmp_path / 'bundle.sientia'
+    path.write_bytes(factory.make_bundle(password=BUNDLE_PASSWORD))
+
+    result = await import_activity.open_import_bundle(open_input(str(path)))
+
+    assert result['metadata']['model_name'] == 'factory_model'
+    assert result['parameters']['target_variable'] == 'SE'
+    assert Path(result['extracted_dir']).is_dir()
+    assert (Path(result['extracted_dir']) / 'artifacts' / 'prediction_model' / 'MLmodel').is_file()
+    assert set(result) == {'extracted_dir', 'metadata', 'parameters', 'digest'}
+    assert BUNDLE_PASSWORD not in str(result)
+
+
+@mark.asyncio
+async def test_opening_a_bundle_measures_the_kdf(import_activity: ModelImport, tmp_path: Path):
+    path = tmp_path / 'bundle.sientia'
+    path.write_bytes(factory.make_bundle(password=BUNDLE_PASSWORD))
+
+    await import_activity.open_import_bundle(open_input(str(path)))
+
+    observed = [
+        call.kwargs
+        for call in import_activity.emit_metric.await_args_list
+        if call.kwargs.get('metric_object') is metrics.MODEL_IMPORT_KDF_DURATION
+    ]
+    assert len(observed) == 1
+    assert observed[0]['method'] == 'observe'
+    assert observed[0]['value'] > 0
+
+
+@mark.asyncio
+async def test_a_wrong_envelope_key_is_not_confused_with_a_wrong_bundle_password(
+    import_activity: ModelImport, tmp_path: Path
+):
+    path = tmp_path / 'bundle.sientia'
+    path.write_bytes(factory.make_bundle(password=BUNDLE_PASSWORD))
+    envelope = factory.build_envelope(BUNDLE_PASSWORD, OTHER_ENVELOPE_KEY)
+
+    with raises(ImportGateRejectedError) as failure:
+        await import_activity.open_import_bundle(open_input(str(path), envelope))
+
+    details = failure.value.details[0]
+    assert details['step'] == ImportStep.DECRYPTION.value
+    assert details['gate'] == 3
+    # The platform's own transport failed, not the user's password. Same step — the step is where the
+    # import stopped, not who is at fault — but a distinguishable cause.
+    assert details['cause'] == 'envelope'
+
+
+@mark.asyncio
+async def test_a_wrong_bundle_password_is_a_decryption_failure_of_its_own(
+    import_activity: ModelImport, tmp_path: Path
+):
+    path = tmp_path / 'bundle.sientia'
+    path.write_bytes(factory.make_bundle(password='a-different-password'))
+
+    with raises(ImportGateRejectedError) as failure:
+        await import_activity.open_import_bundle(open_input(str(path)))
+
+    details = failure.value.details[0]
+    assert details['step'] == ImportStep.DECRYPTION.value
+    assert details['cause'] != 'envelope'
+
+
+@mark.asyncio
+async def test_a_missing_password_key_stops_the_import_at_decryption(
+    import_activity: ModelImport, tmp_path: Path
+):
+    import_activity.import_config['password_key'] = None
+    path = tmp_path / 'bundle.sientia'
+    path.write_bytes(factory.make_bundle(password=BUNDLE_PASSWORD))
+
+    with raises(ImportGateRejectedError) as failure:
+        await import_activity.open_import_bundle(open_input(str(path)))
+
+    assert failure.value.details[0]['cause'] == 'envelope'
+
+
+@mark.asyncio
+@mark.parametrize(
+    'envelope',
+    ['', 'not-base64!!', 'AAAA', 'bm90LWFuLWVudmVsb3BlLWF0LWFsbA=='],
+    ids=['empty', 'not-base64', 'too-short', 'wrong-bytes'],
+)
+async def test_a_tampered_or_malformed_envelope_is_refused(
+    import_activity: ModelImport, tmp_path: Path, envelope: str
+):
+    path = tmp_path / 'bundle.sientia'
+    path.write_bytes(factory.make_bundle(password=BUNDLE_PASSWORD))
+
+    with raises(ImportGateRejectedError) as failure:
+        await import_activity.open_import_bundle(open_input(str(path), envelope))
+
+    assert failure.value.details[0]['cause'] == 'envelope'
+
+
+def bad_magic() -> bytes:
+    return factory.make_bundle(password=BUNDLE_PASSWORD, magic=b'NOTSIEN1')
+
+
+def wrong_digest() -> bytes:
+    return factory.make_bundle(password=BUNDLE_PASSWORD, digest_override=bytes(32))
+
+
+def tampered() -> bytes:
+    return factory.tamper_ciphertext(factory.make_bundle(password=BUNDLE_PASSWORD))
+
+
+def traversal() -> bytes:
+    return factory.make_bundle(
+        factory.ZipSpec(traversal_entry='artifacts/../../escaped.csv'), password=BUNDLE_PASSWORD
+    )
+
+
+def absolute() -> bytes:
+    return factory.make_bundle(
+        factory.ZipSpec(absolute_entry='/etc/passwd'), password=BUNDLE_PASSWORD
+    )
+
+
+def missing_data_model() -> bytes:
+    files = {
+        name: content
+        for name, content in factory.default_tree().items()
+        if not name.startswith('artifacts/data_model/')
+    }
+    return factory.make_bundle(factory.ZipSpec(files=files), password=BUNDLE_PASSWORD)
+
+
+def missing_target_variable() -> bytes:
+    document = factory.build_metadata_document(drop_parameter='target_variable')
+    return factory.make_bundle(factory.ZipSpec(document=document), password=BUNDLE_PASSWORD)
+
+
+def script_in_the_tree() -> bytes:
+    return factory.make_bundle(
+        factory.ZipSpec(extra_files={'artifacts/run.sh': b'#!/bin/sh\nrm -rf /\n'}),
+        password=BUNDLE_PASSWORD,
+    )
+
+
+@mark.asyncio
+@mark.parametrize(
+    ('build', 'step', 'gate'),
+    [
+        (bad_magic, ImportStep.DOWNLOAD, 2),
+        (wrong_digest, ImportStep.DOWNLOAD, 2),
+        (tampered, ImportStep.DECRYPTION, 3),
+        (traversal, ImportStep.ARCHIVE_INSPECTION, 4),
+        (absolute, ImportStep.ARCHIVE_INSPECTION, 4),
+        (missing_data_model, ImportStep.STRUCTURE_VALIDATION, 6),
+        (missing_target_variable, ImportStep.STRUCTURE_VALIDATION, 6),
+        (script_in_the_tree, ImportStep.CONTENT_POLICY, 7),
+    ],
+    ids=lambda value: getattr(value, '__name__', str(value)),
+)
+async def test_a_gate_failure_reaches_the_activity_as_its_own_step(
+    import_activity: ModelImport, tmp_path: Path, build, step: ImportStep, gate: int
+):
+    path = tmp_path / 'bundle.sientia'
+    path.write_bytes(build())
+
+    with raises(ImportGateRejectedError) as failure:
+        await import_activity.open_import_bundle(open_input(str(path)))
+
+    assert failure.value.details[0]['step'] == step.value
+    assert failure.value.details[0]['gate'] == gate
+
+
+@mark.asyncio
+async def test_a_rejected_bundle_leaves_no_extracted_tree(
+    import_activity: ModelImport, tmp_path: Path
+):
+    path = tmp_path / 'bundle.sientia'
+    path.write_bytes(traversal())
+
+    with raises(ImportGateRejectedError):
+        await import_activity.open_import_bundle(open_input(str(path)))
+
+    extract_root = tmp_path / WORKFLOW_ID / 'extract'
+    assert not extract_root.exists() or list(extract_root.rglob('*')) == []
+
+
+# ---------------------------------------------------------------------------- secrecy
+
+
+SECRET_PATHS = [
+    ('happy', lambda: factory.make_bundle(password=BUNDLE_PASSWORD), None),
+    ('wrong-bundle-password', lambda: factory.make_bundle(password='another-password'), None),
+    ('wrong-envelope-key', lambda: factory.make_bundle(password=BUNDLE_PASSWORD), 'other-key'),
+    ('malformed-envelope', lambda: factory.make_bundle(password=BUNDLE_PASSWORD), 'garbage'),
+    ('gate-2', bad_magic, None),
+    ('gate-3', tampered, None),
+    ('gate-4', traversal, None),
+    ('gate-6', missing_data_model, None),
+    ('gate-7', script_in_the_tree, None),
+]
+
+
+@mark.asyncio
+@mark.parametrize(
+    ('build', 'envelope_kind'),
+    [(build, kind) for _, build, kind in SECRET_PATHS],
+    ids=[name for name, _, _ in SECRET_PATHS],
+)
+async def test_no_failure_path_says_the_password_out_loud(
+    import_activity: ModelImport, tmp_path: Path, build, envelope_kind: str | None
+):
+    """Every path through this activity, checked against the three things that must never surface."""
+    path = tmp_path / 'bundle.sientia'
+    path.write_bytes(build())
+    if envelope_kind == 'other-key':
+        envelope = factory.build_envelope(BUNDLE_PASSWORD, OTHER_ENVELOPE_KEY)
+    elif envelope_kind == 'garbage':
+        envelope = 'not-base64!!'
+    else:
+        envelope = factory.build_envelope(BUNDLE_PASSWORD, ENVELOPE_KEY)
+
+    raised: BaseException | None = None
+    result = None
+    try:
+        result = await import_activity.open_import_bundle(open_input(str(path), envelope))
+    except BaseException as error:  # noqa: BLE001 - the point is that *whatever* it is says nothing
+        raised = error
+
+    secrets = [
+        BUNDLE_PASSWORD,
+        'another-password',
+        factory.envelope_key_b64(ENVELOPE_KEY),
+        ENVELOPE_KEY.hex(),
+    ]
+    surfaces = [observable_text(import_activity), str(result)]
+    if raised is not None:
+        surfaces += [str(raised), str(getattr(raised, 'details', '')), repr(raised)]
+    for secret in secrets:
+        for surface in surfaces:
+            assert secret not in surface
+
+
+@mark.asyncio
+async def test_the_envelope_never_leaves_the_activity_that_opens_it(
+    import_activity: ModelImport, tmp_path: Path
+):
+    """The result is what crosses back into the workflow, and therefore into the event history."""
+    path = tmp_path / 'bundle.sientia'
+    path.write_bytes(factory.make_bundle(password=BUNDLE_PASSWORD))
+    envelope = factory.build_envelope(BUNDLE_PASSWORD, ENVELOPE_KEY)
+
+    result = await import_activity.open_import_bundle(open_input(str(path), envelope))
+
+    assert envelope not in str(result)
+    assert 'password' not in str(result).lower()

tests/laborious/activities/test_model_import_errors.py

@@ -0,0 +1,410 @@
+"""The failure catalog and the one channel every import failure passes through.
+
+Two contracts are asserted here, and both are about what a person is allowed to see:
+
+- **the catalog is closed and total** — every pipeline step that can fail has exactly one code and
+  one sentence, the sentence never carries a path, a table name, an exception class or a traceback,
+  and the classification is a function of the *step* alone, so rewording a sentence cannot move a
+  code the frontend switches on;
+- **the channel is the only reporter** — one call reaches the log, the metric and the notification,
+  the exception's text reaches nothing but the log, and no other module in the import path logs an
+  error or notifies (asserted from the source, so the rule cannot rot).
+"""
+
+import ast
+from pathlib import Path
+from typing import Any
+
+from pytest import mark, raises
+
+from laborious import metrics
+from laborious.activities.model_import_errors import (
+    IMPORT_FAILURE_CATALOG,
+    STEPS_WITHOUT_A_USER_FACING_REASON,
+    UNCLASSIFIED_FAILURE,
+    ImportErrorCode,
+    ImportGateRejectedError,
+    ImportInputError,
+    ImportLogRowNotClaimableError,
+    ImportStatusRejectedError,
+    classify_import_failure,
+    failure_location,
+    report_import_failure,
+)
+from laborious.utils.bundle.steps import GATE_TO_STEP, ImportStep
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[3]
+
+# The modules that make up the import path. The channel lives in the third one; the other two may
+# not report a failure of their own.
+IMPORT_MODULES = (
+    REPOSITORY_ROOT / 'laborious/activities/model_import.py',
+    REPOSITORY_ROOT / 'laborious/workflows/import_model.py',
+)
+
+# What a leaked exception message looks like: a work directory, the table, an exception class and a
+# traceback line. None of it may reach a recorded reason.
+LEAKY_MESSAGE = (
+    "KeyError: 'target_variable' while writing public.experiment_run from "
+    '/tmp/sientia-import/import_model-77-abc/extract\n'  # noqa: S108 - the leak under test
+    '  File "laborious/activities/model_import.py", line 1, in write_import_model_document'
+)
+
+# Every fragment of the leaky message that must not reach a recorded reason.
+LEAKS_THAT_MUST_NOT_SURFACE = (
+    'sientia-import',
+    'public.experiment_run',
+    'KeyError',
+    'File "',
+    'line 1',
+)
+
+STEPS_THAT_ONLY_SUPPORT_CAN_ACT_ON = (
+    ImportStep.EXPERIMENT_CREATION,
+    ImportStep.ARTIFACT_UPLOAD,
+    ImportStep.REGISTRATION,
+    ImportStep.MODEL_DOCUMENT,
+)
+
+
+class FakeSinks:
+    """The observability surface the channel writes to, recorded rather than performed."""
+
+    def __init__(self) -> None:
+        self.logged: list[tuple[str, dict]] = []
+        self.metrics: list[tuple[Any, dict]] = []
+        self.notifications: list[dict] = []
+
+    def error(self, message: str, metadata: dict | None = None) -> None:
+        self.logged.append((message, metadata or {}))
+
+    def get_core_labels(self, metadata: dict | None = None, operation_type: str = '-') -> dict:
+        return {
+            'model_name': (metadata or {}).get('model_name', '-'),
+            'operation_type': operation_type,
+        }
+
+    async def emit_metric(
+        self,
+        metric_object: Any,
+        tags: dict[str, Any],
+        method: str = 'inc',
+        value: float | None = None,
+    ) -> None:
+        self.metrics.append((metric_object, tags))
+
+    async def send_notification_async(
+        self,
+        metadata: dict,
+        notification_id: str,
+        message: str,
+        block: str,
+        level: str = 'INFO',
+        attachment_content: str | None = None,
+    ) -> None:
+        self.notifications.append(
+            {
+                'metadata': metadata,
+                'notification_id': notification_id,
+                'message': message,
+                'block': block,
+                'level': level,
+            }
+        )
+
+
+# ------------------------------------------------------------------------- the catalog
+
+
+def test_every_failing_step_has_exactly_one_entry():
+    """A step added without a sentence fails here rather than reaching a user as `None`."""
+    expected = set(ImportStep) - STEPS_WITHOUT_A_USER_FACING_REASON
+    assert set(IMPORT_FAILURE_CATALOG) == expected
+
+
+def test_cleanup_is_the_only_step_without_a_sentence():
+    """Cleanup failing is a log line on a successful import, never the reported failure."""
+    assert STEPS_WITHOUT_A_USER_FACING_REASON == frozenset({ImportStep.CLEANUP})
+    assert ImportStep.CLEANUP not in IMPORT_FAILURE_CATALOG
+
+
+def test_every_gate_maps_to_a_step_that_has_a_sentence():
+    for step in GATE_TO_STEP.values():
+        assert step in IMPORT_FAILURE_CATALOG
+
+
+def test_each_step_has_its_own_code():
+    codes = [code for code, _ in IMPORT_FAILURE_CATALOG.values()]
+    assert len(codes) == len(set(codes))
+
+
+def test_the_codes_that_never_reach_a_record_are_absent_from_the_catalog():
+    catalogued = {code for code, _ in IMPORT_FAILURE_CATALOG.values()}
+    for code in (
+        ImportErrorCode.LOG_ROW_NOT_CLAIMABLE,
+        ImportErrorCode.STATUS_NOT_RECORDED,
+        ImportErrorCode.CLEANUP_INCOMPLETE,
+    ):
+        assert code not in catalogued
+
+
+@mark.parametrize('step', list(IMPORT_FAILURE_CATALOG))
+def test_a_reason_never_carries_the_exception(step: ImportStep):
+    failure = classify_import_failure(step, RuntimeError(LEAKY_MESSAGE))
+
+    for leak in LEAKS_THAT_MUST_NOT_SURFACE:
+        assert leak not in failure.reason
+    assert failure.reason == IMPORT_FAILURE_CATALOG[step][1]
+
+
+@mark.parametrize('step', list(IMPORT_FAILURE_CATALOG))
+def test_the_classification_is_a_function_of_the_step_alone(step: ImportStep):
+    """Which is what makes a reworded sentence keep its code: neither is read from the exception."""
+    first = classify_import_failure(step, RuntimeError(LEAKY_MESSAGE))
+    second = classify_import_failure(step, KeyError('target_variable'))
+    third = classify_import_failure(step, None)
+
+    assert first == second == third
+    assert first.code is IMPORT_FAILURE_CATALOG[step][0]
+
+
+def test_the_decryption_reason_does_not_say_which_of_the_two_it_was():
+    reason = IMPORT_FAILURE_CATALOG[ImportStep.DECRYPTION][1]
+
+    # Both causes are named as possibilities; neither is asserted, because the AEAD cannot tell them
+    # apart and guessing would be a hint about somebody else's password.
+    assert 'password is wrong' in reason
+    assert 'damaged' in reason
+    assert 'altered' in reason
+
+
+@mark.parametrize('step', STEPS_THAT_ONLY_SUPPORT_CAN_ACT_ON)
+def test_a_reason_the_user_cannot_act_on_names_the_import_identifier(step: ImportStep):
+    assert "Contact support with this import's identifier." in IMPORT_FAILURE_CATALOG[step][1]
+
+
+def test_a_failure_with_no_step_is_unclassified():
+    failure = classify_import_failure(None, RuntimeError('boom'))
+
+    assert failure.step is None
+    assert failure.code is ImportErrorCode.UNEXPECTED_ERROR
+    assert failure.reason == UNCLASSIFIED_FAILURE[1]
+
+
+def test_every_reason_is_long_enough_for_the_column():
+    """`error_message` carries a `LENGTH(...) >= 3` CHECK; every sentence here clears it."""
+    for _, reason in IMPORT_FAILURE_CATALOG.values():
+        assert len(reason) >= 3
+
+
+# ----------------------------------------------------------------- the typed failures
+
+
+def test_the_two_unrepeatable_failures_are_non_retryable():
+    assert ImportLogRowNotClaimableError('x').non_retryable is True
+    assert ImportStatusRejectedError('x').non_retryable is True
+    assert ImportInputError('x').non_retryable is True
+
+
+def test_a_gate_rejection_carries_its_step_and_gate_across_the_boundary():
+    """An exception loses its Python type crossing an activity boundary but keeps its details."""
+    error = ImportGateRejectedError('rejected', step=ImportStep.ARCHIVE_INSPECTION, gate=4)
+
+    assert failure_location(error) == ('archive_inspection', 4)
+
+
+def test_an_input_error_reports_the_received_step():
+    assert failure_location(ImportInputError('missing object_key')) == ('received', None)
+
+
+def test_a_located_failure_is_found_through_the_activity_wrapper():
+    wrapper = RuntimeError('activity failed')
+    wrapper.__cause__ = ImportGateRejectedError('nope', step=ImportStep.EXTRACTION, gate=5)
+
+    assert failure_location(wrapper) == ('extraction', 5)
+
+
+def test_an_unlocated_failure_says_nothing_rather_than_guessing():
+    assert failure_location(RuntimeError('boom')) == (None, None)
+    assert failure_location(None) == (None, None)
+
+
+def test_a_cause_chain_cannot_loop_forever():
+    first = RuntimeError('a')
+    second = RuntimeError('b')
+    first.__cause__ = second
+    second.__cause__ = first
+
+    assert failure_location(first) == (None, None)
+
+
+# ------------------------------------------------------------------------- the channel
+
+
+@mark.asyncio
+async def test_the_channel_hits_every_sink_once():
+    sinks = FakeSinks()
+
+    failure = await report_import_failure(
+        sinks,
+        ImportStep.REGISTRATION,
+        RuntimeError(LEAKY_MESSAGE),
+        import_run_id=77,
+        workflow_id='import_model-77-abc',
+        metadata={'model_name': 'imported_model'},
+    )
+
+    assert failure.code is ImportErrorCode.MODEL_NOT_REGISTERED
+    assert len(sinks.logged) == 1
+    assert len(sinks.metrics) == 1
+    assert len(sinks.notifications) == 1
+    assert sinks.metrics[0][0] is metrics.MODEL_IMPORT_ERROR_COUNT
+    assert sinks.metrics[0][1]['operation_type'] == ImportErrorCode.MODEL_NOT_REGISTERED.value
+
+
+@mark.asyncio
+async def test_the_exception_text_reaches_the_log_and_nothing_else():
+    sinks = FakeSinks()
+    try:
+        raise RuntimeError(LEAKY_MESSAGE)
+    except RuntimeError as raised:
+        caught = raised
+
+    await report_import_failure(
+        sinks,
+        ImportStep.MODEL_DOCUMENT,
+        caught,
+        import_run_id=77,
+        workflow_id='import_model-77-abc',
+    )
+
+    logged, _ = sinks.logged[0]
+    assert 'KeyError' in logged
+    assert 'public.experiment_run' in logged
+    assert 'Traceback' in logged
+    notification = sinks.notifications[0]
+    assert notification['message'] == IMPORT_FAILURE_CATALOG[ImportStep.MODEL_DOCUMENT][1]
+    assert 'KeyError' not in notification['message']
+    assert 'KeyError' not in str(sinks.metrics[0][1])
+
+
+@mark.asyncio
+async def test_the_log_line_correlates_the_record_and_the_workflow():
+    sinks = FakeSinks()
+
+    await report_import_failure(
+        sinks,
+        ImportStep.DOWNLOAD,
+        RuntimeError('boom'),
+        import_run_id=77,
+        workflow_id='import_model-77-abc',
+    )
+
+    logged, _ = sinks.logged[0]
+    assert 'import_run_id=77' in logged
+    assert 'workflow_id=import_model-77-abc' in logged
+    assert 'step=download' in logged
+
+
+@mark.asyncio
+async def test_a_failure_that_is_not_a_step_reports_under_its_marker():
+    sinks = FakeSinks()
+
+    failure = await report_import_failure(
+        sinks,
+        None,
+        None,
+        import_run_id=77,
+        workflow_id='wf',
+        code=ImportErrorCode.STATUS_NOT_RECORDED,
+        marker='status_write',
+        verdict='succeeded=True step=model_document gate=None',
+        detail_text='OperationalError: connection refused',
+    )
+
+    logged, _ = sinks.logged[0]
+    assert failure.code is ImportErrorCode.STATUS_NOT_RECORDED
+    assert 'step=status_write' in logged
+    assert 'verdict_not_recorded=succeeded=True step=model_document gate=None' in logged
+    assert 'connection refused' in logged
+    assert sinks.notifications[0]['block'] == 'status_write'
+
+
+@mark.asyncio
+async def test_a_report_without_an_exception_object_says_so():
+    sinks = FakeSinks()
+
+    await report_import_failure(sinks, None, None, import_run_id=77, workflow_id='wf')
+
+    assert 'no exception object available' in sinks.logged[0][0]
+
+
+@mark.asyncio
+async def test_an_explicit_code_keeps_the_step_s_sentence():
+    """The cleanup case: its own code, and a sentence a person can still read."""
+    sinks = FakeSinks()
+
+    failure = await report_import_failure(
+        sinks,
+        ImportStep.CLEANUP,
+        OSError('device busy'),
+        import_run_id=77,
+        workflow_id='wf',
+        code=ImportErrorCode.CLEANUP_INCOMPLETE,
+        marker=ImportStep.CLEANUP.value,
+    )
+
+    assert failure.code is ImportErrorCode.CLEANUP_INCOMPLETE
+    assert failure.reason == UNCLASSIFIED_FAILURE[1]
+    assert sinks.notifications[0]['level'] == 'ERROR'
+
+
+# ------------------------------------------------- the channel is the only reporter
+
+
+def _reporting_calls(path: Path) -> list[str]:
+    """Every call in `path` that would log an error or send a notification."""
+    forbidden = {
+        'error',
+        'custom_error',
+        'critical',
+        'custom_critical',
+        'send_notification',
+        'send_notification_async',
+        'publish_notification',
+    }
+    tree = ast.parse(path.read_text())
+    found = []
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call):
+            continue
+        target = node.func
+        name = target.attr if isinstance(target, ast.Attribute) else getattr(target, 'id', '')
+        if name in forbidden:
+            found.append(f'{path.name}:{node.lineno}:{name}')
+    return found
+
+
+@mark.parametrize('module_path', IMPORT_MODULES, ids=lambda path: path.name)
+def test_no_import_module_reports_a_failure_of_its_own(module_path: Path):
+    """So no failure is reported twice, or half-reported by whoever noticed it first."""
+    assert _reporting_calls(module_path) == []
+
+
+def test_the_channel_itself_is_where_the_reporting_lives():
+    channel = REPOSITORY_ROOT / 'laborious/activities/model_import_errors.py'
+
+    calls = _reporting_calls(channel)
+
+    assert [call.split(':')[2] for call in calls] == ['error', 'send_notification_async']
+
+
+def test_the_reader_neither_logs_nor_notifies():
+    """The bundle reader raises; it is the activity layer that decides what a failure means."""
+    for module in ('laborious/utils/bundle/reader.py', 'laborious/utils/bundle/format.py'):
+        assert _reporting_calls(REPOSITORY_ROOT / module) == []
+
+
+def test_an_unclassified_step_cannot_be_recorded_as_a_known_one():
+    with raises(ValueError, match='not a valid ImportStep'):
+        ImportStep('provisioning')

tests/laborious/activities/test_model_import_provisioning.py

@@ -0,0 +1,656 @@
+"""Provisioning: the experiment, the run, the registered version, the document, the cleanup.
+
+The boundary this file defends is where the import stops. A finished import leaves a version whose
+stage was never touched and a document that is inert — promoting and activating are an operator's
+decisions, owned by another card — so the tests assert not only what is written but what is *not*:
+no stage transition, no alias, no `active: true`, no update of a document already there.
+
+The rest is the model listing's arithmetic (`max(int(id)) + 1`, which is not `count` and not a
+lexicographic sort) and the two operations that must never fail an import that otherwise worked:
+removing the scratch directory and putting an expiry rule on the bundle prefix.
+"""
+
+import json
+import shutil
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import mlflow as mlflow_lib
+from pytest import fixture, mark, raises
+
+from laborious.activities.model_import import ModelImport
+from laborious.utils.repository.model_repository import MLFlowRepository
+from tests.helpers.status_engine import FakeStatusEngine
+
+METADATA = {
+    'model_id': '77',
+    'model_name': 'factory_model',
+    'workflow_name': 'import_model',
+    'schedule_name': '-',
+}
+WORKFLOW_ID = 'import_model-77-abc'
+RUN_ID = '0123456789abcdef0123456789abcdef'
+# What a tracking server with proxied artifacts reports as a run's artifact root, and therefore what
+# the version's `source` is built from. The run id sits at index 2 of this URI — the whole reason the
+# registration cannot store a literal `runs:/…` source.
+ARTIFACT_URI = f'mlflow-artifacts:/17/{RUN_ID}/artifacts'
+RESOLVED_SOURCE = f'{ARTIFACT_URI}/prediction_model'
+
+BUNDLE_PARAMETERS = {
+    'model_name': 'factory_model',
+    'target_variable': 'SE',
+    'input_variables': "['feature_a', 'feature_b']",
+    'retrain': 'False',
+}
+BUNDLE_METADATA = {
+    'model_name': 'factory_model',
+    'experiment_name': 'factory_experiment',
+    'run_id': 'aa11bb22cc33dd44ee55ff6677889900',
+    'model_version': '2',
+    'model_project': 'factory_project',
+    'export_timestamp': '2026-08-25T12:00:00+00:00',
+}
+
+# The whole vocabulary of promotion. None of it may be reached by this change.
+PROMOTION_CALLS = (
+    'transition_model_version_stage',
+    'set_registered_model_alias',
+    'set_model_version_tag',
+    'update_model_version',
+)
+
+
+@fixture
+def mlflow_client() -> MagicMock:
+    client = MagicMock()
+    client.create_run.return_value.info.run_id = RUN_ID
+    client.get_run.return_value.info.artifact_uri = ARTIFACT_URI
+    client.create_model_version.return_value.version = 4
+    return client
+
+
+@fixture
+def mongo() -> AsyncMock:
+    repository = AsyncMock()
+    repository.find_one.return_value = None
+    repository.distinct.return_value = []
+    return repository
+
+
+@fixture
+def import_activity(tmp_path: Path, mlflow_client: MagicMock, mongo: AsyncMock) -> ModelImport:
+    minio_repository = MagicMock()
+    mlflow_repository = MagicMock()
+    mlflow_repository.client = mlflow_client
+    with patch('laborious.activities.model_import.create_engine', return_value=FakeStatusEngine()):
+        activity = ModelImport(
+            status_config={
+                'host': 'h',
+                'port': 5432,
+                'dbname': 'sientia-core-mlops-bff',
+                'user': 'u',
+                'password': 'p',
+                'schema': 'public',
+            },
+            import_config={
+                'password_key': None,
+                'bucket': 'sientia',
+                'prefix': 'imported_models/',
+                'work_dir': str(tmp_path),
+                'max_object_bytes': 1024,
+                'max_entries': 10,
+                'max_uncompressed_bytes': 4096,
+                'max_compression_ratio': 200.0,
+                'models_collection': 'models',
+                'retention_days': 7,
+            },
+            minio_repository=minio_repository,
+            mlflow_repository=mlflow_repository,
+            mongo_repository=mongo,
+            logger=MagicMock(),
+            notification_handler=MagicMock(),
+            metrics_controller=AsyncMock(),
+        )
+    activity.emit_metric = AsyncMock()
+    activity.send_notification_async = AsyncMock()
+    return activity
+
+
+def base(**overrides) -> dict:
+    return {
+        'metadata': METADATA,
+        'import_run_id': 77,
+        'workflow_id': WORKFLOW_ID,
+        **overrides,
+    }
+
+
+# ---------------------------------------------------------------------- the experiment
+
+
+@mark.asyncio
+async def test_the_experiment_keeps_the_origin_s_name(import_activity: ModelImport):
+    repository = import_activity.import_mlflow_repository
+    repository.get_experiment.return_value = MagicMock(experiment_id='17')
+
+    result = await import_activity.create_import_experiment(
+        base(experiment_name='factory_experiment')
+    )
+
+    repository.get_experiment.assert_called_once_with(
+        'factory_experiment', create_if_not_exists=True
+    )
+    assert result == {'experiment_id': '17', 'experiment_name': 'factory_experiment'}
+
+
+@mark.asyncio
+async def test_a_missing_mlflow_repository_is_a_configuration_error(import_activity: ModelImport):
+    import_activity.import_mlflow_repository = None
+
+    with raises(ValueError, match='MLflow repository not initialized'):
+        await import_activity.create_import_experiment(base(experiment_name='e'))
+
+
+# ------------------------------------------------------------------- the run and its tree
+
+
+def upload_input(extracted_dir: Path, **overrides) -> dict:
+    return base(
+        experiment_id='17',
+        run_name='import-factory_model-v2',
+        extracted_dir=str(extracted_dir),
+        parameters=BUNDLE_PARAMETERS,
+        source=BUNDLE_METADATA,
+        **overrides,
+    )
+
+
+@mark.asyncio
+async def test_every_import_creates_a_new_run_with_the_recorded_name(
+    import_activity: ModelImport, mlflow_client: MagicMock, tmp_path: Path
+):
+    """The name is passed in, never recomputed, so the record and the run cannot diverge."""
+    result = await import_activity.upload_import_artifacts(upload_input(tmp_path / 'extract'))
+
+    mlflow_client.create_run.assert_called_once_with(
+        experiment_id='17', run_name='import-factory_model-v2'
+    )
+    assert result == {'run_id': RUN_ID, 'run_name': 'import-factory_model-v2'}
+
+
+@mark.asyncio
+async def test_the_artifact_tree_is_logged_from_the_bundle_s_artifacts_directory(
+    import_activity: ModelImport, mlflow_client: MagicMock, tmp_path: Path
+):
+    extracted = tmp_path / 'extract'
+
+    await import_activity.upload_import_artifacts(upload_input(extracted))
+
+    mlflow_client.log_artifacts.assert_called_once_with(RUN_ID, str(extracted / 'artifacts'))
+
+
+@mark.asyncio
+async def test_the_bundle_s_parameters_are_logged_verbatim(
+    import_activity: ModelImport, mlflow_client: MagicMock, tmp_path: Path
+):
+    await import_activity.upload_import_artifacts(upload_input(tmp_path / 'extract'))
+
+    logged = {
+        call.args[1]: call.args[2]
+        for call in mlflow_client.log_param.call_args_list
+        if not call.args[1].startswith('source.')
+    }
+    assert logged == BUNDLE_PARAMETERS
+    assert logged['target_variable'] == 'SE'
+
+
+@mark.asyncio
+async def test_the_origin_is_logged_as_source_parameters(
+    import_activity: ModelImport, mlflow_client: MagicMock, tmp_path: Path
+):
+    await import_activity.upload_import_artifacts(upload_input(tmp_path / 'extract'))
+
+    logged = {
+        call.args[1]: call.args[2]
+        for call in mlflow_client.log_param.call_args_list
+        if call.args[1].startswith('source.')
+    }
+    assert logged == {f'source.{key}': value for key, value in BUNDLE_METADATA.items()}
+    # A record of where the model came from, not a link: the origin run id is never dereferenced.
+    assert logged['source.run_id'] == BUNDLE_METADATA['run_id']
+    mlflow_client.get_run.assert_not_called()
+    mlflow_client.download_artifacts.assert_not_called()
+
+
+@mark.asyncio
+async def test_the_run_is_closed_when_its_tree_is_in(
+    import_activity: ModelImport, mlflow_client: MagicMock, tmp_path: Path
+):
+    await import_activity.upload_import_artifacts(upload_input(tmp_path / 'extract'))
+
+    mlflow_client.set_terminated.assert_called_once_with(RUN_ID, 'FINISHED')
+
+
+# ------------------------------------------------------------------------ registration
+
+
+@mark.asyncio
+async def test_the_version_source_is_the_shape_the_legacy_loader_parses(
+    import_activity: ModelImport, mlflow_client: MagicMock
+):
+    result = await import_activity.register_import_model_version(
+        base(model_name='factory_model', run_id=RUN_ID)
+    )
+
+    assert result == {'model_name': 'factory_model', 'version': '4', 'source': RESOLVED_SOURCE}
+    mlflow_client.create_model_version.assert_called_once_with(
+        name='factory_model', source=RESOLVED_SOURCE, run_id=RUN_ID
+    )
+    # The one property the legacy loader depends on, asserted directly.
+    assert RESOLVED_SOURCE.split('/')[2] == RUN_ID
+
+
+@mark.asyncio
+async def test_a_literal_runs_uri_would_not_be_parseable_by_the_legacy_loader():
+    """The trap this registration avoids: `create_model_version` stores `source` verbatim, and a
+    `runs:/` URI has only two slashes, so the loader's index 2 is the artifact path."""
+    assert f'runs:/{RUN_ID}/prediction_model'.split('/')[2] == 'prediction_model'
+    assert 'models:/m-9f1c'.split('/')[1:] == ['m-9f1c']
+
+
+@mark.asyncio
+async def test_a_registered_model_is_created_only_when_it_is_absent(
+    import_activity: ModelImport, mlflow_client: MagicMock
+):
+    mlflow_client.get_registered_model.side_effect = RuntimeError('not found')
+
+    await import_activity.register_import_model_version(base(model_name='new_model', run_id=RUN_ID))
+
+    mlflow_client.create_registered_model.assert_called_once_with('new_model')
+
+
+@mark.asyncio
+async def test_a_second_import_reuses_the_registered_model(
+    import_activity: ModelImport, mlflow_client: MagicMock
+):
+    mlflow_client.get_registered_model.return_value = MagicMock(name='factory_model')
+
+    await import_activity.register_import_model_version(
+        base(model_name='factory_model', run_id=RUN_ID)
+    )
+
+    mlflow_client.create_registered_model.assert_not_called()
+
+
+@mark.asyncio
+async def test_the_import_never_promotes_anything(
+    import_activity: ModelImport, mlflow_client: MagicMock, tmp_path: Path
+):
+    """A `COMPLETED` import is a registered model, not a servable one."""
+    await import_activity.create_import_experiment(base(experiment_name='factory_experiment'))
+    await import_activity.upload_import_artifacts(upload_input(tmp_path / 'extract'))
+    await import_activity.register_import_model_version(
+        base(model_name='factory_model', run_id=RUN_ID)
+    )
+
+    for call_name in PROMOTION_CALLS:
+        getattr(mlflow_client, call_name).assert_not_called()
+
+
+@mark.asyncio
+async def test_no_provisioning_call_passes_a_stage(
+    import_activity: ModelImport, mlflow_client: MagicMock, tmp_path: Path
+):
+    await import_activity.upload_import_artifacts(upload_input(tmp_path / 'extract'))
+    result = await import_activity.register_import_model_version(
+        base(model_name='factory_model', run_id=RUN_ID)
+    )
+
+    for call in mlflow_client.method_calls:
+        assert 'stage' not in call.kwargs
+        assert 'alias' not in call.kwargs
+    assert 'stage' not in result
+
+
+# ------------------------------------------------- the legacy loader reads what we wrote
+
+
+def legacy_repository() -> MLFlowRepository:
+    """A real `MLFlowRepository`, with the process-wide side effects of its constructor stubbed."""
+    with (
+        patch.object(mlflow_lib, 'set_tracking_uri'),
+        patch.object(mlflow_lib.tracking, 'MlflowClient'),
+    ):
+        return MLFlowRepository(
+            host='http://localhost:5000',
+            username='admin',
+            password='admin',
+            logger=MagicMock(),
+            notification_handler=MagicMock(),
+            metrics_controller=AsyncMock(),
+        )
+
+
+def registry_holding(source: str) -> MagicMock:
+    client = MagicMock()
+    client.search_registered_models.return_value = [MagicMock()]
+    version = MagicMock(version='4', source=source)
+    version.current_stage = 'Production'
+    client.search_model_versions.return_value = [version]
+    return client
+
+
+def test_the_legacy_loader_resolves_the_run_the_import_created():
+    """The contract the source shape exists for: the real production code, fed a version this
+    import would create, returns this import's run id."""
+    repository = legacy_repository()
+    repository.client = registry_holding(RESOLVED_SOURCE)
+
+    assert repository.get_model_run_id('factory_model') == RUN_ID
+
+
+def test_the_legacy_loader_would_read_a_runs_uri_as_the_artifact_path():
+    """Left as an executable warning: registering with the unresolved URI silently breaks loading,
+    because nothing raises — the loader just returns the wrong string."""
+    repository = legacy_repository()
+    repository.client = registry_holding(f'runs:/{RUN_ID}/prediction_model')
+
+    assert repository.get_model_run_id('factory_model') == 'prediction_model'
+
+
+def test_the_legacy_loader_cannot_read_a_logged_model_uri():
+    repository = legacy_repository()
+    repository.client = registry_holding('models:/m-9f1c')
+
+    with raises(mlflow_lib.exceptions.MlflowException, match='invalid source URI'):
+        repository.get_model_run_id('factory_model')
+
+
+# ------------------------------------------------------------------- the model document
+
+
+@mark.asyncio
+async def test_the_document_id_continues_the_collection_across_gaps(
+    import_activity: ModelImport, mongo: AsyncMock
+):
+    mongo.distinct.return_value = [str(value) for value in [*range(1, 19), *range(20, 25), 30]]
+
+    result = await import_activity.write_import_model_document(
+        base(model_name='factory_model', target='SE')
+    )
+
+    assert result['id'] == '31'
+
+
+@mark.asyncio
+async def test_the_document_id_is_numeric_not_lexicographic(
+    import_activity: ModelImport, mongo: AsyncMock
+):
+    mongo.distinct.return_value = ['9', '30']
+
+    result = await import_activity.write_import_model_document(
+        base(model_name='factory_model', target='SE')
+    )
+
+    assert result['id'] == '31'
+
+
+@mark.asyncio
+async def test_an_empty_collection_starts_at_one(import_activity: ModelImport, mongo: AsyncMock):
+    result = await import_activity.write_import_model_document(
+        base(model_name='factory_model', target='SE')
+    )
+
+    assert result['id'] == '1'
+
+
+@mark.asyncio
+async def test_the_document_is_exactly_the_contracted_shape(
+    import_activity: ModelImport, mongo: AsyncMock
+):
+    mongo.distinct.return_value = ['7']
+
+    await import_activity.write_import_model_document(base(model_name='factory_model', target='SE'))
+
+    collection, document, metadata = mongo.insert.await_args.args
+    assert collection == 'models'
+    assert metadata == METADATA
+    assert document == {
+        'id': '8',
+        'name': 'factory_model',
+        'active': False,
+        'model_config': {
+            'transform_flavor': 'sklearn',
+            'predict_flavor': 'joblib',
+            'retention_minutes': 60,
+            'target': 'SE',
+        },
+    }
+
+
+@mark.asyncio
+async def test_the_document_carries_no_field_the_runtime_does_not_read(
+    import_activity: ModelImport, mongo: AsyncMock
+):
+    await import_activity.write_import_model_document(base(model_name='factory_model', target='SE'))
+
+    _, document, _ = mongo.insert.await_args.args
+    for absent in ('is_compressed', 'retention_target', 'stage', 'source', 'project'):
+        assert absent not in document
+        assert absent not in document['model_config']
+
+
+@mark.asyncio
+async def test_a_name_already_in_the_collection_is_left_exactly_as_it_is(
+    import_activity: ModelImport, mongo: AsyncMock
+):
+    """`active`, `target` and the flavors of a model already on the platform are operator-owned."""
+    mongo.find_one.return_value = {'id': '12', 'name': 'factory_model', 'active': True}
+
+    result = await import_activity.write_import_model_document(
+        base(model_name='factory_model', target='SE')
+    )
+
+    assert result == {'written': False, 'skipped': True, 'id': '12'}
+    mongo.insert.assert_not_awaited()
+    mongo.distinct.assert_not_awaited()
+
+
+@mark.asyncio
+async def test_unusable_ids_are_ignored_and_counted(import_activity: ModelImport, mongo: AsyncMock):
+    mongo.distinct.return_value = ['1', 'legacy-model', None, '3']
+
+    result = await import_activity.write_import_model_document(
+        base(model_name='factory_model', target='SE')
+    )
+
+    assert result['id'] == '4'
+    assert result['ignored_ids'] == 2
+    assert any(
+        'ignored 2 non-numeric' in str(call)
+        for call in import_activity.logger.custom_info.call_args_list
+    )
+
+
+@mark.asyncio
+async def test_a_duplicate_id_fails_the_step_without_recomputing_it(
+    import_activity: ModelImport, mongo: AsyncMock
+):
+    """No retry and no second guess: the unique index says the collection moved under us."""
+    mongo.distinct.return_value = ['1']
+    mongo.insert.side_effect = RuntimeError('E11000 duplicate key error: id_1')
+
+    with raises(RuntimeError, match='duplicate key'):
+        await import_activity.write_import_model_document(
+            base(model_name='factory_model', target='SE')
+        )
+
+    assert mongo.distinct.await_count == 1
+    assert mongo.insert.await_count == 1
+
+
+@mark.asyncio
+async def test_the_import_only_reads_and_inserts_documents(
+    import_activity: ModelImport, mongo: AsyncMock
+):
+    """No update, no replace, no upsert: an import cannot change a model already on the platform."""
+    await import_activity.write_import_model_document(base(model_name='factory_model', target='SE'))
+
+    used = {name for name, _, _ in mongo.mock_calls}
+    assert used <= {'find_one', 'distinct', 'insert'}
+
+
+@mark.asyncio
+async def test_a_missing_mongo_repository_is_a_configuration_error(
+    import_activity: ModelImport,
+):
+    import_activity.mongo_repository = None
+
+    with raises(ValueError, match='MongoDB repository not initialized'):
+        await import_activity.write_import_model_document(base(model_name='m', target='SE'))
+
+
+def test_no_import_module_writes_an_active_document():
+    """Asserted from the source too, so `active: false` cannot be flipped by a later edit."""
+    source = Path('laborious/activities/model_import.py').read_text()
+
+    assert "'active': False" in source
+    assert "'active': True" not in source
+    for forbidden in ('update_one', 'replace_one', 'update_many', 'upsert'):
+        assert forbidden not in source
+
+
+# ------------------------------------------------------------------ cleanup and retention
+
+
+@mark.asyncio
+async def test_cleanup_removes_this_import_s_scratch_directory(
+    import_activity: ModelImport, tmp_path: Path
+):
+    work_dir = tmp_path / WORKFLOW_ID
+    (work_dir / 'extract' / 'artifacts').mkdir(parents=True)
+    (work_dir / 'bundle.sientia').write_bytes(b'downloaded')
+
+    result = await import_activity.cleanup_import_files(base())
+
+    assert result == {'cleaned': True, 'existed': True}
+    assert not work_dir.exists()
+
+
+@mark.asyncio
+async def test_cleanup_of_a_run_that_never_downloaded_anything_is_not_a_failure(
+    import_activity: ModelImport,
+):
+    result = await import_activity.cleanup_import_files(base())
+
+    assert result == {'cleaned': True, 'existed': False}
+
+
+@mark.asyncio
+async def test_cleanup_that_fails_is_reported_and_swallowed(
+    import_activity: ModelImport, tmp_path: Path
+):
+    """The user's model was imported; a leaked directory is a log line, not an ERROR."""
+    (tmp_path / WORKFLOW_ID).mkdir()
+
+    with patch.object(shutil, 'rmtree', side_effect=OSError('device or resource busy')):
+        result = await import_activity.cleanup_import_files(base())
+
+    assert result['cleaned'] is False
+    assert result['reason'] == 'OSError'
+    logged = import_activity.logger.custom_error.call_args_list[0].args[0]
+    assert 'IMPORT_CLEANUP_INCOMPLETE' in logged
+    import_activity.send_notification_async.assert_awaited_once()
+
+
+@mark.asyncio
+async def test_cleanup_never_touches_the_uploaded_object(
+    import_activity: ModelImport, tmp_path: Path
+):
+    (tmp_path / WORKFLOW_ID).mkdir()
+
+    await import_activity.cleanup_import_files(base())
+
+    client = import_activity.import_minio_repository.minio_client
+    client.remove_object.assert_not_called()
+    client.remove_objects.assert_not_called()
+
+
+@mark.asyncio
+async def test_the_expiry_rule_is_applied_to_the_import_prefix(import_activity: ModelImport):
+    client = import_activity.import_minio_repository.minio_client
+    client.get_bucket_lifecycle.return_value = None
+
+    result = await import_activity.ensure_import_bundle_retention(base())
+
+    assert result == {'bucket': 'sientia', 'prefix': 'imported_models/', 'days': 7, 'rules': 1}
+    config = client.set_bucket_lifecycle.call_args.args[1]
+    rule = config.rules[0]
+    assert rule.rule_id == 'sientia-import-bundle-expiry'
+    assert rule.expiration.days == 7
+    assert rule.rule_filter.prefix == 'imported_models/'
+
+
+@mark.asyncio
+async def test_applying_the_rule_twice_leaves_one_rule(import_activity: ModelImport):
+    client = import_activity.import_minio_repository.minio_client
+    client.get_bucket_lifecycle.return_value = None
+
+    await import_activity.ensure_import_bundle_retention(base())
+    applied = client.set_bucket_lifecycle.call_args.args[1]
+    client.get_bucket_lifecycle.return_value = applied
+    await import_activity.ensure_import_bundle_retention(base())
+
+    config = client.set_bucket_lifecycle.call_args.args[1]
+    assert [rule.rule_id for rule in config.rules] == ['sientia-import-bundle-expiry']
+
+
+@mark.asyncio
+async def test_a_rule_that_belongs_to_somebody_else_is_kept(import_activity: ModelImport):
+    client = import_activity.import_minio_repository.minio_client
+    existing = MagicMock()
+    existing.rules = [MagicMock(rule_id='predictions-offload-expiry')]
+    client.get_bucket_lifecycle.return_value = existing
+
+    await import_activity.ensure_import_bundle_retention(base())
+
+    config = client.set_bucket_lifecycle.call_args.args[1]
+    assert [rule.rule_id for rule in config.rules] == [
+        'predictions-offload-expiry',
+        'sientia-import-bundle-expiry',
+    ]
+
+
+@mark.asyncio
+async def test_a_bucket_with_no_lifecycle_configuration_at_all_is_handled(
+    import_activity: ModelImport,
+):
+    """Some deployments answer with an error, others with `None`; both mean "no rule yet"."""
+    client = import_activity.import_minio_repository.minio_client
+    client.get_bucket_lifecycle.side_effect = RuntimeError('NoSuchLifecycleConfiguration')
+
+    result = await import_activity.ensure_import_bundle_retention(base())
+
+    assert result['rules'] == 1
+
+
+@mark.asyncio
+async def test_the_retention_rule_can_be_applied_to_a_requested_bucket(
+    import_activity: ModelImport,
+):
+    client = import_activity.import_minio_repository.minio_client
+    client.get_bucket_lifecycle.return_value = None
+
+    result = await import_activity.ensure_import_bundle_retention(base(bucket='other-bucket'))
+
+    assert result['bucket'] == 'other-bucket'
+    assert client.set_bucket_lifecycle.call_args.args[0] == 'other-bucket'
+
+
+def test_the_model_document_shape_is_documented_in_one_place():
+    """A guard against the shape drifting between the code and what the runtime reads."""
+    source = Path('laborious/activities/model_import.py').read_text()
+
+    for key in ('transform_flavor', 'predict_flavor', 'retention_minutes', 'target'):
+        assert f"'{key}'" in source
+    assert json.dumps({'transform_flavor': 'sklearn'})  # sanity: the flavors are plain strings

tests/laborious/activities/test_model_import_status.py

@@ -0,0 +1,754 @@
+"""The import log row: the claim, the hints, the names and the one authoritative write.
+
+What these tests defend, in the order the workflow does it:
+
+- the **state mapping** is total — every `ImportStep` has a status, so a new step cannot be added
+  without deciding what the record says when it fails;
+- the **claim** is the only statement that checks ownership, and a claim that matches nothing writes
+  nothing anywhere and fails loudly everywhere else;
+- every **later** write keys on the id alone, never rewrites a column the inserting side owns, and
+  never drops the workflow identifier out of the JSON detail;
+- the **terminal** write is the single authority: one classification, `NULL` (never `''`) on success,
+  and a failure of its own that neither invents a failed import nor masks a real one.
+"""
+
+import json
+from datetime import datetime
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from pytest import fixture, mark, raises
+from sqlalchemy.exc import DataError, IntegrityError, OperationalError, ProgrammingError
+
+from laborious import metrics
+from laborious.activities.model_import import (
+    IMPORT_LOG_TABLE,
+    IMPORT_STATUS_HINT_RETRY_POLICY,
+    IMPORT_STATUS_RETRY_POLICY,
+    RUN_NAME_MAX_LENGTH,
+    ExperimentRunStatus,
+    ImportState,
+    ModelImport,
+    build_import_detail,
+    derive_run_name,
+    map_state_to_status,
+)
+from laborious.activities.model_import_errors import (
+    ImportErrorCode,
+    ImportLogRowNotClaimableError,
+    ImportStatusRejectedError,
+)
+from laborious.utils.bundle.steps import ImportStep
+from tests.helpers.status_engine import FakeStatusEngine
+
+STATUS_CONFIG = {
+    'host': 'bff-db.internal',
+    'port': 5432,
+    'dbname': 'sientia-core-mlops-bff',
+    'user': 'mlops_bff_user',
+    'password': 'bff-password',
+    'schema': 'public',
+}
+
+IMPORT_CONFIG = {
+    'password_key': None,
+    'bucket': 'sientia',
+    'prefix': 'imported_models/',
+    'work_dir': '/tmp/sientia-import',  # noqa: S108 - never written to in this module's tests
+    'max_object_bytes': 1024,
+    'max_entries': 10,
+    'max_uncompressed_bytes': 4096,
+    'max_compression_ratio': 200.0,
+    'models_collection': 'models',
+    'retention_days': 7,
+}
+
+METADATA = {
+    'model_id': '77',
+    'model_name': 'imported_model',
+    'workflow_name': 'import_model',
+    'schedule_name': '-',
+}
+
+# The columns the inserting side owns. No write in this module may ever assign one of them: the
+# record is shared with the training flow and with whoever created the row.
+BFF_OWNED_COLUMNS = frozenset(
+    {'username', 'request_data', 'bucket_name', 'file_name', 'run_type', 'created_at'}
+)
+
+# What the record's status must be for a failure at each step. Written out rather than derived, so a
+# new step forces a decision here instead of silently inheriting one.
+EXPECTED_ERROR_STATUS = {
+    ImportStep.RECEIVED: ExperimentRunStatus.ORCHESTRATOR_VALIDATION_ERROR,
+    ImportStep.DOWNLOAD: ExperimentRunStatus.ORCHESTRATOR_VALIDATION_ERROR,
+    ImportStep.DECRYPTION: ExperimentRunStatus.ORCHESTRATOR_VALIDATION_ERROR,
+    ImportStep.ARCHIVE_INSPECTION: ExperimentRunStatus.ORCHESTRATOR_VALIDATION_ERROR,
+    ImportStep.EXTRACTION: ExperimentRunStatus.ORCHESTRATOR_VALIDATION_ERROR,
+    ImportStep.STRUCTURE_VALIDATION: ExperimentRunStatus.ORCHESTRATOR_VALIDATION_ERROR,
+    ImportStep.CONTENT_POLICY: ExperimentRunStatus.ORCHESTRATOR_VALIDATION_ERROR,
+    ImportStep.EXPERIMENT_CREATION: ExperimentRunStatus.IMPORT_ERROR,
+    ImportStep.ARTIFACT_UPLOAD: ExperimentRunStatus.IMPORT_ERROR,
+    ImportStep.REGISTRATION: ExperimentRunStatus.IMPORT_ERROR,
+    ImportStep.MODEL_DOCUMENT: ExperimentRunStatus.IMPORT_ERROR,
+    ImportStep.CLEANUP: ExperimentRunStatus.IMPORT_ERROR,
+}
+
+
+def build_activity(engine: FakeStatusEngine) -> ModelImport:
+    """A `ModelImport` whose only connection is the recording engine double."""
+    with patch('laborious.activities.model_import.create_engine', return_value=engine):
+        activity = ModelImport(
+            status_config=STATUS_CONFIG,
+            import_config=IMPORT_CONFIG,
+            minio_repository=MagicMock(),
+            mlflow_repository=MagicMock(),
+            mongo_repository=AsyncMock(),
+            logger=MagicMock(),
+            notification_handler=MagicMock(),
+            metrics_controller=AsyncMock(),
+        )
+    activity.emit_metric = AsyncMock()
+    activity.send_notification_async = AsyncMock()
+    return activity
+
+
+@fixture
+def engine() -> FakeStatusEngine:
+    return FakeStatusEngine()
+
+
+@fixture
+def import_activity(engine: FakeStatusEngine) -> ModelImport:
+    return build_activity(engine)
+
+
+def detail_of(statement) -> dict:
+    """The `import` object a write put into `orchestrator_response_data`."""
+    return json.loads(statement.parameters['detail'])['import']
+
+
+def claim_input(**overrides) -> dict:
+    return {
+        'metadata': METADATA,
+        'import_run_id': 77,
+        'workflow_id': 'import_model-77-abc',
+        **overrides,
+    }
+
+
+# --------------------------------------------------------------------- the state mapping
+
+
+def test_running_is_the_orchestrator_holding_the_row():
+    assert map_state_to_status(ImportState.RUNNING) == ExperimentRunStatus.ORCHESTRATOR_WAITING_PROC
+    # The step must not change a running row's status: the row is running whatever it is doing.
+    for step in ImportStep:
+        assert (
+            map_state_to_status(ImportState.RUNNING, step)
+            == ExperimentRunStatus.ORCHESTRATOR_WAITING_PROC
+        )
+
+
+def test_completed_is_import_success():
+    assert map_state_to_status(ImportState.COMPLETED) == ExperimentRunStatus.IMPORT_SUCCESS
+    assert (
+        map_state_to_status(ImportState.COMPLETED, ImportStep.MODEL_DOCUMENT)
+        == ExperimentRunStatus.IMPORT_SUCCESS
+    )
+
+
+@mark.parametrize('step', list(ImportStep))
+def test_every_step_has_an_error_status(step: ImportStep):
+    assert map_state_to_status(ImportState.ERROR, step) == EXPECTED_ERROR_STATUS[step]
+    # The value form is what the workflow actually passes across the activity boundary.
+    assert map_state_to_status(ImportState.ERROR, step.value) == EXPECTED_ERROR_STATUS[step]
+
+
+def test_the_error_mapping_covers_the_whole_enum():
+    """A step added without a decision here fails this test rather than defaulting quietly."""
+    assert set(EXPECTED_ERROR_STATUS) == set(ImportStep)
+
+
+def test_an_unclassified_error_is_a_platform_error():
+    assert map_state_to_status(ImportState.ERROR) == ExperimentRunStatus.IMPORT_ERROR
+
+
+# ------------------------------------------------------------------------- the run name
+
+
+def test_run_name_has_the_documented_shape():
+    assert derive_run_name('sales_forecast', '3') == 'import-sales_forecast-v3'
+
+
+def test_run_name_truncates_inside_the_name_segment():
+    name = derive_run_name('a' * 120, '12')
+    assert name.startswith('import-')
+    assert name.endswith('-v12')
+    assert len(name) == RUN_NAME_MAX_LENGTH
+
+
+def test_run_name_survives_a_version_that_leaves_no_room():
+    name = derive_run_name('model', 'v' * 80)
+    assert name.startswith('import-')
+    assert len(name) <= RUN_NAME_MAX_LENGTH
+
+
+def test_run_name_is_deterministic():
+    assert derive_run_name('model', '4') == derive_run_name('model', '4')
+
+
+def test_run_name_is_never_shorter_than_the_column_floor():
+    # `error_message` is not the only column with a length CHECK; the prefix is what guarantees it.
+    assert len(derive_run_name('', '')) >= 3
+
+
+# --------------------------------------------------------------------- the JSON detail
+
+
+def test_the_detail_always_carries_the_workflow_id():
+    detail = json.loads(build_import_detail(workflow_id='wf-1'))['import']
+    assert detail['workflow_id'] == 'wf-1'
+
+
+def test_the_detail_omits_a_code_when_there_is_none():
+    detail = json.loads(build_import_detail(workflow_id='wf-1', step=ImportStep.RECEIVED))['import']
+    assert 'code' not in detail
+    assert detail['step'] == 'received'
+
+
+def test_the_detail_records_a_cleanup_failure_without_a_code():
+    detail = json.loads(build_import_detail(workflow_id='wf-1', cleanup_failed=True))['import']
+    assert detail['cleanup_failed'] is True
+    assert 'code' not in detail
+
+
+# ------------------------------------------------------------------------------ the claim
+
+
+@mark.asyncio
+async def test_claim_writes_the_running_status_and_the_workflow_id(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    engine.rows = [{'run_type': 'IMPORT', 'bucket_name': 'sientia', 'file_name': None}]
+
+    result = await import_activity.claim_import_status(claim_input())
+
+    assert result['claimed'] is True
+    write = engine.writes[0]
+    assert write.table == IMPORT_LOG_TABLE
+    assert write.parameters['status'] == ExperimentRunStatus.ORCHESTRATOR_WAITING_PROC.value
+    assert detail_of(write) == {
+        'workflow_id': 'import_model-77-abc',
+        'gate': None,
+        'step': 'received',
+    }
+
+
+@mark.asyncio
+async def test_claim_is_the_only_statement_that_guards_ownership(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    engine.rows = [{'run_type': 'IMPORT'}]
+
+    await import_activity.claim_import_status(claim_input())
+
+    sql = engine.writes[0].sql
+    assert "run_type = 'IMPORT'" in sql
+    assert "status = 'PENDING'" in sql
+    # The idempotency disjunct: a re-execution of the same workflow re-claims its own row instead of
+    # colliding with itself. It reads the identifier out of the JSON detail because the change adds
+    # no column for it.
+    assert "orchestrator_response_data -> 'import' ->> 'workflow_id' = :workflow_id" in sql
+
+
+@mark.asyncio
+async def test_claim_returns_the_values_the_input_guards_are_checked_against(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    engine.rows = [
+        {
+            'run_type': 'IMPORT',
+            'bucket_name': 'sientia',
+            'file_name': 'imported_models/model.sientia',
+            'request_data': {'expected_digest': 'AB' * 32},
+        }
+    ]
+
+    result = await import_activity.claim_import_status(claim_input())
+
+    assert result['bucket_name'] == 'sientia'
+    assert result['file_name'] == 'imported_models/model.sientia'
+    assert result['request_digest'] == 'ab' * 32
+
+
+@mark.asyncio
+@mark.parametrize('key', ['expected_digest', 'digest', 'sha256', 'file_digest'])
+async def test_claim_finds_the_digest_wherever_the_bff_recorded_it(key: str):
+    engine = FakeStatusEngine(rows=[{'run_type': 'IMPORT', 'request_data': {key: 'f' * 64}}])
+    activity = build_activity(engine)
+
+    result = await activity.claim_import_status(claim_input())
+
+    assert result['request_digest'] == 'f' * 64
+
+
+@mark.asyncio
+@mark.parametrize('request_data', [None, 'not-a-dict', {}, {'expected_digest': ''}])
+async def test_claim_treats_an_absent_digest_as_absent(request_data):
+    engine = FakeStatusEngine(rows=[{'run_type': 'IMPORT', 'request_data': request_data}])
+    activity = build_activity(engine)
+
+    result = await activity.claim_import_status(claim_input())
+
+    assert result['request_digest'] is None
+
+
+@mark.asyncio
+async def test_claim_reclaimed_by_the_same_workflow_succeeds(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    engine.rows = [
+        {'run_type': 'IMPORT', 'status': 'ORCHESTRATOR_WAITING_PROC', 'claimed_by': 'wf-mine'}
+    ]
+
+    result = await import_activity.claim_import_status(claim_input(workflow_id='wf-mine'))
+
+    assert result['claimed'] is True
+
+
+@mark.asyncio
+async def test_claim_opens_one_transaction_and_leaves_none_open(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    engine.rows = [{'run_type': 'IMPORT'}]
+
+    await import_activity.claim_import_status(claim_input())
+
+    assert engine.transactions == 1
+    assert engine.open_scopes == 0
+
+
+@mark.asyncio
+async def test_the_claim_writes_before_it_reads(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    """A read-then-write claim would race two workflows into the same row; this one cannot."""
+    engine.rows = [{'run_type': 'IMPORT'}]
+
+    await import_activity.claim_import_status(claim_input())
+
+    assert engine.statements[0].is_write is True
+    assert engine.statements[1].verb == 'SELECT'
+
+
+@mark.asyncio
+@mark.parametrize(
+    ('row', 'expected'),
+    [
+        (None, 'no such import log row'),
+        ({'run_type': 'TRAIN', 'status': 'PENDING'}, 'is a TRAIN run'),
+        (
+            {
+                'run_type': 'IMPORT',
+                'status': 'ORCHESTRATOR_WAITING_PROC',
+                'claimed_by': 'another-workflow',
+            },
+            'already claimed by workflow another-workflow',
+        ),
+        ({'run_type': 'IMPORT', 'status': 'IMPORT_SUCCESS'}, 'in status IMPORT_SUCCESS'),
+    ],
+)
+async def test_an_unclaimable_row_names_its_cause(row, expected: str):
+    engine = FakeStatusEngine(rowcounts=[0], rows=[row])
+    activity = build_activity(engine)
+
+    with raises(ImportLogRowNotClaimableError) as failure:
+        await activity.claim_import_status(claim_input())
+
+    assert expected in str(failure.value)
+    assert failure.value.import_run_id == 77
+
+
+@mark.asyncio
+async def test_an_unclaimable_row_is_not_retried():
+    engine = FakeStatusEngine(rowcounts=[0], rows=[None])
+    activity = build_activity(engine)
+
+    with raises(ImportLogRowNotClaimableError) as failure:
+        await activity.claim_import_status(claim_input())
+
+    assert failure.value.non_retryable is True
+    assert failure.value.type == 'ImportLogRowNotClaimableError'
+    # And the policy that governs the claim agrees, so the ten attempts are not spent on a verdict
+    # that cannot change.
+    assert 'ImportLogRowNotClaimableError' in IMPORT_STATUS_RETRY_POLICY.non_retryable_error_types
+    assert (
+        'ImportLogRowNotClaimableError' in IMPORT_STATUS_HINT_RETRY_POLICY.non_retryable_error_types
+    )
+
+
+@mark.asyncio
+async def test_an_unclaimable_row_is_never_written_to():
+    """The only statement executed is the guarded claim that matched nothing, and no other write."""
+    engine = FakeStatusEngine(rowcounts=[0], rows=[{'run_type': 'TRAIN'}])
+    activity = build_activity(engine)
+
+    with raises(ImportLogRowNotClaimableError):
+        await activity.claim_import_status(claim_input())
+
+    assert len(engine.writes) == 1
+    assert engine.writes[0].sql == engine.statements[0].sql
+    assert "run_type = 'IMPORT'" in engine.writes[0].sql
+    assert [statement.verb for statement in engine.statements[1:]] == ['SELECT']
+    assert engine.tables == {IMPORT_LOG_TABLE}
+
+
+@mark.asyncio
+async def test_an_unclaimable_row_is_reported_through_every_sink():
+    engine = FakeStatusEngine(rowcounts=[0], rows=[None])
+    activity = build_activity(engine)
+
+    with raises(ImportLogRowNotClaimableError):
+        await activity.claim_import_status(claim_input())
+
+    logged = activity.logger.custom_error.call_args_list[0].args[0]
+    assert ImportErrorCode.LOG_ROW_NOT_CLAIMABLE.value in logged
+    assert 'not_claimable' in logged
+    metric = activity.emit_metric.await_args.kwargs
+    assert metric['metric_object'] is metrics.MODEL_IMPORT_ERROR_COUNT
+    assert metric['tags']['operation_type'] == ImportErrorCode.LOG_ROW_NOT_CLAIMABLE.value
+    notification = activity.send_notification_async.await_args.kwargs
+    assert notification['notification_id'] == ImportErrorCode.LOG_ROW_NOT_CLAIMABLE.value
+    assert notification['level'] == 'ERROR'
+
+
+@mark.asyncio
+async def test_a_claimed_row_counts_as_an_import_started(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    engine.rows = [{'run_type': 'IMPORT'}]
+
+    await import_activity.claim_import_status(claim_input())
+
+    metric = import_activity.emit_metric.await_args.kwargs
+    assert metric['metric_object'] is metrics.MODEL_IMPORT_STARTED_COUNT
+
+
+# ------------------------------------------------------------------- progress and names
+
+
+@mark.asyncio
+async def test_a_progress_write_keys_on_the_id_alone(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    await import_activity.record_import_status({**claim_input(), 'step': ImportStep.DOWNLOAD.value})
+
+    write = engine.writes[0]
+    assert 'WHERE id = :import_run_id' in write.sql
+    assert 'run_type' not in write.sql
+    assert write.parameters['status'] == ExperimentRunStatus.ORCHESTRATOR_WAITING_PROC.value
+    assert detail_of(write)['step'] == 'download'
+
+
+@mark.asyncio
+async def test_a_lost_progress_write_is_not_a_failure(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    engine.rowcounts = [0]
+
+    result = await import_activity.record_import_status(
+        {**claim_input(), 'step': ImportStep.DOWNLOAD.value}
+    )
+
+    assert result == {'recorded': False, 'rows': 0}
+
+
+@mark.asyncio
+async def test_the_names_write_touches_only_the_two_name_columns(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    await import_activity.record_import_names(
+        {
+            **claim_input(),
+            'experiment_name': 'sales_forecast',
+            'run_name': 'import-sales_forecast-v3',
+        }
+    )
+
+    write = engine.writes[0]
+    assert write.assignments == {
+        'experiment_name',
+        'run_name',
+        'updated_at',
+        'orchestrator_response_data',
+    }
+    # Deliberately no `status`: the names are written mid-flight and must not disturb the verdict.
+    assert 'status' not in write.assignments
+    assert write.parameters['experiment_name'] == 'sales_forecast'
+    assert write.parameters['run_name'] == 'import-sales_forecast-v3'
+
+
+@mark.asyncio
+async def test_the_names_write_preserves_the_workflow_id(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    await import_activity.record_import_names(
+        {**claim_input(), 'experiment_name': 'e', 'run_name': 'import-e-v1'}
+    )
+
+    assert detail_of(engine.writes[0])['workflow_id'] == 'import_model-77-abc'
+
+
+# ------------------------------------------------------------------- the terminal write
+
+
+@mark.asyncio
+async def test_a_successful_import_writes_a_null_reason_and_no_code(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    verdict = await import_activity.record_import_terminal_status(
+        {**claim_input(), 'succeeded': True, 'step': ImportStep.MODEL_DOCUMENT.value}
+    )
+
+    write = engine.writes[0]
+    assert write.parameters['status'] == ExperimentRunStatus.IMPORT_SUCCESS.value
+    # `''` would violate the column's `LENGTH(error_message) >= 3` CHECK.
+    assert write.parameters['error_message'] is None
+    assert 'code' not in detail_of(write)
+    assert verdict['status'] == ExperimentRunStatus.IMPORT_SUCCESS.value
+    assert verdict['code'] is None
+
+
+@mark.asyncio
+@mark.parametrize(
+    ('step', 'status', 'code'),
+    [
+        (
+            ImportStep.ARCHIVE_INSPECTION,
+            ExperimentRunStatus.ORCHESTRATOR_VALIDATION_ERROR,
+            ImportErrorCode.BUNDLE_REJECTED,
+        ),
+        (
+            ImportStep.STRUCTURE_VALIDATION,
+            ExperimentRunStatus.ORCHESTRATOR_VALIDATION_ERROR,
+            ImportErrorCode.BUNDLE_INCOMPLETE,
+        ),
+        (
+            ImportStep.REGISTRATION,
+            ExperimentRunStatus.IMPORT_ERROR,
+            ImportErrorCode.MODEL_NOT_REGISTERED,
+        ),
+        (
+            ImportStep.MODEL_DOCUMENT,
+            ExperimentRunStatus.IMPORT_ERROR,
+            ImportErrorCode.MODEL_SETTINGS_NOT_SAVED,
+        ),
+    ],
+)
+async def test_a_failure_is_recorded_with_its_step_code_and_reason(
+    step: ImportStep, status: ExperimentRunStatus, code: ImportErrorCode
+):
+    engine = FakeStatusEngine()
+    activity = build_activity(engine)
+
+    verdict = await activity.record_import_terminal_status(
+        {**claim_input(), 'succeeded': False, 'step': step.value, 'gate': 4}
+    )
+
+    write = engine.writes[0]
+    assert write.parameters['status'] == status.value
+    assert write.parameters['error_message'] == verdict['reason']
+    detail = detail_of(write)
+    assert detail['step'] == step.value
+    assert detail['code'] == code.value
+    assert detail['gate'] == 4
+
+
+@mark.asyncio
+async def test_a_cleanup_failure_does_not_turn_a_success_into_an_error(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    verdict = await import_activity.record_import_terminal_status(
+        {**claim_input(), 'succeeded': True, 'cleanup_failed': True}
+    )
+
+    write = engine.writes[0]
+    assert write.parameters['status'] == ExperimentRunStatus.IMPORT_SUCCESS.value
+    assert write.parameters['error_message'] is None
+    assert detail_of(write)['cleanup_failed'] is True
+    assert verdict['code'] is None
+
+
+@mark.asyncio
+async def test_the_terminal_write_returns_the_verdict_it_wrote(
+    import_activity: ModelImport,
+):
+    verdict = await import_activity.record_import_terminal_status(
+        {**claim_input(), 'succeeded': False, 'step': ImportStep.DECRYPTION.value}
+    )
+
+    assert verdict == {
+        'status': ExperimentRunStatus.ORCHESTRATOR_VALIDATION_ERROR.value,
+        'code': ImportErrorCode.BUNDLE_CANNOT_BE_OPENED.value,
+        'reason': verdict['reason'],
+        'step': ImportStep.DECRYPTION.value,
+    }
+    assert 'password' in verdict['reason']
+
+
+@mark.asyncio
+async def test_executing_the_terminal_write_twice_leaves_the_row_the_same(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    """What retries must be idempotent: two executions of the same verdict, one row state."""
+    payload = {**claim_input(), 'succeeded': True, 'step': ImportStep.MODEL_DOCUMENT.value}
+
+    await import_activity.record_import_terminal_status(payload)
+    await import_activity.record_import_terminal_status(payload)
+
+    first, second = engine.writes
+    assert first.sql == second.sql
+    assert {k: v for k, v in first.parameters.items() if k != 'updated_at'} == {
+        k: v for k, v in second.parameters.items() if k != 'updated_at'
+    }
+
+
+@mark.asyncio
+async def test_every_write_advances_updated_at(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    stamps = [datetime(2026, 8, 25, 12, 0, 0), datetime(2026, 8, 25, 12, 0, 5)]
+    with patch('laborious.activities.model_import.now', side_effect=stamps):
+        await import_activity.record_import_status(
+            {**claim_input(), 'step': ImportStep.DOWNLOAD.value}
+        )
+        await import_activity.record_import_terminal_status(
+            {**claim_input(), 'succeeded': True, 'step': ImportStep.MODEL_DOCUMENT.value}
+        )
+
+    assert engine.writes[0].parameters['updated_at'] < engine.writes[1].parameters['updated_at']
+
+
+@mark.asyncio
+async def test_no_status_write_touches_a_column_the_inserting_side_owns(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    engine.rows = [{'run_type': 'IMPORT'}]
+
+    await import_activity.claim_import_status(claim_input())
+    await import_activity.record_import_status({**claim_input(), 'step': ImportStep.DOWNLOAD.value})
+    await import_activity.record_import_names(
+        {**claim_input(), 'experiment_name': 'e', 'run_name': 'import-e-v1'}
+    )
+    await import_activity.record_import_terminal_status({**claim_input(), 'succeeded': True})
+
+    assert len(engine.writes) == 4
+    for write in engine.writes:
+        assert write.assignments & BFF_OWNED_COLUMNS == set()
+        assert write.table == IMPORT_LOG_TABLE
+    # One transaction per write: no status write shares a transaction with anything else.
+    assert engine.transactions == 4
+
+
+@mark.asyncio
+async def test_every_post_claim_write_keys_on_the_id_alone(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    """Ownership was settled by the claim. Re-checking it later can only lose a real verdict."""
+    await import_activity.record_import_status({**claim_input(), 'step': ImportStep.DOWNLOAD.value})
+    await import_activity.record_import_names(
+        {**claim_input(), 'experiment_name': 'e', 'run_name': 'import-e-v1'}
+    )
+    await import_activity.record_import_terminal_status({**claim_input(), 'succeeded': True})
+
+    for write in engine.writes:
+        where = write.sql[write.sql.index('WHERE') :]
+        assert where == 'WHERE id = :import_run_id'
+
+
+@mark.asyncio
+async def test_a_row_edited_by_hand_after_the_claim_still_receives_the_verdict(
+    import_activity: ModelImport, engine: FakeStatusEngine
+):
+    """The predicate carries no status, so an operator's edit cannot orphan the record."""
+    engine.rowcounts = [1]
+
+    verdict = await import_activity.record_import_terminal_status(
+        {**claim_input(), 'succeeded': False, 'step': ImportStep.REGISTRATION.value}
+    )
+
+    assert verdict['status'] == ExperimentRunStatus.IMPORT_ERROR.value
+    assert 'status' not in engine.writes[0].sql[engine.writes[0].sql.index('WHERE') :]
+
+
+@mark.asyncio
+@mark.parametrize('error_class', [IntegrityError, ProgrammingError, DataError])
+async def test_a_refused_statement_fails_on_the_first_attempt(error_class):
+    """A missing `V12` is this: the connection worked and the statement was refused."""
+    engine = FakeStatusEngine(
+        write_errors=[error_class('UPDATE ...', {}, Exception('refused'))],
+    )
+    activity = build_activity(engine)
+
+    with raises(ImportStatusRejectedError) as failure:
+        await activity.record_import_terminal_status({**claim_input(), 'succeeded': True})
+
+    assert failure.value.non_retryable is True
+    assert failure.value.type == 'ImportStatusRejectedError'
+    assert 'ImportStatusRejectedError' in IMPORT_STATUS_RETRY_POLICY.non_retryable_error_types
+
+
+@mark.asyncio
+async def test_a_connection_failure_is_left_to_the_retry():
+    """Repeating it is exactly what fixes it, so it must not be swallowed or reclassified."""
+    engine = FakeStatusEngine(write_errors=[OperationalError('UPDATE ...', {}, OSError('down'))])
+    activity = build_activity(engine)
+
+    with raises(OperationalError):
+        await activity.record_import_terminal_status({**claim_input(), 'succeeded': True})
+
+
+@mark.asyncio
+async def test_a_status_write_that_fails_then_succeeds_records_the_right_verdict():
+    engine = FakeStatusEngine(
+        write_errors=[OperationalError('UPDATE ...', {}, OSError('down')), None]
+    )
+    activity = build_activity(engine)
+    payload = {**claim_input(), 'succeeded': False, 'step': ImportStep.REGISTRATION.value}
+
+    with raises(OperationalError):
+        await activity.record_import_terminal_status(payload)
+    verdict = await activity.record_import_terminal_status(payload)
+
+    assert verdict['status'] == ExperimentRunStatus.IMPORT_ERROR.value
+    assert verdict['code'] == ImportErrorCode.MODEL_NOT_REGISTERED.value
+    assert engine.writes[-1].parameters['status'] == ExperimentRunStatus.IMPORT_ERROR.value
+
+
+@mark.asyncio
+async def test_a_verdict_that_cannot_be_written_is_reported_with_the_verdict():
+    engine = FakeStatusEngine()
+    activity = build_activity(engine)
+
+    result = await activity.report_import_status_write_failure(
+        {
+            **claim_input(),
+            'detail': 'OperationalError: connection refused',
+            'verdict': 'succeeded=True step=model_document gate=None',
+        }
+    )
+
+    assert result == {'reported': True}
+    # The one failure with no row to land on: nothing is written anywhere.
+    assert engine.statements == []
+    logged = activity.logger.custom_error.call_args_list[0].args[0]
+    assert ImportErrorCode.STATUS_NOT_RECORDED.value in logged
+    assert 'status_write' in logged
+    assert 'verdict_not_recorded=succeeded=True step=model_document' in logged
+    assert 'connection refused' in logged
+
+
+@mark.asyncio
+async def test_the_engine_is_disposed_on_shutdown(import_activity: ModelImport, engine):
+    import_activity.close()
+
+    assert engine.disposed is True

tests/laborious/test_import_absence.py

@@ -0,0 +1,264 @@
+"""What the import deliberately does **not** do.
+
+Every absence here was a decision, and every one of them is the kind of thing a later well-meaning
+edit adds back: a rollback for a failed import, a resume for a half-finished one, a signal to retry,
+a `joblib.load` to "check the model is valid", a reason string built from the exception. None of
+those are missing features — a failed import stays failed, and trying again is a new upload — so the
+tests are written against the source itself, because there is no behaviour to observe when the
+behaviour is the point of not existing.
+"""
+
+import ast
+from pathlib import Path
+
+from pytest import mark
+
+from laborious.workflows.import_model import ImportModel
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
+
+WORKFLOW_MODULE = REPOSITORY_ROOT / 'laborious/workflows/import_model.py'
+ACTIVITIES_MODULE = REPOSITORY_ROOT / 'laborious/activities/model_import.py'
+ERRORS_MODULE = REPOSITORY_ROOT / 'laborious/activities/model_import_errors.py'
+READER_MODULE = REPOSITORY_ROOT / 'laborious/utils/bundle/reader.py'
+FORMAT_MODULE = REPOSITORY_ROOT / 'laborious/utils/bundle/format.py'
+
+IMPORT_PATH_MODULES = (
+    WORKFLOW_MODULE,
+    ACTIVITIES_MODULE,
+    ERRORS_MODULE,
+    READER_MODULE,
+    FORMAT_MODULE,
+)
+
+# The vocabulary of recovery. A partial import is not repaired; it is reported and abandoned.
+RECOVERY_WORDS = ('rollback', 'compensate', 'resume', 'reconcile', 'undo')
+
+# Loading a model is the *execution* side, owned by another card. The importer moves bytes and
+# writes metadata; it never unpickles what it just accepted from a user.
+MODEL_LOADING_CALLS = (
+    'mlflow.pyfunc.load_model',
+    'mlflow.sklearn.load_model',
+    'joblib.load',
+    'pickle.load',
+    'pickle.loads',
+)
+
+# Decryption lives in the activity that uses the password, never in the workflow, which would put
+# the plaintext into the event history.
+DECRYPTION_PRIMITIVES = (
+    'AESGCM',
+    'crypto_pwhash',
+    'crypto_secretstream',
+    'decrypt_password_envelope',
+    'load_envelope_key',
+    'IMPORT_PASSWORD_KEY',
+    'Argon2',
+)
+
+
+def parse(path: Path) -> ast.Module:
+    return ast.parse(path.read_text())
+
+
+def called_names(tree: ast.Module) -> list[str]:
+    """Every called name, as written: `a.b.c(...)` becomes `a.b.c`."""
+    names = []
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call):
+            continue
+        names.append(ast.unparse(node.func))
+    return names
+
+
+def defined_names(tree: ast.Module) -> list[str]:
+    return [
+        node.name
+        for node in ast.walk(tree)
+        if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef)
+    ]
+
+
+# ------------------------------------------------------------- no retry beyond the record
+
+
+def test_the_workflow_holds_no_recovery_routine():
+    tree = parse(WORKFLOW_MODULE)
+
+    for name in defined_names(tree) + called_names(tree):
+        for word in RECOVERY_WORDS:
+            assert word not in name.lower(), f'{name} looks like recovery'
+
+
+def test_the_activities_hold_no_recovery_routine():
+    tree = parse(ACTIVITIES_MODULE)
+
+    for name in defined_names(tree) + called_names(tree):
+        for word in RECOVERY_WORDS:
+            assert word not in name.lower(), f'{name} looks like recovery'
+
+
+def test_the_workflow_never_continues_as_new():
+    """A single import is one workflow execution, start to finish."""
+    assert not [name for name in called_names(parse(WORKFLOW_MODULE)) if 'continue_as_new' in name]
+
+
+# ------------------------------------------------------------------ no handlers, no waiting
+
+
+def test_the_workflow_declares_no_signal_or_query_handler():
+    """Nothing outside can steer an import once it starts: no retry signal, no progress query."""
+    for name in dir(ImportModel):
+        member = getattr(ImportModel, name, None)
+        for marker in (
+            '__temporal_signal_definition',
+            '__temporal_query_definition',
+            '__temporal_update_definition',
+        ):
+            assert not hasattr(member, marker), f'{name} declares {marker}'
+
+
+def test_the_workflow_module_carries_no_handler_decorator():
+    tree = parse(WORKFLOW_MODULE)
+    decorators = [
+        ast.unparse(decorator)
+        for node in ast.walk(tree)
+        if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef)
+        for decorator in node.decorator_list
+    ]
+
+    assert set(decorators) == {'workflow.run', 'staticmethod'}
+    assert decorators.count('workflow.run') == 1
+
+
+def test_the_workflow_never_waits_for_a_condition():
+    """No `wait_condition`: there is no external event an import waits on."""
+    assert not [name for name in called_names(parse(WORKFLOW_MODULE)) if 'wait_condition' in name]
+
+
+# --------------------------------------------------------------------- nothing is loaded
+
+
+@mark.parametrize('module_path', IMPORT_PATH_MODULES, ids=lambda path: path.name)
+def test_no_module_in_the_import_path_loads_a_model(module_path: Path):
+    source = module_path.read_text()
+
+    for call in MODEL_LOADING_CALLS:
+        assert call not in source
+
+
+def test_the_reader_knows_nothing_about_mlflow():
+    """It stays liftable into a shared package: the reader is about the file, not the platform."""
+    source = READER_MODULE.read_text()
+
+    assert 'mlflow' not in source
+    assert 'temporalio' not in source
+
+
+# ----------------------------------------------------------- the workflow holds no secret
+
+
+@mark.parametrize('primitive', DECRYPTION_PRIMITIVES)
+def test_the_workflow_module_holds_no_decryption_primitive(primitive: str):
+    assert primitive not in WORKFLOW_MODULE.read_text()
+
+
+def test_the_workflow_only_ever_holds_the_envelope():
+    """The envelope crosses the workflow; the password it wraps never does."""
+    source = WORKFLOW_MODULE.read_text()
+
+    assert 'password_envelope' in source
+    # The only other mention of a password is the input field the workflow refuses to accept.
+    assert source.count("'password'") == 1
+
+
+def test_the_plaintext_password_field_is_refused_by_name():
+    from laborious.workflows.import_model import REJECTED_INPUT_FIELDS
+
+    assert 'password' in REJECTED_INPUT_FIELDS
+
+
+# ------------------------------------------------------- every reason comes from the catalog
+
+
+def _error_message_values(tree: ast.Module) -> list[ast.expr]:
+    """Every expression written into the record's `error_message`."""
+    values = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Dict):
+            for key, value in zip(node.keys, node.values, strict=True):
+                if isinstance(key, ast.Constant) and key.value == 'error_message':
+                    values.append(value)
+        if isinstance(node, ast.keyword) and node.arg == 'error_message':
+            values.append(node.value)
+    return values
+
+
+def test_no_recorded_reason_is_built_from_an_exception():
+    """The catalog is the only source of a sentence a user reads."""
+    values = _error_message_values(parse(ACTIVITIES_MODULE))
+
+    assert values, 'no error_message write found — has the terminal statement moved?'
+    for value in values:
+        assert isinstance(value, ast.Name | ast.Constant), ast.unparse(value)
+        assert not isinstance(value, ast.JoinedStr | ast.BinOp)
+
+
+def test_the_reason_written_is_the_classified_one():
+    values = _error_message_values(parse(ACTIVITIES_MODULE))
+
+    assert {ast.unparse(value) for value in values} == {'reason'}
+
+
+def test_the_classification_ignores_the_exception_by_construction():
+    """`classify_import_failure` deletes its exception argument; nothing can leak from it."""
+    source = ERRORS_MODULE.read_text()
+
+    assert 'del exception' in source
+
+
+@mark.parametrize('module_path', (WORKFLOW_MODULE, ACTIVITIES_MODULE), ids=lambda p: p.name)
+def test_no_status_detail_interpolates_an_exception(module_path: Path):
+    """The JSON detail is built from a code and a step, both of them enum values."""
+    tree = parse(module_path)
+
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call) or ast.unparse(node.func) != 'build_import_detail':
+            continue
+        for keyword in node.keywords:
+            rendered = ast.unparse(keyword.value)
+            assert 'error' not in rendered
+            assert 'exception' not in rendered
+            assert not isinstance(keyword.value, ast.JoinedStr)
+
+
+# ------------------------------------------------------------------ no relational target
+
+
+def test_the_import_writes_to_one_table_and_names_it_once():
+    """D22: no project table, no model table, no second database. One `UPDATE`, one relation."""
+    statements = []
+    for node in ast.walk(parse(ACTIVITIES_MODULE)):
+        if not isinstance(node, ast.Call) or ast.unparse(node.func) != 'text':
+            continue
+        argument = node.args[0]
+        if isinstance(argument, ast.Constant):
+            statements.append(argument.value)
+
+    assert statements, 'no SQL statement found in the import activities'
+    for statement in statements:
+        words = statement.split()
+        relations = {
+            words[index + 1]
+            for index, word in enumerate(words)
+            if word.upper() in ('UPDATE', 'FROM', 'INTO', 'JOIN')
+        }
+        assert relations == {'public.experiment_run'}
+
+
+def test_the_import_never_inserts_a_relational_row():
+    source = ACTIVITIES_MODULE.read_text().upper()
+
+    assert 'INSERT INTO' not in source
+    assert 'DELETE FROM' not in source
+    assert 'CREATE TABLE' not in source

tests/laborious/worker/test_runtime_task_queues.py

@@ -1,10 +1,27 @@
+"""Queue names and how many workers a runtime starts.
+
+Every queue is runtime-scoped, so two runtimes never poll each other's work. The import worker is
+the only conditional one: unset, the four default workers are exactly what they were, and no import
+queue is polled at all.
+"""
+
+import ast
+from pathlib import Path
+
+from pytest import mark
 from sientia_do.temporal.worker.prepare_worker import build_queue_name
 
+from laborious.worker.worker import import_worker_enabled
 from laborious.workflows.drift import Drift
+from laborious.workflows.import_model import ImportModel
 from laborious.workflows.minimal_retrain import MinimalRetrain
 from laborious.workflows.predictions_batch import PredictionsBatch
 from laborious.workflows.simple_metrics import SimpleMetrics
 
+WORKER_MODULE = Path(__file__).resolve().parents[3] / 'laborious/worker/worker.py'
+
+DEFAULT_WORKFLOWS = (PredictionsBatch, MinimalRetrain, Drift, SimpleMetrics)
+
 
 def test_runtime_scoped_queue_names():
     runtime = 'prod-a'
@@ -12,3 +29,86 @@ def test_runtime_scoped_queue_names():
     assert build_queue_name(MinimalRetrain.__name__, runtime) == 'minimal_retrain-prod-a-queue'
     assert build_queue_name(Drift.__name__, runtime) == 'drift-prod-a-queue'
     assert build_queue_name(SimpleMetrics.__name__, runtime) == 'simple_metrics-prod-a-queue'
+
+
+def test_the_import_queue_shares_the_runtime_suffix():
+    assert build_queue_name(ImportModel.__name__, 'prod-a') == 'import_model-prod-a-queue'
+
+
+def test_every_queue_carries_the_same_runtime():
+    runtime = 'prod-a'
+    queues = [
+        build_queue_name(workflow.__name__, runtime)
+        for workflow in (*DEFAULT_WORKFLOWS, ImportModel)
+    ]
+
+    assert all(queue.endswith(f'-{runtime}-queue') for queue in queues)
+    assert len(set(queues)) == 5
+
+
+@mark.parametrize('value', ['1', 'true', 'TRUE', 'yes', 'on', ' true '])
+def test_the_import_worker_is_registered_when_the_flag_says_so(monkeypatch, value: str):
+    monkeypatch.setenv('IMPORT_MODEL_ENABLED', value)
+
+    assert import_worker_enabled() is True
+
+
+@mark.parametrize('value', ['', '0', 'false', 'no', 'off', 'maybe'])
+def test_the_import_worker_stays_out_by_default(monkeypatch, value: str):
+    monkeypatch.setenv('IMPORT_MODEL_ENABLED', value)
+
+    assert import_worker_enabled() is False
+
+
+def test_an_unset_flag_leaves_the_import_worker_out(monkeypatch):
+    monkeypatch.delenv('IMPORT_MODEL_ENABLED', raising=False)
+
+    assert import_worker_enabled() is False
+
+
+def _prepare_worker_calls(tree: ast.Module) -> tuple[int, int]:
+    """How many workers are built unconditionally, and how many behind the flag."""
+    conditional = 0
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.If):
+            continue
+        if 'import_worker_enabled' not in ast.unparse(node.test):
+            continue
+        conditional += sum(
+            1
+            for inner in ast.walk(node)
+            if isinstance(inner, ast.Call) and ast.unparse(inner.func) == 'prepare_worker'
+        )
+    total = sum(
+        1
+        for node in ast.walk(tree)
+        if isinstance(node, ast.Call) and ast.unparse(node.func) == 'prepare_worker'
+    )
+    return total - conditional, conditional
+
+
+def test_four_workers_by_default_and_five_with_the_flag():
+    unconditional, conditional = _prepare_worker_calls(ast.parse(WORKER_MODULE.read_text()))
+
+    assert unconditional == 4
+    assert conditional == 1
+
+
+def test_the_import_worker_registers_only_the_import_activities():
+    """The containment boundary: the import runtime polls one queue and holds one workflow."""
+    tree = ast.parse(WORKER_MODULE.read_text())
+    activities: list[str] = []
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.If) or 'import_worker_enabled' not in ast.unparse(node.test):
+            continue
+        for inner in ast.walk(node):
+            if isinstance(inner, ast.keyword) and inner.arg == 'activities':
+                activities = [ast.unparse(element) for element in inner.value.elts]
+            if isinstance(inner, ast.keyword) and inner.arg == 'main_workflow':
+                assert ast.unparse(inner.value) == 'ImportModel'
+            if isinstance(inner, ast.keyword) and inner.arg == 'other_workflows':
+                assert ast.unparse(inner.value) == '[]'
+
+    assert len(activities) == 13
+    assert all(name.startswith('activities.') for name in activities)
+    assert all('import' in name for name in activities)

tests/laborious/workflows/test_import_model.py

@@ -0,0 +1,781 @@
+"""The `import_model` workflow: order, fail-fast, and which calls are allowed to retry.
+
+The workflow itself holds no logic worth testing in isolation except the two things that can only be
+wrong here: **the order** — the claim before anything else, the names before MLflow, the document
+last, cleanup and then one terminal write in the `finally` — and **the retry split**, where a single
+misplaced policy either replays a registry call or spends 151 seconds on a hint nobody reads.
+
+Everything is asserted through the recorded activity calls, because that is exactly what the
+workflow is: a sequence of activity invocations with policies attached.
+"""
+
+from datetime import timedelta
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from pytest import fixture, mark, raises
+from sientia_do.temporal.policies import retry_policy
+
+from laborious.activities.activities import Activities
+from laborious.activities.model_import import (
+    IMPORT_STATUS_HINT_RETRY_POLICY,
+    IMPORT_STATUS_RETRY_POLICY,
+)
+from laborious.activities.model_import_errors import (
+    ImportGateRejectedError,
+    ImportInputError,
+    ImportLogRowNotClaimableError,
+)
+from laborious.utils.bundle.steps import ImportStep
+from laborious.workflows.import_model import (
+    REJECTED_INPUT_FIELDS,
+    STATUS_WRITE_ENVELOPE,
+    STATUS_WRITE_TIMEOUT,
+    ImportModel,
+)
+
+WORKFLOW_ID = 'import_model-77-abc'
+RUN_ID = '0123456789abcdef0123456789abcdef'
+DIGEST = 'a' * 64
+OBJECT_KEY = 'imported_models/factory_model_v2.sientia'
+
+BUNDLE_METADATA = {
+    'model_name': 'factory_model',
+    'experiment_name': 'factory_experiment',
+    'run_id': 'aa11bb22cc33dd44ee55ff6677889900',
+    'model_version': '2',
+    'model_project': 'factory_project',
+    'export_timestamp': '2026-08-25T12:00:00+00:00',
+}
+BUNDLE_PARAMETERS = {'model_name': 'factory_model', 'target_variable': 'SE'}
+
+# The activities that write the import log row. These four are the only calls allowed to retry.
+STATUS_ACTIVITIES = (
+    Activities.claim_import_status,
+    Activities.record_import_status,
+    Activities.record_import_names,
+    Activities.record_import_terminal_status,
+)
+CLAIM_AND_TERMINAL = (Activities.claim_import_status, Activities.record_import_terminal_status)
+HINTS = (Activities.record_import_status, Activities.record_import_names)
+
+# What a correct import does, in order.
+EXPECTED_ORDER = [
+    Activities.claim_import_status,
+    Activities.record_import_status,
+    Activities.ensure_import_bundle_retention,
+    Activities.download_import_bundle,
+    Activities.record_import_status,
+    Activities.open_import_bundle,
+    Activities.record_import_names,
+    Activities.record_import_status,
+    Activities.create_import_experiment,
+    Activities.record_import_status,
+    Activities.upload_import_artifacts,
+    Activities.record_import_status,
+    Activities.register_import_model_version,
+    Activities.record_import_status,
+    Activities.write_import_model_document,
+    Activities.cleanup_import_files,
+    Activities.record_import_terminal_status,
+]
+
+PROVISIONING = (
+    Activities.create_import_experiment,
+    Activities.upload_import_artifacts,
+    Activities.register_import_model_version,
+    Activities.write_import_model_document,
+)
+
+HAPPY_RESULTS = {
+    Activities.claim_import_status: {
+        'claimed': True,
+        'bucket_name': None,
+        'file_name': None,
+        'request_digest': None,
+    },
+    Activities.download_import_bundle: {
+        'path': '/work/bundle.sientia',
+        'size': 4096,
+        'digest': DIGEST,
+    },
+    Activities.open_import_bundle: {
+        'extracted_dir': '/work/extract',
+        'metadata': BUNDLE_METADATA,
+        'parameters': BUNDLE_PARAMETERS,
+        'digest': DIGEST,
+    },
+    Activities.create_import_experiment: {'experiment_id': '17'},
+    Activities.upload_import_artifacts: {'run_id': RUN_ID, 'run_name': 'import-factory_model-v2'},
+    Activities.register_import_model_version: {'model_name': 'factory_model', 'version': '4'},
+    Activities.write_import_model_document: {'written': True, 'id': '8'},
+    Activities.cleanup_import_files: {'cleaned': True, 'existed': True},
+    Activities.record_import_terminal_status: {'status': 'IMPORT_SUCCESS'},
+}
+
+
+class Recorder:
+    """Stands in for `workflow.execute_activity_method`, recording every invocation."""
+
+    def __init__(self, failures: dict | None = None, results: dict | None = None) -> None:
+        self.results = {**HAPPY_RESULTS, **(results or {})}
+        self.failures = dict(failures or {})
+        self.calls: list[SimpleNamespace] = []
+
+    async def __call__(self, method, payload, **kwargs):
+        self.calls.append(SimpleNamespace(method=method, payload=payload, kwargs=kwargs))
+        if method in self.failures:
+            raise self.failures[method]
+        return self.results.get(method, {})
+
+    @property
+    def order(self) -> list:
+        return [call.method for call in self.calls]
+
+    def payload_of(self, method):
+        return next(call.payload for call in self.calls if call.method is method)
+
+    def last_payload_of(self, method):
+        return [call.payload for call in self.calls if call.method is method][-1]
+
+    def calls_to(self, method) -> list[SimpleNamespace]:
+        return [call for call in self.calls if call.method is method]
+
+
+def valid_input(**overrides) -> dict:
+    return {
+        'import_run_id': 77,
+        'object_key': OBJECT_KEY,
+        'expected_digest': DIGEST,
+        'password_envelope': 'ZW52ZWxvcGU=',
+        **overrides,
+    }
+
+
+@fixture
+def import_workflow() -> ImportModel:
+    return ImportModel()
+
+
+def run_with(recorder: Recorder):
+    """Patch the workflow module's `workflow` object and hand back the patch context."""
+    patcher = patch('laborious.workflows.import_model.workflow')
+    workflow_mock = patcher.start()
+    workflow_mock.info.return_value = MagicMock(workflow_id=WORKFLOW_ID)
+    # Assigned directly rather than wrapped in an `AsyncMock`: a mock whose side effect is an
+    # instance with an `async def __call__` hands back the coroutine instead of awaiting it.
+    workflow_mock.execute_activity_method = recorder
+    return patcher, workflow_mock
+
+
+@fixture
+def recorder():
+    return Recorder()
+
+
+@fixture
+def workflow_env(recorder: Recorder):
+    patcher, workflow_mock = run_with(recorder)
+    yield SimpleNamespace(recorder=recorder, workflow=workflow_mock)
+    patcher.stop()
+
+
+# ------------------------------------------------------------------------------- order
+
+
+@mark.asyncio
+async def test_a_successful_import_calls_its_activities_in_order(
+    import_workflow: ImportModel, workflow_env
+):
+    result = await import_workflow.run(valid_input())
+
+    assert workflow_env.recorder.order == EXPECTED_ORDER
+    assert result == {
+        'import_run_id': 77,
+        'run_id': RUN_ID,
+        'model_name': 'factory_model',
+        'version': '4',
+        'document_id': '8',
+    }
+
+
+@mark.asyncio
+async def test_the_claim_is_the_first_thing_that_happens(
+    import_workflow: ImportModel, workflow_env
+):
+    await import_workflow.run(valid_input())
+
+    assert workflow_env.recorder.order[0] is Activities.claim_import_status
+
+
+@mark.asyncio
+async def test_the_names_are_recorded_before_mlflow_is_touched(
+    import_workflow: ImportModel, workflow_env
+):
+    """So a person watching the import list sees which model is being imported, while it runs."""
+    await import_workflow.run(valid_input())
+
+    order = workflow_env.recorder.order
+    assert order.index(Activities.record_import_names) < order.index(
+        Activities.create_import_experiment
+    )
+    names = workflow_env.recorder.payload_of(Activities.record_import_names)
+    assert names['experiment_name'] == 'factory_experiment'
+    assert names['run_name'] == 'import-factory_model-v2'
+
+
+@mark.asyncio
+async def test_the_run_is_created_with_exactly_the_recorded_name(
+    import_workflow: ImportModel, workflow_env
+):
+    await import_workflow.run(valid_input())
+
+    recorded = workflow_env.recorder.payload_of(Activities.record_import_names)['run_name']
+    used = workflow_env.recorder.payload_of(Activities.upload_import_artifacts)['run_name']
+    assert used == recorded
+
+
+@mark.asyncio
+async def test_the_model_document_is_the_last_thing_provisioned(
+    import_workflow: ImportModel, workflow_env
+):
+    await import_workflow.run(valid_input())
+
+    order = workflow_env.recorder.order
+    for step in PROVISIONING[:-1]:
+        assert order.index(step) < order.index(Activities.write_import_model_document)
+    assert order[-2:] == [Activities.cleanup_import_files, Activities.record_import_terminal_status]
+
+
+@mark.asyncio
+async def test_the_document_target_comes_from_the_bundle_s_parameters(
+    import_workflow: ImportModel, workflow_env
+):
+    await import_workflow.run(valid_input())
+
+    assert (
+        workflow_env.recorder.payload_of(Activities.write_import_model_document)['target'] == 'SE'
+    )
+
+
+@mark.asyncio
+async def test_the_terminal_write_reports_a_success_with_its_last_step(
+    import_workflow: ImportModel, workflow_env
+):
+    await import_workflow.run(valid_input())
+
+    terminal = workflow_env.recorder.payload_of(Activities.record_import_terminal_status)
+    assert terminal['succeeded'] is True
+    assert terminal['step'] == ImportStep.MODEL_DOCUMENT.value
+    assert terminal['cleanup_failed'] is False
+    assert terminal['source'] == BUNDLE_METADATA
+
+
+@mark.asyncio
+async def test_the_download_is_handed_the_digest_the_request_carried(
+    import_workflow: ImportModel, workflow_env
+):
+    await import_workflow.run(valid_input())
+
+    payload = workflow_env.recorder.payload_of(Activities.download_import_bundle)
+    assert payload['expected_digest'] == DIGEST
+    assert payload['object_key'] == OBJECT_KEY
+
+
+# --------------------------------------------------------------------------- fail-fast
+
+
+@mark.asyncio
+@mark.parametrize(
+    ('failing', 'step'),
+    [
+        (Activities.download_import_bundle, ImportStep.DOWNLOAD),
+        (Activities.open_import_bundle, ImportStep.DECRYPTION),
+        (Activities.create_import_experiment, ImportStep.EXPERIMENT_CREATION),
+        (Activities.upload_import_artifacts, ImportStep.ARTIFACT_UPLOAD),
+        (Activities.register_import_model_version, ImportStep.REGISTRATION),
+        (Activities.write_import_model_document, ImportStep.MODEL_DOCUMENT),
+    ],
+    ids=lambda value: getattr(value, '__name__', str(value)),
+)
+async def test_a_failing_step_is_recorded_as_that_step_and_stops_the_import(
+    import_workflow: ImportModel, failing, step: ImportStep
+):
+    recorder = Recorder(failures={failing: RuntimeError('dependency unavailable')})
+    patcher, _ = run_with(recorder)
+    try:
+        with raises(RuntimeError, match='dependency unavailable'):
+            await import_workflow.run(valid_input())
+    finally:
+        patcher.stop()
+
+    terminal = recorder.payload_of(Activities.record_import_terminal_status)
+    assert terminal['succeeded'] is False
+    assert terminal['step'] == step.value
+    # Nothing after the failing step was attempted, and cleanup still ran before the verdict.
+    order = recorder.order
+    assert order[-2:] == [Activities.cleanup_import_files, Activities.record_import_terminal_status]
+    later = EXPECTED_ORDER[EXPECTED_ORDER.index(failing) + 1 :]
+    for method in later:
+        if method in (Activities.cleanup_import_files, Activities.record_import_terminal_status):
+            continue
+        if method in STATUS_ACTIVITIES:
+            continue
+        assert method not in order
+
+
+@mark.asyncio
+async def test_nothing_is_provisioned_before_the_gates_pass(import_workflow: ImportModel):
+    recorder = Recorder(
+        failures={
+            Activities.open_import_bundle: ImportGateRejectedError(
+                'rejected', step=ImportStep.ARCHIVE_INSPECTION, gate=4
+            )
+        }
+    )
+    patcher, _ = run_with(recorder)
+    try:
+        with raises(ImportGateRejectedError):
+            await import_workflow.run(valid_input())
+    finally:
+        patcher.stop()
+
+    for method in PROVISIONING:
+        assert method not in recorder.order
+
+
+@mark.asyncio
+async def test_a_gate_rejection_is_recorded_with_the_gate_that_rejected_it(
+    import_workflow: ImportModel,
+):
+    """Gates 2 to 7 share one activity, so the step travels in the error's details, not in the
+    step the workflow was attempting."""
+    recorder = Recorder(
+        failures={
+            Activities.open_import_bundle: ImportGateRejectedError(
+                'rejected', step=ImportStep.CONTENT_POLICY, gate=7
+            )
+        }
+    )
+    patcher, _ = run_with(recorder)
+    try:
+        with raises(ImportGateRejectedError):
+            await import_workflow.run(valid_input())
+    finally:
+        patcher.stop()
+
+    terminal = recorder.payload_of(Activities.record_import_terminal_status)
+    assert terminal['step'] == ImportStep.CONTENT_POLICY.value
+    assert terminal['gate'] == 7
+
+
+@mark.asyncio
+async def test_an_ambiguous_decryption_failure_keeps_its_step(import_workflow: ImportModel):
+    recorder = Recorder(
+        failures={
+            Activities.open_import_bundle: ImportGateRejectedError(
+                'the bundle could not be opened', step=ImportStep.DECRYPTION, gate=3
+            )
+        }
+    )
+    patcher, _ = run_with(recorder)
+    try:
+        with raises(ImportGateRejectedError):
+            await import_workflow.run(valid_input())
+    finally:
+        patcher.stop()
+
+    terminal = recorder.payload_of(Activities.record_import_terminal_status)
+    assert terminal['step'] == ImportStep.DECRYPTION.value
+    assert terminal['gate'] == 3
+
+
+# ------------------------------------------------------------------- the input contract
+
+
+@mark.asyncio
+@mark.parametrize('field', list(REJECTED_INPUT_FIELDS))
+async def test_an_unknown_field_is_rejected_rather_than_ignored(
+    import_workflow: ImportModel, field: str
+):
+    """A caller that believes it recorded a project link must be told it did not."""
+    recorder = Recorder()
+    patcher, _ = run_with(recorder)
+    try:
+        with raises(ImportInputError, match='does not accept'):
+            await import_workflow.run(valid_input(**{field: 'whatever'}))
+    finally:
+        patcher.stop()
+
+    # After the claim, so the refusal lands on the record instead of only in the logs.
+    assert recorder.order[0] is Activities.claim_import_status
+    assert Activities.download_import_bundle not in recorder.order
+    terminal = recorder.payload_of(Activities.record_import_terminal_status)
+    assert terminal['step'] == ImportStep.RECEIVED.value
+    assert terminal['succeeded'] is False
+
+
+@mark.asyncio
+@mark.parametrize(
+    'digest',
+    [None, '', 'not-a-digest', 'a' * 63, 'a' * 65, 'g' * 64],
+    ids=['missing', 'empty', 'short-text', 'too-short', 'too-long', 'not-hex'],
+)
+async def test_a_missing_or_malformed_digest_fails_before_anything_is_downloaded(
+    import_workflow: ImportModel, digest
+):
+    recorder = Recorder()
+    patcher, _ = run_with(recorder)
+    try:
+        with raises(ImportInputError):
+            await import_workflow.run(valid_input(expected_digest=digest))
+    finally:
+        patcher.stop()
+
+    assert Activities.download_import_bundle not in recorder.order
+    assert (
+        recorder.payload_of(Activities.record_import_terminal_status)['step']
+        == ImportStep.RECEIVED.value
+    )
+
+
+@mark.asyncio
+@mark.parametrize('missing', ['object_key', 'password_envelope'])
+async def test_a_request_missing_a_required_field_fails_at_received(
+    import_workflow: ImportModel, missing: str
+):
+    recorder = Recorder()
+    patcher, _ = run_with(recorder)
+    payload = valid_input()
+    payload.pop(missing)
+    try:
+        with raises(ImportInputError, match=missing):
+            await import_workflow.run(payload)
+    finally:
+        patcher.stop()
+
+    assert Activities.download_import_bundle not in recorder.order
+
+
+@mark.asyncio
+async def test_a_record_naming_another_object_is_a_received_failure(
+    import_workflow: ImportModel,
+):
+    recorder = Recorder(
+        results={
+            Activities.claim_import_status: {
+                'claimed': True,
+                'file_name': 'imported_models/somebody_elses.sientia',
+                'request_digest': None,
+            }
+        }
+    )
+    patcher, _ = run_with(recorder)
+    try:
+        with raises(ImportInputError, match='different object'):
+            await import_workflow.run(valid_input())
+    finally:
+        patcher.stop()
+
+    assert Activities.download_import_bundle not in recorder.order
+
+
+@mark.asyncio
+async def test_a_record_naming_another_digest_is_a_received_failure(
+    import_workflow: ImportModel,
+):
+    recorder = Recorder(
+        results={
+            Activities.claim_import_status: {
+                'claimed': True,
+                'file_name': OBJECT_KEY,
+                'request_digest': 'b' * 64,
+            }
+        }
+    )
+    patcher, _ = run_with(recorder)
+    try:
+        with raises(ImportInputError, match='different digest'):
+            await import_workflow.run(valid_input())
+    finally:
+        patcher.stop()
+
+    assert Activities.download_import_bundle not in recorder.order
+
+
+@mark.asyncio
+@mark.parametrize('value', [None, 0, -1, '77', 7.0, True], ids=lambda v: f'{type(v).__name__}-{v}')
+async def test_an_unusable_record_id_fails_before_any_activity(import_workflow: ImportModel, value):
+    """There is no record to claim and none to write: the run ends FAILED and nothing is touched."""
+    recorder = Recorder()
+    patcher, _ = run_with(recorder)
+    try:
+        with raises(ImportInputError, match='positive integer'):
+            await import_workflow.run(valid_input(import_run_id=value))
+    finally:
+        patcher.stop()
+
+    assert recorder.calls == []
+
+
+@mark.asyncio
+async def test_a_claim_that_matches_nothing_writes_nothing_at_all(import_workflow: ImportModel):
+    recorder = Recorder(
+        failures={
+            Activities.claim_import_status: ImportLogRowNotClaimableError(
+                'import log row 77 could not be claimed: no such import log row'
+            )
+        }
+    )
+    patcher, _ = run_with(recorder)
+    try:
+        with raises(ImportLogRowNotClaimableError):
+            await import_workflow.run(valid_input())
+    finally:
+        patcher.stop()
+
+    # The claim is outside the try/finally on purpose: no row is owned, so there is nothing to clean
+    # up and nothing to record.
+    assert recorder.order == [Activities.claim_import_status]
+
+
+# ----------------------------------------------------------- failures that must not fail
+
+
+@mark.asyncio
+async def test_a_lost_hint_does_not_fail_a_working_import(import_workflow: ImportModel):
+    recorder = Recorder(failures={Activities.record_import_status: RuntimeError('db down')})
+    patcher, _ = run_with(recorder)
+    try:
+        result = await import_workflow.run(valid_input())
+    finally:
+        patcher.stop()
+
+    assert result['run_id'] == RUN_ID
+    assert Activities.report_import_status_write_failure in recorder.order
+    assert recorder.payload_of(Activities.record_import_terminal_status)['succeeded'] is True
+
+
+@mark.asyncio
+async def test_a_lost_names_write_does_not_fail_a_working_import(import_workflow: ImportModel):
+    recorder = Recorder(failures={Activities.record_import_names: RuntimeError('db down')})
+    patcher, _ = run_with(recorder)
+    try:
+        result = await import_workflow.run(valid_input())
+    finally:
+        patcher.stop()
+
+    assert result['version'] == '4'
+    assert Activities.upload_import_artifacts in recorder.order
+
+
+@mark.asyncio
+async def test_a_retention_rule_that_cannot_be_set_does_not_fail_an_import(
+    import_workflow: ImportModel,
+):
+    recorder = Recorder(
+        failures={Activities.ensure_import_bundle_retention: RuntimeError('access denied')}
+    )
+    patcher, _ = run_with(recorder)
+    try:
+        result = await import_workflow.run(valid_input())
+    finally:
+        patcher.stop()
+
+    assert result['run_id'] == RUN_ID
+
+
+@mark.asyncio
+async def test_a_cleanup_failure_is_carried_into_the_verdict_not_raised(
+    import_workflow: ImportModel,
+):
+    recorder = Recorder(failures={Activities.cleanup_import_files: RuntimeError('busy')})
+    patcher, _ = run_with(recorder)
+    try:
+        result = await import_workflow.run(valid_input())
+    finally:
+        patcher.stop()
+
+    terminal = recorder.payload_of(Activities.record_import_terminal_status)
+    assert result['run_id'] == RUN_ID
+    assert terminal['succeeded'] is True
+    assert terminal['cleanup_failed'] is True
+
+
+@mark.asyncio
+async def test_a_terminal_write_that_cannot_land_does_not_invent_a_failed_import(
+    import_workflow: ImportModel,
+):
+    recorder = Recorder(
+        failures={Activities.record_import_terminal_status: RuntimeError('still down')}
+    )
+    patcher, _ = run_with(recorder)
+    try:
+        result = await import_workflow.run(valid_input())
+    finally:
+        patcher.stop()
+
+    assert result['run_id'] == RUN_ID
+    reported = recorder.payload_of(Activities.report_import_status_write_failure)
+    assert reported['verdict'] == 'succeeded=True step=model_document gate=None'
+    assert 'still down' in reported['detail']
+
+
+@mark.asyncio
+async def test_a_terminal_write_failure_does_not_mask_a_real_failure(
+    import_workflow: ImportModel,
+):
+    recorder = Recorder(
+        failures={
+            Activities.register_import_model_version: RuntimeError('registry refused'),
+            Activities.record_import_terminal_status: RuntimeError('status db down'),
+        }
+    )
+    patcher, _ = run_with(recorder)
+    try:
+        with raises(RuntimeError, match='registry refused'):
+            await import_workflow.run(valid_input())
+    finally:
+        patcher.stop()
+
+    reported = recorder.payload_of(Activities.report_import_status_write_failure)
+    assert reported['verdict'] == 'succeeded=False step=registration gate=None'
+
+
+@mark.asyncio
+async def test_reporting_the_report_is_where_it_stops(import_workflow: ImportModel):
+    """A failure in the failure channel cannot become the import's outcome."""
+    recorder = Recorder(
+        failures={
+            Activities.record_import_terminal_status: RuntimeError('down'),
+            Activities.report_import_status_write_failure: RuntimeError('also down'),
+        }
+    )
+    patcher, _ = run_with(recorder)
+    try:
+        result = await import_workflow.run(valid_input())
+    finally:
+        patcher.stop()
+
+    assert result['run_id'] == RUN_ID
+
+
+# ------------------------------------------------------------------- the retry split
+
+
+@mark.asyncio
+async def test_only_the_status_writes_retry(import_workflow: ImportModel, workflow_env):
+    await import_workflow.run(valid_input())
+
+    for call in workflow_env.recorder.calls:
+        if call.method in STATUS_ACTIVITIES:
+            continue
+        assert call.kwargs['retry_policy'] is retry_policy
+        assert retry_policy.maximum_attempts == 1
+
+
+@mark.asyncio
+async def test_the_claim_and_the_terminal_write_use_the_long_envelope(
+    import_workflow: ImportModel, workflow_env
+):
+    await import_workflow.run(valid_input())
+
+    for method in CLAIM_AND_TERMINAL:
+        for call in workflow_env.recorder.calls_to(method):
+            assert call.kwargs['retry_policy'] is IMPORT_STATUS_RETRY_POLICY
+            assert call.kwargs['start_to_close_timeout'] == STATUS_WRITE_TIMEOUT
+            assert call.kwargs['schedule_to_close_timeout'] == STATUS_WRITE_ENVELOPE
+
+
+@mark.asyncio
+async def test_the_hints_use_the_short_policy(import_workflow: ImportModel, workflow_env):
+    await import_workflow.run(valid_input())
+
+    for method in HINTS:
+        for call in workflow_env.recorder.calls_to(method):
+            assert call.kwargs['retry_policy'] is IMPORT_STATUS_HINT_RETRY_POLICY
+            # No long envelope: a hint that cannot land must not hold the import open.
+            assert 'schedule_to_close_timeout' not in call.kwargs
+
+
+@mark.asyncio
+async def test_no_provisioning_call_references_a_retrying_policy(
+    import_workflow: ImportModel, workflow_env
+):
+    """A repeated registry call or Mongo insert writes twice; that is why they cannot retry."""
+    await import_workflow.run(valid_input())
+
+    for call in workflow_env.recorder.calls:
+        if call.method in PROVISIONING or call.method is Activities.open_import_bundle:
+            assert call.kwargs['retry_policy'] not in (
+                IMPORT_STATUS_RETRY_POLICY,
+                IMPORT_STATUS_HINT_RETRY_POLICY,
+            )
+
+
+@mark.asyncio
+async def test_the_only_retrying_calls_are_the_four_status_writes(
+    import_workflow: ImportModel, workflow_env
+):
+    await import_workflow.run(valid_input())
+
+    retrying = {
+        call.method
+        for call in workflow_env.recorder.calls
+        if call.kwargs['retry_policy']
+        in (IMPORT_STATUS_RETRY_POLICY, IMPORT_STATUS_HINT_RETRY_POLICY)
+    }
+    assert retrying == set(STATUS_ACTIVITIES)
+
+
+def test_the_retry_policies_are_the_values_the_design_chose():
+    assert IMPORT_STATUS_RETRY_POLICY.maximum_attempts == 10
+    assert IMPORT_STATUS_RETRY_POLICY.initial_interval == timedelta(seconds=1)
+    assert IMPORT_STATUS_RETRY_POLICY.backoff_coefficient == 2.0
+    assert IMPORT_STATUS_RETRY_POLICY.maximum_interval == timedelta(seconds=30)
+    assert IMPORT_STATUS_HINT_RETRY_POLICY.maximum_attempts == 3
+    assert IMPORT_STATUS_HINT_RETRY_POLICY.maximum_interval == timedelta(seconds=2)
+
+
+def test_the_terminal_write_s_envelope_outlasts_its_own_retries():
+    """Set this smaller and Temporal cuts the retries short, silently undoing the retry split."""
+    backoff = sum(
+        min(
+            IMPORT_STATUS_RETRY_POLICY.initial_interval.total_seconds()
+            * IMPORT_STATUS_RETRY_POLICY.backoff_coefficient**attempt,
+            IMPORT_STATUS_RETRY_POLICY.maximum_interval.total_seconds(),
+        )
+        for attempt in range(IMPORT_STATUS_RETRY_POLICY.maximum_attempts - 1)
+    )
+
+    assert backoff == 151
+    assert STATUS_WRITE_ENVELOPE.total_seconds() > backoff
+
+
+def test_a_hint_gives_up_inside_a_short_bound():
+    """Three attempts capped at two seconds: a lost hint costs the import seconds, not minutes."""
+    backoff = sum(
+        min(
+            IMPORT_STATUS_HINT_RETRY_POLICY.initial_interval.total_seconds()
+            * IMPORT_STATUS_HINT_RETRY_POLICY.backoff_coefficient**attempt,
+            IMPORT_STATUS_HINT_RETRY_POLICY.maximum_interval.total_seconds(),
+        )
+        for attempt in range(IMPORT_STATUS_HINT_RETRY_POLICY.maximum_attempts - 1)
+    )
+
+    assert backoff <= 3
+
+
+@mark.asyncio
+async def test_no_retrying_call_carries_a_payload_that_would_insert(
+    import_workflow: ImportModel, workflow_env
+):
+    """What retries must be idempotent, so the payload may not name a row to create."""
+    await import_workflow.run(valid_input())
+
+    for call in workflow_env.recorder.calls:
+        if call.method not in STATUS_ACTIVITIES:
+            continue
+        assert call.payload['import_run_id'] == 77
+        for owned in ('username', 'request_data', 'bucket_name', 'file_name', 'run_type'):
+            assert owned not in call.payload