feat(030): Dataset Lifecycle Workspace — Stats Bar, split-view, inline-edit, bulk actions

Backend: - Add MetricItem, StatsCounts, ColumnDescriptionUpdate, MetricDescriptionUpdate DTOs - Add metric_count to DatasetItem and DatasetDetailResponse - Add server-side filtering via ?filter=unmapped|mapped|linked|all - Add PUT endpoints for column/metric description inline-edit - Add HTML stripping validation (max 2000 chars, plain text) - Add WebSocket dataset.updated event on task completion - RBAC: plugin:migration:WRITE for mutations, READ for reads - 14 pytest tests (stats, filter, metrics, inline-edit, validation, 502/503) Frontend: - Rewrite +page.svelte as split-view orchestrator (387 lines) - StatsBar.svelte — 4 metric tiles with aria-pressed, server-side filter dispatch - DatasetList.svelte — card grid with progress bars, checkboxes, search - DatasetPreview.svelte — presentational detail panel (container fetches) - ColumnsTable.svelte — inline-edit with type chips, bold/italic fallback - MetricsTable.svelte — inline-edit with expression hints - 52 vitest tests across 7 test files - i18n: 11 EN + 11 RU keys (with_mapping/с_маппингом) Spec: - 15 issues resolved (semantic label, filtering, endpoints, conflict, metric_count, RBAC, ID types, ports, i18n, validation, a11y, propagation, resize, perf, ownership) Status: 14/14 backend tests pass, 52/52 frontend tests pass, build succeeds Risk: Low — T051 browser validation pending (non-blocking)
2026-05-21 13:58:47 +03:00
parent 673c0bb217
commit b529e8082a
32 changed files with 2191 additions and 942 deletions
--- a/backend/src/api/routes/datasets.py
+++ b/backend/src/api/routes/datasets.py
@@ -13,7 +13,9 @@
 # @INVARIANT: All dataset responses include last_task metadata


-from fastapi import APIRouter, Depends, HTTPException
+import re
+
+from fastapi import APIRouter, Depends, HTTPException, Query
 from pydantic import BaseModel, Field

 from ...core.logger import belief_scope, logger
@@ -45,6 +47,7 @@ class DatasetItem(BaseModel):
    database: str
    mapped_fields: MappedFields | None = None
    last_task: LastTask | None = None
+    metric_count: int = 0

    class Config:
        allow_population_by_field_name = True
@@ -69,8 +72,21 @@ class DatasetColumn(BaseModel):
    description: str | None = None
 # #endregion DatasetColumn

-# #region DatasetDetailResponse [C:1] [TYPE DataClass]
-# @BRIEF Detailed DTO for a dataset including columns and links
+# #region MetricItem [C:1] [TYPE DataClass]
+# @BRIEF Pydantic DTO for a dataset metric — carries Superset metric metadata.
+# @RELATION DEPENDS_ON -> [DatasetDetailResponse]
+class MetricItem(BaseModel):
+    id: int
+    metric_name: str
+    expression: str | None = None
+    verbose_name: str | None = None
+    description: str | None = None
+    metric_type: str | None = None
+# #endregion MetricItem
+
+# #region DatasetDetailResponse [C:2] [TYPE DataClass]
+# @BRIEF Detailed DTO for a dataset including columns, linked dashboards, and metrics.
+# @RELATION DEPENDS_ON -> [MetricItem]
 class DatasetDetailResponse(BaseModel):
    id: int
    table_name: str | None = None
@@ -79,6 +95,8 @@ class DatasetDetailResponse(BaseModel):
    description: str | None = None
    columns: list[DatasetColumn]
    column_count: int
+    metrics: list[MetricItem] = []
+    metric_count: int = 0
    sql: str | None = None
    linked_dashboards: list[LinkedDashboard]
    linked_dashboard_count: int
@@ -90,10 +108,21 @@ class DatasetDetailResponse(BaseModel):
        allow_population_by_field_name = True
 # #endregion DatasetDetailResponse

-# #region DatasetsResponse [C:1] [TYPE DataClass]
-# @BRIEF Paginated response DTO for dataset listings
+# #region StatsCounts [C:1] [TYPE DataClass]
+# @BRIEF Aggregate statistics for the Stats Bar — computed from full dataset list.
+class StatsCounts(BaseModel):
+    total: int
+    unmapped_count: int
+    mapped_count: int
+    linked_count: int
+# #endregion StatsCounts
+
+# #region DatasetsResponse [C:2] [TYPE DataClass]
+# @BRIEF Paginated response DTO for dataset listings with stats.
+# @RELATION DEPENDS_ON -> [StatsCounts]
 class DatasetsResponse(BaseModel):
    datasets: list[DatasetItem]
+    stats: StatsCounts
    total: int
    page: int
    page_size: int
@@ -106,6 +135,18 @@ class TaskResponse(BaseModel):
    task_id: str
 # #endregion TaskResponse

+# #region ColumnDescriptionUpdate [C:1] [TYPE DataClass]
+# @BRIEF Request DTO for inline-edit of a column description.
+class ColumnDescriptionUpdate(BaseModel):
+    description: str
+# #endregion ColumnDescriptionUpdate
+
+# #region MetricDescriptionUpdate [C:1] [TYPE DataClass]
+# @BRIEF Request DTO for inline-edit of a metric description.
+class MetricDescriptionUpdate(BaseModel):
+    description: str
+# #endregion MetricDescriptionUpdate
+
 # #region get_dataset_ids [C:4] [TYPE Function]
 # @BRIEF Fetch list of all dataset IDs from a specific environment (without pagination)
 # @PRE: env_id must be a valid environment ID
@@ -155,17 +196,19 @@ async def get_dataset_ids(
 # #endregion get_dataset_ids

 # #region get_datasets [C:4] [TYPE Function]
-# @BRIEF Fetch list of datasets from a specific environment with mapping progress
+# @BRIEF Fetch list of datasets from a specific environment with mapping progress and stats.
 # @PRE: env_id must be a valid environment ID
 # @PRE: page must be >= 1 if provided
 # @PRE: page_size must be between 1 and 100 if provided
-# @POST: Returns a list of datasets with enhanced metadata and pagination info
-# @POST: Response includes pagination metadata (page, page_size, total, total_pages)
+# @POST: Returns a list of datasets with enhanced metadata, pagination info, and StatsCounts.
+# @POST: Response includes pagination metadata (page, page_size, total, total_pages) and stats object.
+# @RATIONALE Stats counts returned in the same response as datasets to avoid an extra API call for the Stats Bar (FR-022).
 # @RELATION CALLS -> [get_datasets_with_status]
@router.get("", response_model=DatasetsResponse)
 async def get_datasets(
    env_id: str,
    search: str | None = None,
+    filter: str | None = None,
    page: int = 1,
    page_size: int = 10,
    config_manager=Depends(get_config_manager),
@@ -173,7 +216,7 @@ async def get_datasets(
    resource_service=Depends(get_resource_service),
    _ = Depends(has_permission("plugin:migration", "READ"))
 ):
-    with belief_scope("get_datasets", f"env_id={env_id}, search={search}, page={page}, page_size={page_size}"):
+    with belief_scope("get_datasets", f"env_id={env_id}, search={search}, filter={filter}, page={page}, page_size={page_size}"):
        # Validate pagination parameters
        if page < 1:
            logger.error(f"[get_datasets][Coherence:Failed] Invalid page: {page}")
@@ -196,6 +239,45 @@ async def get_datasets(
            # Fetch datasets with status using ResourceService
            datasets = await resource_service.get_datasets_with_status(env, all_tasks)

+            # Compute stats from full list BEFORE filtering/pagination
+            total_datasets = len(datasets)
+            unmapped = sum(
+                1 for d in datasets
+                if d.get("mapped_fields") and d["mapped_fields"].get("mapped", 0) == 0
+            )
+            mapped = sum(
+                1 for d in datasets
+                if d.get("mapped_fields") and d["mapped_fields"].get("mapped", 0) > 0
+            )
+            linked = sum(
+                1 for d in datasets
+                if d.get("linked_dashboard_count", 0) > 0
+            )
+            stats = StatsCounts(
+                total=total_datasets,
+                unmapped_count=unmapped,
+                mapped_count=mapped,
+                linked_count=linked,
+            )
+
+            # Apply server-side filter (before pagination) — FR-026
+            if filter and filter in ("unmapped", "mapped", "linked"):
+                if filter == "unmapped":
+                    datasets = [
+                        d for d in datasets
+                        if d.get("mapped_fields") and d["mapped_fields"].get("mapped", 0) == 0
+                    ]
+                elif filter == "mapped":
+                    datasets = [
+                        d for d in datasets
+                        if d.get("mapped_fields") and d["mapped_fields"].get("mapped", 0) > 0
+                    ]
+                elif filter == "linked":
+                    datasets = [
+                        d for d in datasets
+                        if d.get("linked_dashboard_count", 0) > 0
+                    ]
+
            # Apply search filter if provided
            if search:
                search_lower = search.lower()
@@ -213,10 +295,11 @@ async def get_datasets(
            # Slice datasets for current page
            paginated_datasets = datasets[start_idx:end_idx]

-            logger.info(f"[get_datasets][Coherence:OK] Returning {len(paginated_datasets)} datasets (page {page}/{total_pages}, total: {total})")
+            logger.info(f"[get_datasets][Coherence:OK] Returning {len(paginated_datasets)} datasets (page {page}/{total_pages}, total: {total}, stats: total={stats.total} unmapped={stats.unmapped_count} mapped={stats.mapped_count} linked={stats.linked_count})")

            return DatasetsResponse(
                datasets=paginated_datasets,
+                stats=stats,
                total=total,
                page=page,
                page_size=page_size,
@@ -399,4 +482,194 @@ async def get_dataset_detail(
            raise HTTPException(status_code=503, detail=f"Failed to fetch dataset detail: {e!s}")
 # #endregion get_dataset_detail

+# #region _strip_html_tags [C:2] [TYPE Function] [SEMANTICS helper,validation,html]
+# @BRIEF Detect and strip HTML tags from a text string. Uses simple regex <[^>]*>.
+#        If HTML is detected, strips tags and logs a warning. Returns cleaned text.
+# @REJECTED Reject-with-400 approach was rejected in favour of silent stripping — more user-friendly
+#           and matches the UX expectation that inline-edited descriptions are plain text only.
+def _strip_html_tags(text: str) -> str:
+    if not text:
+        return text
+    if re.search(r"<[^>]*>", text):
+        cleaned = re.sub(r"<[^>]*>", "", text)
+        logger.warning(f"[_strip_html_tags] HTML tags detected and stripped from description")
+        return cleaned
+    return text
+# #endregion _strip_html_tags
+
+# #region update_column_description [C:4] [TYPE Endpoint]
+# @BRIEF Save description for a single dataset column. Internal: GET full dataset from Superset, modify one column's description, PUT back.
+# @PRE: dataset_id and column_id must exist in the target environment.
+# @POST: Column description in Superset is updated. Response confirms success.
+# @SIDE_EFFECT: Mutates dataset metadata in upstream Superset instance via PUT.
+# @VALIDATION: description — string, max 2000 chars, plain text only (HTML stripped), may be empty string to clear description, no trimming applied. 404 if dataset or column not found. 502 if Superset upstream fails.
+# @ERROR: 400 — description exceeds max length or contains non-plaintext content. 404 — dataset_id or column_id not found. 502 — Superset upstream failure (GET or PUT).
+# @RELATION CALLS -> [SupersetClient.get_dataset]
+# @RELATION CALLS -> [SupersetClient.update_dataset]
+# @RATIONALE Must perform GET→modify→PUT because Superset has no PATCH for individual columns — only full object PUT with override_columns=false.
+# @REJECTED Direct PUT from frontend — rejected because frontend would need to handle full Superset payload structure.
+@router.put("/{dataset_id}/columns/{column_id}/description")
+async def update_column_description(
+    dataset_id: int,
+    column_id: int,
+    env_id: str = Query(..., description="Environment ID"),
+    body: ColumnDescriptionUpdate = ...,
+    config_manager=Depends(get_config_manager),
+    _ = Depends(has_permission("plugin:migration", "WRITE"))
+):
+    with belief_scope("update_column_description", f"dataset_id={dataset_id}, column_id={column_id}, env_id={env_id}"):
+        # Validate and sanitize description
+        description = _strip_html_tags(body.description)
+        if len(description) > 2000:
+            logger.error(f"[update_column_description][Coherence:Failed] Description exceeds 2000 chars: {len(description)}")
+            raise HTTPException(status_code=400, detail="Description must not exceed 2000 characters")
+
+        # Validate environment
+        environments = config_manager.get_environments()
+        env = next((e for e in environments if e.id == env_id), None)
+        if not env:
+            logger.error(f"[update_column_description][Coherence:Failed] Environment not found: {env_id}")
+            raise HTTPException(status_code=404, detail="Environment not found")
+
+        try:
+            client = SupersetClient(env)
+
+            # GET full dataset
+            logger.reason("Fetching full dataset from Superset for column description update",
+                          extra={"src": "update_column_description"})
+            try:
+                response = client.get_dataset(dataset_id)
+            except Exception as e:
+                logger.error(f"[update_column_description][Coherence:Failed] Superset GET failed: {e}")
+                raise HTTPException(status_code=502, detail=f"Failed to fetch dataset from Superset: {e!s}")
+
+            if isinstance(response, dict) and "result" in response:
+                dataset = response["result"]
+            else:
+                dataset = response
+
+            if not dataset or not dataset.get("id"):
+                logger.error(f"[update_column_description][Coherence:Failed] Dataset {dataset_id} not found")
+                raise HTTPException(status_code=404, detail=f"Dataset {dataset_id} not found")
+
+            # Find and modify the target column
+            columns = dataset.get("columns", [])
+            found = False
+            for col in columns:
+                if col.get("id") == column_id:
+                    col["description"] = description
+                    found = True
+                    break
+
+            if not found:
+                logger.error(f"[update_column_description][Coherence:Failed] Column {column_id} not found in dataset {dataset_id}")
+                raise HTTPException(status_code=404, detail=f"Column {column_id} not found in dataset {dataset_id}")
+
+            # PUT back with override_columns=false
+            put_payload = dataset
+            try:
+                client.update_dataset(dataset_id, put_payload, override_columns=False)
+            except Exception as e:
+                logger.error(f"[update_column_description][Coherence:Failed] Superset PUT failed: {e}")
+                raise HTTPException(status_code=502, detail=f"Failed to update dataset in Superset: {e!s}")
+
+            logger.reflect(
+                f"Updated column {column_id} description for dataset {dataset_id}",
+                extra={"src": "update_column_description"},
+            )
+            return {"success": True, "description": description}
+
+        except HTTPException:
+            raise
+        except Exception as e:
+            logger.error(f"[update_column_description][Coherence:Failed] Unexpected error: {e}")
+            raise HTTPException(status_code=500, detail=f"Unexpected error: {e!s}")
+# #endregion update_column_description
+
+# #region update_metric_description [C:4] [TYPE Endpoint]
+# @BRIEF Save description for a single dataset metric. Mirror of update_column_description for metrics.
+# @PRE: dataset_id and metric_id must exist in the target environment.
+# @POST: Metric description in Superset is updated.
+# @SIDE_EFFECT: Mutates dataset metadata in upstream Superset instance via PUT.
+# @VALIDATION: description — string, max 2000 chars, plain text only (HTML stripped), may be empty string to clear description, no trimming applied. 404 if dataset or metric not found. 502 if Superset upstream fails.
+# @ERROR: 400 — description exceeds max length or contains non-plaintext content. 404 — dataset_id or metric_id not found. 502 — Superset upstream failure (GET or PUT).
+# @RELATION CALLS -> [SupersetClient.get_dataset]
+# @RELATION CALLS -> [SupersetClient.update_dataset]
+@router.put("/{dataset_id}/metrics/{metric_id}/description")
+async def update_metric_description(
+    dataset_id: int,
+    metric_id: int,
+    env_id: str = Query(..., description="Environment ID"),
+    body: MetricDescriptionUpdate = ...,
+    config_manager=Depends(get_config_manager),
+    _ = Depends(has_permission("plugin:migration", "WRITE"))
+):
+    with belief_scope("update_metric_description", f"dataset_id={dataset_id}, metric_id={metric_id}, env_id={env_id}"):
+        # Validate and sanitize description
+        description = _strip_html_tags(body.description)
+        if len(description) > 2000:
+            logger.error(f"[update_metric_description][Coherence:Failed] Description exceeds 2000 chars: {len(description)}")
+            raise HTTPException(status_code=400, detail="Description must not exceed 2000 characters")
+
+        # Validate environment
+        environments = config_manager.get_environments()
+        env = next((e for e in environments if e.id == env_id), None)
+        if not env:
+            logger.error(f"[update_metric_description][Coherence:Failed] Environment not found: {env_id}")
+            raise HTTPException(status_code=404, detail="Environment not found")
+
+        try:
+            client = SupersetClient(env)
+
+            # GET full dataset
+            logger.reason("Fetching full dataset from Superset for metric description update",
+                          extra={"src": "update_metric_description"})
+            try:
+                response = client.get_dataset(dataset_id)
+            except Exception as e:
+                logger.error(f"[update_metric_description][Coherence:Failed] Superset GET failed: {e}")
+                raise HTTPException(status_code=502, detail=f"Failed to fetch dataset from Superset: {e!s}")
+
+            if isinstance(response, dict) and "result" in response:
+                dataset = response["result"]
+            else:
+                dataset = response
+
+            if not dataset or not dataset.get("id"):
+                logger.error(f"[update_metric_description][Coherence:Failed] Dataset {dataset_id} not found")
+                raise HTTPException(status_code=404, detail=f"Dataset {dataset_id} not found")
+
+            # Find and modify the target metric
+            metrics = dataset.get("metrics", [])
+            found = False
+            for metric in metrics:
+                if metric.get("id") == metric_id:
+                    metric["description"] = description
+                    found = True
+                    break
+
+            if not found:
+                logger.error(f"[update_metric_description][Coherence:Failed] Metric {metric_id} not found in dataset {dataset_id}")
+                raise HTTPException(status_code=404, detail=f"Metric {metric_id} not found in dataset {dataset_id}")
+
+            # PUT back with override_columns=false
+            try:
+                client.update_dataset(dataset_id, dataset, override_columns=False)
+            except Exception as e:
+                logger.error(f"[update_metric_description][Coherence:Failed] Superset PUT failed: {e}")
+                raise HTTPException(status_code=502, detail=f"Failed to update dataset in Superset: {e!s}")
+
+            logger.reflect(
+                f"Updated metric {metric_id} description for dataset {dataset_id}",
+                extra={"src": "update_metric_description"},
+            )
+            return {"success": True, "description": description}
+
+        except HTTPException:
+            raise
+        except Exception as e:
+            logger.error(f"[update_metric_description][Coherence:Failed] Unexpected error: {e}")
+            raise HTTPException(status_code=500, detail=f"Unexpected error: {e!s}")
+# #endregion update_metric_description
+
 # #endregion DatasetsApi
--- a/backend/src/app.py
+++ b/backend/src/app.py
@@ -393,6 +393,33 @@ async def websocket_endpoint(
                extra={"task_id": task_id},
            )
 # #endregion websocket_endpoint
+# #region dataset_websocket_endpoint [C:4] [TYPE Function]
+# @BRIEF WebSocket endpoint for dataset.updated events — auto-refresh on task completion.
+# @RELATION CALLS -> [TaskManagerPackage]
+# @PRE env_id must reference a known environment.
+# @POST WebSocket streams dataset.updated events until disconnect.
+# @SIDE_EFFECT Subscribes to dataset event queue in task manager lifecycle.
+@app.websocket("/ws/datasets/{env_id}")
+async def dataset_websocket_endpoint(websocket: WebSocket, env_id: str):
+    seed_trace_id()
+    with belief_scope("dataset_websocket_endpoint", f"env_id={env_id}"):
+        await websocket.accept()
+        logger.reason("Accepted dataset event WebSocket", extra={"env_id": env_id})
+        task_manager = get_task_manager()
+        queue = await task_manager.subscribe_dataset_events(env_id)
+        try:
+            while True:
+                event = await queue.get()
+                await websocket.send_json(event)
+                logger.reflect("Forwarded dataset.updated event to client", extra={"env_id": env_id})
+        except WebSocketDisconnect:
+            logger.reason("WebSocket client disconnected from dataset events", extra={"env_id": env_id})
+        except Exception as exc:
+            logger.explore("WebSocket dataset streaming failed", extra={"env_id": env_id, "error": str(exc)})
+        finally:
+            task_manager.unsubscribe_dataset_events(env_id, queue)
+            logger.reflect("Released dataset event subscription", extra={"env_id": env_id})
+# #endregion dataset_websocket_endpoint
 # #region StaticFiles [C:1] [TYPE Mount] [SEMANTICS static, frontend, spa]
 # @BRIEF Mounts the frontend build directory to serve static assets.
 frontend_path = project_root / "frontend" / "build"
--- a/backend/src/core/superset_client/_datasets.py
+++ b/backend/src/core/superset_client/_datasets.py
@@ -86,6 +86,22 @@ class SupersetDatasetsMixin:
                        "description": col.get("description", ""),
                    }
                )
+            metrics = dataset.get("metrics", [])
+            metric_info = []
+            for metric in metrics:
+                metric_id = metric.get("id")
+                if metric_id is None:
+                    continue
+                metric_info.append(
+                    {
+                        "id": int(metric_id),
+                        "metric_name": metric.get("metric_name"),
+                        "expression": metric.get("expression"),
+                        "verbose_name": metric.get("verbose_name"),
+                        "description": metric.get("description", ""),
+                        "metric_type": metric.get("metric_type"),
+                    }
+                )
            linked_dashboards = []
            try:
                related_objects = self.network.request(
@@ -139,6 +155,8 @@ class SupersetDatasetsMixin:
                "description": dataset.get("description", ""),
                "columns": column_info,
                "column_count": len(column_info),
+                "metrics": metric_info,
+                "metric_count": len(metric_info),
                "sql": sql,
                "linked_dashboards": linked_dashboards,
                "linked_dashboard_count": len(linked_dashboards),
@@ -147,7 +165,7 @@ class SupersetDatasetsMixin:
                "changed_on": dataset.get("changed_on"),
            }
            app_logger.reflect(
-                f"Got dataset {dataset_id} with {len(column_info)} columns and {len(linked_dashboards)} linked dashboards",
+                f"Got dataset {dataset_id} with {len(column_info)} columns, {len(metric_info)} metrics, and {len(linked_dashboards)} linked dashboards",
                extra={"src": "get_dataset_detail"},
            )
            return result
@@ -169,12 +187,13 @@ class SupersetDatasetsMixin:
    # @PURPOSE: Обновляет данные датасета по его ID.
    # @SIDE_EFFECT: Modifies resource in upstream Superset environment.
    # @RELATION: CALLS -> [APIClient]
-    def update_dataset(self, dataset_id: int, data: dict) -> dict:
+    def update_dataset(self, dataset_id: int, data: dict, override_columns: bool = False) -> dict:
        with belief_scope("SupersetClient.update_dataset", f"id={dataset_id}"):
            app_logger.info("[update_dataset][Enter] Updating dataset %s.", dataset_id)
+            endpoint = f"/dataset/{dataset_id}?override_columns={'true' if override_columns else 'false'}"
            response = self.network.request(
                method="PUT",
-                endpoint=f"/dataset/{dataset_id}",
+                endpoint=endpoint,
                data=json.dumps(data),
                headers={"Content-Type": "application/json"},
            )
--- a/backend/src/core/task_manager/lifecycle.py
+++ b/backend/src/core/task_manager/lifecycle.py
@@ -71,6 +71,7 @@ class JobLifecycle:
        self.persistence_service = persistence_service
        self.executor = executor
        self.loop = loop
+        self._dataset_subscribers: dict[str, list[asyncio.Queue]] = {}
    # #endregion __init__

    # #region create_task [C:4] [TYPE Function] [SEMANTICS task,create,persist,schedule]
@@ -190,6 +191,12 @@ class JobLifecycle:
                    "Task lifecycle reached persisted terminal state",
                    extra={"task_id": task_id, "status": str(task.status)},
                )
+                # Broadcast dataset.updated for mapping and documentation tasks (FR-024, R4)
+                if task.plugin_id in ("dataset-mapper", "llm_documentation"):
+                    dataset_ids = task.params.get("dataset_ids") or [task.params.get("dataset_id")]
+                    env_id = task.params.get("env") or task.params.get("environment_id", "")
+                    if dataset_ids and env_id:
+                        self._broadcast_dataset_updated(env_id, dataset_ids)
    # #endregion _run_task

    # #region resolve_task [C:3] [TYPE Function] [SEMANTICS task,resolve,mapping,resume]
@@ -309,5 +316,37 @@ class JobLifecycle:
                )
            self.graph.resolve_future(task_id, True)
    # #endregion resume_task_with_password
+
+    # #region subscribe_dataset_events [C:2] [TYPE Function] [SEMANTICS dataset,subscribe,events,websocket]
+    # @BRIEF Subscribe to dataset.updated events for a specific environment.
+    async def subscribe_dataset_events(self, env_id: str) -> asyncio.Queue:
+        queue = asyncio.Queue()
+        if env_id not in self._dataset_subscribers:
+            self._dataset_subscribers[env_id] = []
+        self._dataset_subscribers[env_id].append(queue)
+        return queue
+    # #endregion subscribe_dataset_events
+
+    # #region unsubscribe_dataset_events [C:2] [TYPE Function] [SEMANTICS dataset,unsubscribe,events]
+    # @BRIEF Unsubscribe from dataset.updated events for a specific environment.
+    def unsubscribe_dataset_events(self, env_id: str, queue: asyncio.Queue):
+        if env_id in self._dataset_subscribers:
+            if queue in self._dataset_subscribers[env_id]:
+                self._dataset_subscribers[env_id].remove(queue)
+            if not self._dataset_subscribers[env_id]:
+                del self._dataset_subscribers[env_id]
+    # #endregion unsubscribe_dataset_events
+
+    # #region _broadcast_dataset_updated [C:3] [TYPE Function] [SEMANTICS dataset,broadcast,event,websocket]
+    # @BRIEF Broadcast dataset.updated event to all subscribers for a given environment.
+    def _broadcast_dataset_updated(self, env_id: str, dataset_ids: list[int]):
+        event = {"type": "dataset.updated", "payload": {"env_id": env_id, "dataset_ids": dataset_ids}}
+        if env_id in self._dataset_subscribers:
+            for queue in self._dataset_subscribers[env_id]:
+                try:
+                    self.loop.call_soon_threadsafe(queue.put_nowait, event)
+                except Exception:
+                    pass
+    # #endregion _broadcast_dataset_updated
 # #endregion JobLifecycle
 # #endregion JobLifecycleModule
--- a/backend/src/core/task_manager/manager.py
+++ b/backend/src/core/task_manager/manager.py
@@ -318,5 +318,19 @@ class TaskManager:
            add_log_callback=self._add_log,
        )
    # #endregion resume_task_with_password
+
+    # ── Dataset event delegates to JobLifecycle ──
+
+    # #region subscribe_dataset_events [TYPE Function] [C:2]
+    # @BRIEF Subscribe to dataset.updated events for an environment.
+    async def subscribe_dataset_events(self, env_id: str) -> asyncio.Queue:
+        return await self.lifecycle.subscribe_dataset_events(env_id)
+    # #endregion subscribe_dataset_events
+
+    # #region unsubscribe_dataset_events [TYPE Function] [C:2]
+    # @BRIEF Unsubscribe from dataset.updated events.
+    def unsubscribe_dataset_events(self, env_id: str, queue: asyncio.Queue):
+        self.lifecycle.unsubscribe_dataset_events(env_id, queue)
+    # #endregion unsubscribe_dataset_events
 # #endregion TaskManager
 # #endregion TaskManagerModule