diff --git a/docs/decisions/ADR-001-user-facing-activity-logs.md b/docs/decisions/ADR-001-user-facing-activity-logs.md
new file mode 100644
index 0000000..93afdeb
--- /dev/null
+++ b/docs/decisions/ADR-001-user-facing-activity-logs.md
@@ -0,0 +1,95 @@
+# ADR-001: User-Facing Activity Logs
+
+## Status
+
+Accepted — implemented in PR #66
+
+## Date
+
+2026-04-17
+
+## Context
+
+The bottom Logs panel showed raw developer-internal messages like
+`Executing Model Node: Database: globe_testing Database Type: postgres ...`.
+Users had no way to tell what happened after triggering an action. We needed
+plain-language activity messages ("Model orders built successfully in 0.42s")
+without breaking the existing developer log pipeline.
+
+### Constraints
+
+- The event pipeline (`fire_event` -> `EventManager` -> `LogHelper` -> Celery -> Socket.IO) is shared across 100+ developer events. Any change must be additive.
+- WebSocket rooms are keyed by session ID. We cannot add a second channel without frontend proxy changes (React dev server doesn't proxy WS by default).
+- Events are defined as dataclasses backed by protobuf message types. Adding new events is cheap; modifying existing ones risks breaking the proto contract.
+
+## Options Considered
+
+### A. Add an `audience` flag to existing events
+
+Reuse existing event classes. Add `audience="user"` to a subset of them.
+
+- **Pro:** No new classes. Smallest diff.
+- **Con:** Developer events have different message formats, verbosity, and lifecycle semantics. Retrofitting "user-friendly" messages onto events designed for `DEBUG`/`INFO` log dumps is fragile. A developer event like `MaterializationTypeTable` fires mid-execution with internal context — rewriting its `message()` to be user-friendly breaks developer debugging.
+
+**Rejected** — mixes concerns, high coupling between developer and user message formats.
+
+### B. Separate `UserLevel` base class (chosen)
+
+New `UserLevel(BaseEvent)` dataclass that overrides `audience()` to return `"user"`. 20 event classes (U001-U020) in 5 priority tiers. Socket payload gains an `audience` field. Frontend filters by audience in the log-level dropdown.
+
+- **Pro:** Clean separation. Existing events untouched. Adding a new user event = one class + one `fire_event()` call. Rich UI metadata (`title()`, `subtitle()`, `event_status()`) lives on the event class, not in the transport layer.
+- **Con:** 20 new classes. More code than option A.
+
+**Accepted** — separation of concerns outweighs class count.
+
+### C. Separate WebSocket channel for user logs
+
+Dedicated socket event (`user_activity`) instead of reusing the `logs:{session_id}` channel.
+
+- **Pro:** Complete decoupling. Frontend subscribes to two events independently.
+- **Con:** Duplicates the LogHelper -> Celery -> socket pipeline. Frontend must manage two subscriptions. Mixed "All logs" view requires merging two streams client-side. React dev server proxy needs `ws: true` configuration for each channel.
+
+**Rejected** — too much infrastructure duplication for a filtering problem.
+
+## Decision
+
+**Option B.** Single inheritance (`UserLevel` extends `BaseEvent`) keeps the event pipeline unified while providing a clean audience separation point.
+
+### Pipeline flow
+
+```
+fire_event(UserLevelEvent)
+  -> EventManager.fire_event()
+  -> Logger.write_line()          # checks audience()
+     if "user": build rich payload {title, subtitle, status, code, timestamp}
+     if "developer": build plain payload {level, message}
+  -> LogHelper.publish_log()      # Celery queue
+  -> logs_consumer task
+  -> sio.emit("logs:{session_id}")
+  -> Frontend filters by audience
+```
+
+### Key design choices
+
+1. **Audience is a class-level method, not a message field.** `UserLevel.audience()` returns `"user"` — the event *is* user-facing by type, not by configuration. This prevents accidentally marking a developer event as user-facing.
+
+2. **Rich metadata on the event class.** `title()`, `subtitle()`, `event_status()` are methods on `UserLevel`, not fields on the socket payload. The transport layer reads them; the event class owns the rendering logic.
+
+3. **Frontend "User activity" as default view.** The dropdown defaults to showing only `audience === "user"` entries. Developer logs are one click away ("All logs") but not the default noise.
+
+4. **Events fire after success, not inside try.** Each `fire_event()` call is placed after the operation succeeds to prevent false activity entries. If the event itself fails, the existing `write_line` try/except logs it without interrupting the operation.
+
+## Consequences
+
+- Every new user-facing action requires a new `UserLevel` subclass + proto type + `fire_event()` call at the correct point in the operation.
+- The `audience` field is now part of the WebSocket payload contract — removing it would break frontend filtering.
+- Developer logs are completely unaffected — zero changes to existing ~100 event classes.
+- The 5-tier priority structure (P1 core ops, P2 scheduler, P3 CRUD, P4 connections, P5 environments) provides a framework for deciding which future actions deserve user-facing events.
+
+## References
+
+- PR #66: feat: user-facing activity logs with 19 curated events
+- PR #59: log-level filter infrastructure (prerequisite)
+- Event definitions: `backend/visitran/events/types.py` (lines 829-1153)
+- UserLevel base class: `backend/visitran/events/base_types.py` (lines 94-114)
+- Audience routing: `backend/visitran/events/eventmgr.py` (lines 126-151)
diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py
new file mode 100644
index 0000000..12aa0bc
--- /dev/null
+++ b/tests/unit/conftest.py
@@ -0,0 +1,15 @@
+import os
+import sys
+
+# Ensure backend/ is on sys.path so both `visitran.*` and `backend.*` are importable
+backend_dir = os.path.join(os.path.dirname(__file__), "..", "..", "backend")
+backend_dir = os.path.abspath(backend_dir)
+if backend_dir not in sys.path:
+    sys.path.insert(0, backend_dir)
+
+# Configure Django before any Django-dependent import
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.server.settings.dev")
+
+import django  # noqa: E402
+
+django.setup()
diff --git a/tests/unit/test_user_activity_events.py b/tests/unit/test_user_activity_events.py
new file mode 100644
index 0000000..ca4c7bd
--- /dev/null
+++ b/tests/unit/test_user_activity_events.py
@@ -0,0 +1,654 @@
+"""Tests for user-facing activity log events (U001–U020).
+
+Each event is tested for:
+- audience() returns "user"
+- code() returns the correct U-code
+- message() renders correctly with given fields
+- title() / subtitle() produce expected UI text
+- event_status() returns the correct status string
+- level_tag() returns the expected EventLevel
+"""
+
+import pytest
+
+from visitran.events.base_types import EventLevel, UserLevel
+from visitran.events.types import (
+    ConnectionCreated,
+    ConnectionDeletedEvt,
+    ConnectionDeleteFailedEvt,
+    ConnectionTested,
+    EnvironmentCreated,
+    EnvironmentDeleted,
+    FileDeleted,
+    FileRenamed,
+    JobCreated,
+    JobDeleted,
+    JobTriggered,
+    JobUpdated,
+    ModelConfigured,
+    ModelCreated,
+    ModelRunFailed,
+    ModelRunStarted,
+    ModelRunSucceeded,
+    SeedCompleted,
+    TransformationApplied,
+    TransformationDeleted,
+)
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestUserLevelBase:
+    """Verify the UserLevel base class contract."""
+
+    def test_audience_is_user(self):
+        """All UserLevel subclasses must return audience='user'."""
+        event = ModelCreated(model_name="test")
+        assert event.audience() == "user"
+
+    def test_default_level_is_info(self):
+        event = ModelCreated(model_name="test")
+        assert event.level_tag() == EventLevel.INFO
+
+    def test_default_event_status_is_success(self):
+        event = ModelCreated(model_name="test")
+        assert event.event_status() == "success"
+
+    def test_default_subtitle_is_empty(self):
+        """UserLevel.subtitle() defaults to empty string unless overridden."""
+        event = JobUpdated(job_name="nightly")
+        assert event.subtitle() == ""
+
+
+# ── P1: Core model operations ──────────────────────────────────────────
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestModelRunStarted:
+    @pytest.fixture()
+    def event(self):
+        return ModelRunStarted(
+            model_name="orders",
+            source_table="raw.orders",
+            destination_table="analytics.orders",
+            materialization="table",
+        )
+
+    def test_is_user_level(self, event):
+        assert isinstance(event, UserLevel)
+        assert event.audience() == "user"
+
+    def test_code(self, event):
+        assert event.code() == "U001"
+
+    def test_message(self, event):
+        msg = event.message()
+        assert '"orders"' in msg
+        assert "analytics.orders" in msg
+        assert "as table" in msg
+        assert "raw.orders" in msg
+
+    def test_message_without_materialization(self):
+        event = ModelRunStarted(
+            model_name="orders",
+            source_table="raw.orders",
+            destination_table="analytics.orders",
+            materialization="",
+        )
+        assert "as " not in event.message()
+
+    def test_title(self, event):
+        assert event.title() == 'Building model "orders"'
+
+    def test_subtitle(self, event):
+        sub = event.subtitle()
+        assert "TABLE" in sub
+        assert "raw.orders → analytics.orders" in sub
+
+    def test_event_status(self, event):
+        assert event.event_status() == "running"
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestModelRunSucceeded:
+    @pytest.fixture()
+    def event(self):
+        return ModelRunSucceeded(model_name="orders", duration_seconds=1.234)
+
+    def test_code(self, event):
+        assert event.code() == "U002"
+
+    def test_message(self, event):
+        msg = event.message()
+        assert '"orders"' in msg
+        assert "successfully" in msg
+        assert "1.2s" in msg
+
+    def test_message_without_duration(self):
+        event = ModelRunSucceeded(model_name="orders", duration_seconds=0.0)
+        msg = event.message()
+        assert "successfully" in msg
+        assert " in " not in msg
+
+    def test_title(self, event):
+        assert event.title() == 'Model "orders" built successfully'
+
+    def test_subtitle_with_duration(self, event):
+        assert "1.2s" in event.subtitle()
+
+    def test_subtitle_without_duration(self):
+        event = ModelRunSucceeded(model_name="orders", duration_seconds=0.0)
+        assert event.subtitle() == ""
+
+    def test_event_status(self, event):
+        assert event.event_status() == "success"
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestModelRunFailed:
+    @pytest.fixture()
+    def event(self):
+        return ModelRunFailed(model_name="orders", error="division by zero")
+
+    def test_code(self, event):
+        assert event.code() == "U003"
+
+    def test_message(self, event):
+        msg = event.message()
+        assert '"orders"' in msg
+        assert "failed" in msg
+        assert "division by zero" in msg
+
+    def test_message_truncates_long_error(self):
+        long_error = "x" * 200
+        event = ModelRunFailed(model_name="orders", error=long_error)
+        msg = event.message()
+        assert len(msg) < 200
+        assert msg.endswith("…")
+
+    def test_title(self, event):
+        assert event.title() == 'Model "orders" failed'
+
+    def test_subtitle_truncates_long_error(self):
+        long_error = "x" * 200
+        event = ModelRunFailed(model_name="orders", error=long_error)
+        sub = event.subtitle()
+        assert len(sub) <= 151  # 150 chars + ellipsis
+        assert sub.endswith("…")
+
+    def test_event_status(self, event):
+        assert event.event_status() == "error"
+
+    def test_level_is_error(self, event):
+        assert event.level_tag() == EventLevel.ERROR
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestTransformationApplied:
+    @pytest.fixture()
+    def event(self):
+        return TransformationApplied(model_name="orders", transformation_type="sort")
+
+    def test_code(self, event):
+        assert event.code() == "U004"
+
+    def test_message(self, event):
+        assert event.message() == 'Applied sort transformation on "orders"'
+
+    def test_title(self, event):
+        assert event.title() == 'Applied sort transformation on "orders"'
+
+    def test_audience(self, event):
+        assert event.audience() == "user"
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestTransformationDeleted:
+    @pytest.fixture()
+    def event(self):
+        return TransformationDeleted(model_name="orders", transformation_type="filter")
+
+    def test_code(self, event):
+        assert event.code() == "U005"
+
+    def test_message(self, event):
+        assert event.message() == 'Removed filter transformation from "orders"'
+
+    def test_title(self, event):
+        assert event.title() == 'Removed filter transformation from "orders"'
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestModelConfigured:
+    @pytest.fixture()
+    def event(self):
+        return ModelConfigured(
+            model_name="orders", source="raw_db.public", destination="analytics.dw"
+        )
+
+    def test_code(self, event):
+        assert event.code() == "U006"
+
+    def test_message(self, event):
+        msg = event.message()
+        assert '"orders"' in msg
+        assert "raw_db.public" in msg
+        assert "analytics.dw" in msg
+
+    def test_title(self, event):
+        assert event.title() == 'Configured model "orders"'
+
+    def test_subtitle(self, event):
+        assert event.subtitle() == "raw_db.public → analytics.dw"
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestSeedCompleted:
+    def test_code(self):
+        event = SeedCompleted(seed_name="countries", schema_name="public", status="Success")
+        assert event.code() == "U007"
+
+    def test_message_success(self):
+        event = SeedCompleted(seed_name="countries", schema_name="public", status="Success")
+        assert event.message() == 'Seed "countries" loaded into "public"'
+
+    def test_message_failure(self):
+        event = SeedCompleted(seed_name="countries", schema_name="public", status="Failure")
+        assert event.message() == 'Seed "countries" failed in "public"'
+
+    def test_title_success(self):
+        event = SeedCompleted(seed_name="countries", schema_name="public", status="Success")
+        assert event.title() == 'Seed "countries" loaded successfully'
+
+    def test_title_failure(self):
+        event = SeedCompleted(seed_name="countries", schema_name="public", status="Failure")
+        assert event.title() == 'Seed "countries" failed'
+
+    def test_subtitle(self):
+        event = SeedCompleted(seed_name="countries", schema_name="public", status="Success")
+        assert event.subtitle() == "Schema: public"
+
+    def test_event_status_success(self):
+        event = SeedCompleted(seed_name="countries", schema_name="public", status="Success")
+        assert event.event_status() == "success"
+
+    def test_event_status_failure(self):
+        event = SeedCompleted(seed_name="countries", schema_name="public", status="Failure")
+        assert event.event_status() == "error"
+
+
+# ── P2: Job scheduler ──────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestJobCreated:
+    @pytest.fixture()
+    def event(self):
+        return JobCreated(job_name="nightly_sync", environment_name="production")
+
+    def test_code(self, event):
+        assert event.code() == "U008"
+
+    def test_message(self, event):
+        assert event.message() == 'Job "nightly_sync" created for environment "production"'
+
+    def test_title(self, event):
+        assert event.title() == 'Job "nightly_sync" created'
+
+    def test_subtitle(self, event):
+        assert event.subtitle() == "Environment: production"
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestJobUpdated:
+    @pytest.fixture()
+    def event(self):
+        return JobUpdated(job_name="nightly_sync")
+
+    def test_code(self, event):
+        assert event.code() == "U009"
+
+    def test_message(self, event):
+        assert event.message() == 'Job "nightly_sync" updated'
+
+    def test_title(self, event):
+        assert event.title() == 'Job "nightly_sync" updated'
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestJobDeleted:
+    @pytest.fixture()
+    def event(self):
+        return JobDeleted(job_name="nightly_sync")
+
+    def test_code(self, event):
+        assert event.code() == "U010"
+
+    def test_message(self, event):
+        assert event.message() == 'Job "nightly_sync" deleted'
+
+    def test_event_status(self, event):
+        assert event.event_status() == "warning"
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestJobTriggered:
+    def test_code(self):
+        event = JobTriggered(job_name="nightly_sync", scope="job")
+        assert event.code() == "U011"
+
+    def test_message_full_job(self):
+        event = JobTriggered(job_name="nightly_sync", scope="job")
+        assert "all models" in event.message()
+
+    def test_message_single_model(self):
+        event = JobTriggered(job_name="nightly_sync", scope="orders")
+        assert "model orders" in event.message()
+
+    def test_title(self):
+        event = JobTriggered(job_name="nightly_sync", scope="job")
+        assert event.title() == 'Job "nightly_sync" triggered'
+
+    def test_subtitle_full_job(self):
+        event = JobTriggered(job_name="nightly_sync", scope="job")
+        sub = event.subtitle()
+        assert "All models" in sub
+        assert "Manual trigger" in sub
+
+    def test_subtitle_single_model(self):
+        event = JobTriggered(job_name="nightly_sync", scope="orders")
+        sub = event.subtitle()
+        assert "Model: orders" in sub
+
+    def test_event_status(self):
+        event = JobTriggered(job_name="nightly_sync", scope="job")
+        assert event.event_status() == "running"
+
+
+# ── P3: Model/file CRUD ────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestModelCreated:
+    @pytest.fixture()
+    def event(self):
+        return ModelCreated(model_name="orders")
+
+    def test_code(self, event):
+        assert event.code() == "U012"
+
+    def test_message(self, event):
+        assert event.message() == 'Model "orders" created'
+
+    def test_title(self, event):
+        assert event.title() == 'Model "orders" created'
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestFileDeleted:
+    @pytest.fixture()
+    def event(self):
+        return FileDeleted(file_names="orders.sql, users.sql")
+
+    def test_code(self, event):
+        assert event.code() == "U013"
+
+    def test_message(self, event):
+        assert event.message() == "Deleted: orders.sql, users.sql"
+
+    def test_event_status(self, event):
+        assert event.event_status() == "warning"
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestFileRenamed:
+    @pytest.fixture()
+    def event(self):
+        return FileRenamed(old_name="orders_v1", new_name="orders_v2")
+
+    def test_code(self, event):
+        assert event.code() == "U014"
+
+    def test_message(self, event):
+        assert event.message() == 'Renamed "orders_v1" → "orders_v2"'
+
+    def test_title(self, event):
+        assert event.title() == "File renamed"
+
+    def test_subtitle(self, event):
+        assert event.subtitle() == '"orders_v1" → "orders_v2"'
+
+
+# ── P4: Connections ─────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestConnectionCreated:
+    @pytest.fixture()
+    def event(self):
+        return ConnectionCreated(connection_name="prod_pg", datasource="postgres")
+
+    def test_code(self, event):
+        assert event.code() == "U015"
+
+    def test_message(self, event):
+        assert event.message() == "Connection created (postgres)"
+
+    def test_title(self, event):
+        assert event.title() == "Connection created"
+
+    def test_subtitle(self, event):
+        assert event.subtitle() == "Datasource: postgres"
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestConnectionTested:
+    def test_code(self):
+        event = ConnectionTested(datasource="postgres", result="success")
+        assert event.code() == "U016"
+
+    def test_message(self):
+        event = ConnectionTested(datasource="postgres", result="success")
+        assert event.message() == "Connection test: success (postgres)"
+
+    def test_title_success(self):
+        event = ConnectionTested(datasource="postgres", result="success")
+        assert event.title() == "Connection test passed"
+
+    def test_title_failure(self):
+        event = ConnectionTested(datasource="postgres", result="failed")
+        assert event.title() == "Connection test failed"
+
+    def test_subtitle(self):
+        event = ConnectionTested(datasource="snowflake", result="success")
+        assert event.subtitle() == "Datasource: snowflake"
+
+    def test_event_status_success(self):
+        event = ConnectionTested(datasource="postgres", result="success")
+        assert event.event_status() == "success"
+
+    def test_event_status_failure(self):
+        event = ConnectionTested(datasource="postgres", result="failed")
+        assert event.event_status() == "success"  # inherits default, no override
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestConnectionDeletedEvt:
+    @pytest.fixture()
+    def event(self):
+        return ConnectionDeletedEvt(connection_name="prod_pg")
+
+    def test_code(self, event):
+        assert event.code() == "U017"
+
+    def test_message(self, event):
+        assert event.message() == 'Connection "prod_pg" deleted'
+
+    def test_event_status(self, event):
+        assert event.event_status() == "warning"
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestConnectionDeleteFailedEvt:
+    @pytest.fixture()
+    def event(self):
+        return ConnectionDeleteFailedEvt(
+            connection_name="prod_pg",
+            reason="Connection is used by job nightly_sync",
+        )
+
+    def test_code(self, event):
+        assert event.code() == "U020"
+
+    def test_message(self, event):
+        msg = event.message()
+        assert "prod_pg" in msg
+        assert "Connection is used by job nightly_sync" in msg
+
+    def test_message_truncates_long_reason(self):
+        long_reason = "x" * 200
+        event = ConnectionDeleteFailedEvt(connection_name="prod_pg", reason=long_reason)
+        msg = event.message()
+        assert msg.endswith("…")
+
+    def test_title(self, event):
+        assert event.title() == 'Failed to delete connection "prod_pg"'
+
+    def test_subtitle_truncates_long_reason(self):
+        long_reason = "x" * 200
+        event = ConnectionDeleteFailedEvt(connection_name="prod_pg", reason=long_reason)
+        sub = event.subtitle()
+        assert len(sub) <= 151
+        assert sub.endswith("…")
+
+    def test_event_status(self, event):
+        assert event.event_status() == "error"
+
+    def test_level_is_error(self, event):
+        assert event.level_tag() == EventLevel.ERROR
+
+
+# ── P5: Environments ───────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestEnvironmentCreated:
+    @pytest.fixture()
+    def event(self):
+        return EnvironmentCreated(environment_name="staging")
+
+    def test_code(self, event):
+        assert event.code() == "U018"
+
+    def test_message(self, event):
+        assert event.message() == 'Environment "staging" created'
+
+    def test_title(self, event):
+        assert event.title() == 'Environment "staging" created'
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestEnvironmentDeleted:
+    @pytest.fixture()
+    def event(self):
+        return EnvironmentDeleted(environment_name="staging")
+
+    def test_code(self, event):
+        assert event.code() == "U019"
+
+    def test_message(self, event):
+        assert event.message() == 'Environment "staging" deleted'
+
+    def test_event_status(self, event):
+        assert event.event_status() == "warning"
+
+
+# ── Cross-cutting: all events share the UserLevel contract ─────────────
+
+
+ALL_USER_EVENTS = [
+    ModelRunStarted(model_name="m", source_table="s", destination_table="d", materialization="table"),
+    ModelRunSucceeded(model_name="m", duration_seconds=1.0),
+    ModelRunFailed(model_name="m", error="err"),
+    TransformationApplied(model_name="m", transformation_type="sort"),
+    TransformationDeleted(model_name="m", transformation_type="sort"),
+    ModelConfigured(model_name="m", source="s", destination="d"),
+    SeedCompleted(seed_name="s", schema_name="sc", status="Success"),
+    JobCreated(job_name="j", environment_name="e"),
+    JobUpdated(job_name="j"),
+    JobDeleted(job_name="j"),
+    JobTriggered(job_name="j", scope="job"),
+    ModelCreated(model_name="m"),
+    FileDeleted(file_names="f.sql"),
+    FileRenamed(old_name="a", new_name="b"),
+    ConnectionCreated(connection_name="c", datasource="postgres"),
+    ConnectionTested(datasource="postgres", result="success"),
+    ConnectionDeletedEvt(connection_name="c"),
+    ConnectionDeleteFailedEvt(connection_name="c", reason="in use"),
+    EnvironmentCreated(environment_name="e"),
+    EnvironmentDeleted(environment_name="e"),
+]
+
+
+@pytest.mark.unit
+@pytest.mark.minimal_core
+class TestAllUserEventsContract:
+    """Every user event must satisfy the UserLevel contract."""
+
+    @pytest.mark.parametrize("event", ALL_USER_EVENTS, ids=lambda e: type(e).__name__)
+    def test_is_user_level_instance(self, event):
+        assert isinstance(event, UserLevel)
+
+    @pytest.mark.parametrize("event", ALL_USER_EVENTS, ids=lambda e: type(e).__name__)
+    def test_audience_is_user(self, event):
+        assert event.audience() == "user"
+
+    @pytest.mark.parametrize("event", ALL_USER_EVENTS, ids=lambda e: type(e).__name__)
+    def test_code_starts_with_u(self, event):
+        assert event.code().startswith("U")
+
+    @pytest.mark.parametrize("event", ALL_USER_EVENTS, ids=lambda e: type(e).__name__)
+    def test_message_is_non_empty_string(self, event):
+        msg = event.message()
+        assert isinstance(msg, str)
+        assert len(msg) > 0
+
+    @pytest.mark.parametrize("event", ALL_USER_EVENTS, ids=lambda e: type(e).__name__)
+    def test_title_is_non_empty_string(self, event):
+        title = event.title()
+        assert isinstance(title, str)
+        assert len(title) > 0
+
+    @pytest.mark.parametrize("event", ALL_USER_EVENTS, ids=lambda e: type(e).__name__)
+    def test_subtitle_is_string(self, event):
+        sub = event.subtitle()
+        assert isinstance(sub, str)
+
+    @pytest.mark.parametrize("event", ALL_USER_EVENTS, ids=lambda e: type(e).__name__)
+    def test_event_status_is_valid(self, event):
+        valid_statuses = {"running", "success", "error", "warning", "info"}
+        assert event.event_status() in valid_statuses
+
+    @pytest.mark.parametrize("event", ALL_USER_EVENTS, ids=lambda e: type(e).__name__)
+    def test_level_tag_is_event_level(self, event):
+        assert isinstance(event.level_tag(), EventLevel)