правки по задачам

backend/alembic/versions/0019_task_results_index.py

@@ -0,0 +1,25 @@
+"""0019 task results composite index
+
+Revision ID: 0019
+Revises: 0018
+Create Date: 2026-05-06
+"""
+
+from alembic import op
+
+revision = "0019"
+down_revision = "0018"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_index(
+        "ix_task_results_task_device",
+        "task_results",
+        ["task_id", "device_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_task_results_task_device", table_name="task_results")

backend/app/routers/alerts.py

@@ -91,6 +91,42 @@ async def list_alerts(
     return [_to_read(a) for a in result.scalars().all()]
 
 
+@router.post("/acknowledge-by-object", response_model=list[AlertRead])
+async def acknowledge_by_object(
+    object_id: int = Query(...),
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    result = await db.execute(
+        select(Alert)
+        .join(Device, Alert.device_id == Device.id)
+        .where(Device.object_id == object_id, Alert.status == "open")
+    )
+    alerts = result.scalars().all()
+    now = datetime.now(timezone.utc)
+    for alert in alerts:
+        alert.status = "acknowledged"
+        alert.acknowledged_at = now
+        alert.acknowledged_by = current_user.id
+    await db.flush()
+
+    alert_ids = [a.id for a in alerts]
+    loaded = []
+    for aid in alert_ids:
+        loaded.append(await _load_alert(db, aid))
+
+    for alert in loaded:
+        await sse_manager.publish(
+            SSEEvent(
+                type="alert_updated",
+                task_id=MONITOR_CHANNEL,
+                data={"alert_id": alert.id, "status": "acknowledged", "device_id": alert.device_id},
+            )
+        )
+
+    return [_to_read(a) for a in loaded]
+
+
 @router.post("/{alert_id}/acknowledge", response_model=AlertRead)
 async def acknowledge_alert(
     alert_id: int,

backend/app/services/dashboard_monitor.py

@@ -549,6 +549,19 @@ async def run_dashboard_monitor_cycle(
                             ", ".join(ips[:10]) + ("…" if len(ips) > 10 else ""),
                         )
 
+                    # Загружаем предыдущие ping-статусы до запуска — нужны для retry-логики
+                    pre_ping_status_map: dict[int, str] = {}
+                    if check.check_key == "ping":
+                        pre_state_rows = (
+                            await db.execute(
+                                select(DashboardDeviceLiveState).where(
+                                    DashboardDeviceLiveState.check_key == "ping",
+                                    DashboardDeviceLiveState.device_id.in_([d.id for d in due_devices]),
+                                )
+                            )
+                        ).scalars().all()
+                        pre_ping_status_map = {row.device_id: row.status for row in pre_state_rows}
+
                     sem = asyncio.Semaphore(max(1, check.concurrency))
                     tasks = [
                         asyncio.ensure_future(_run_check_on_device(sem, check, device, device.object))
@@ -589,6 +602,78 @@ async def run_dashboard_monitor_cycle(
                                 )
                             )
 
+                    # Retry для offline-устройств (только ping):
+                    # срабатывает только если устройство было online (переход online→offline)
+                    if check.check_key == "ping":
+                        offline_devices = [
+                            r[0] for r in results
+                            if r[2] == "offline" and pre_ping_status_map.get(r[0].id) == "online"
+                        ]
+                        for retry_num in range(1, 3):
+                            if not offline_devices:
+                                break
+                            logger.info(
+                                "Dashboard monitor [ping] retry %d/2: %d offline устройств, ждём 60 сек",
+                                retry_num, len(offline_devices),
+                            )
+                            await sse_manager.publish(
+                                SSEEvent(
+                                    type="dashboard_monitor_update",
+                                    task_id=DASHBOARD_MONITOR_CHANNEL,
+                                    data={
+                                        "check_key": check.check_key,
+                                        "processed": completed_count,
+                                        "due": len(due_devices),
+                                        "failed": failed_count,
+                                        "retry": retry_num,
+                                        "retry_pending": len(offline_devices),
+                                        "ts": datetime.now(timezone.utc).isoformat(),
+                                    },
+                                )
+                            )
+                            await asyncio.sleep(60)
+
+                            retry_tasks = [
+                                asyncio.ensure_future(_run_check_on_device(sem, check, device, device.object))
+                                for device in offline_devices
+                            ]
+                            retry_results = []
+                            for coro in asyncio.as_completed(retry_tasks):
+                                r = await coro
+                                retry_results.append(r)
+                                logger.debug(
+                                    "Dashboard monitor [ping] retry %d %s: %s",
+                                    retry_num, r[0].ip, r[2],
+                                )
+
+                            # Обновляем results: заменяем результат устройства если оно восстановилось
+                            retry_map = {r[0].id: r for r in retry_results}
+                            results = [
+                                retry_map.get(r[0].id, r) for r in results
+                            ]
+                            offline_devices = [r[0] for r in retry_results if r[2] == "offline"]
+                            recovered = len(retry_results) - len(offline_devices)
+                            failed_count = sum(1 for r in results if not r[1])
+                            logger.info(
+                                "Dashboard monitor [ping] retry %d/2 done: восстановилось %d, всё ещё offline %d",
+                                retry_num, recovered, len(offline_devices),
+                            )
+                            await sse_manager.publish(
+                                SSEEvent(
+                                    type="dashboard_monitor_update",
+                                    task_id=DASHBOARD_MONITOR_CHANNEL,
+                                    data={
+                                        "check_key": check.check_key,
+                                        "processed": len(due_devices),
+                                        "due": len(due_devices),
+                                        "failed": failed_count,
+                                        "retry": retry_num,
+                                        "retry_recovered": recovered,
+                                        "ts": datetime.now(timezone.utc).isoformat(),
+                                    },
+                                )
+                            )
+
                     async with AsyncSessionLocal() as dbw:
                         async with dbw.begin():
                             # Загружаем предыдущие статусы и baseline только для ping

backend/app/services/task.py

@@ -87,7 +87,6 @@ async def _resolve_scope(scope: dict) -> list[tuple[Device, Object]]:
                 query = query.where(Device.object_id == scope["object_id"])
         elif scope_type == "filter":
             # Cross-object filter scope (from WorkDevices global page)
-            from app.models.object import Object as Obj
             if scope.get("category"):
                 cats = scope["category"]
                 if isinstance(cats, list):
@@ -96,10 +95,12 @@ async def _resolve_scope(scope: dict) -> list[tuple[Device, Object]]:
                     query = query.where(Device.category == cats)
             if scope.get("object_id"):
                 query = query.where(Device.object_id == scope["object_id"])
-            if scope.get("city"):
+            if scope.get("city") or scope.get("client"):
-                query = query.where(Obj.city == scope["city"])
+                query = query.join(Object, Device.object_id == Object.id)
-            if scope.get("client"):
+                if scope.get("city"):
-                query = query.where(Obj.client == scope["client"])
+                    query = query.where(Object.city == scope["city"])
+                if scope.get("client"):
+                    query = query.where(Object.client == scope["client"])
         else:
             return []
 
@@ -117,20 +118,14 @@ async def _run_on_device(
     sem: asyncio.Semaphore,
 ) -> bool | None:
     async with sem:
-        # Bail out after acquiring the semaphore so large target lists don't
+        # Session 1: cancellation check + create TaskResult
-        # stampede the DB pool with cancellation checks.
-        async with AsyncSessionLocal() as db:
-            task_row = await db.get(Task, task_id)
-            if task_row and task_row.cancel_requested:
-                return None  # None = skipped due to cancellation
-
-        emit = lambda msg, level="info": _emit(task_id, msg, level)
-
-        # Create task_result row
-        result_id: int | None = None
         started_at = datetime.now(timezone.utc)
+        result_id: int | None = None
         async with AsyncSessionLocal() as db:
             async with db.begin():
+                task_row = await db.get(Task, task_id)
+                if task_row and task_row.cancel_requested:
+                    return None  # None = skipped due to cancellation
                 tr = TaskResult(
                     task_id=task_id,
                     device_id=device.id,
@@ -141,6 +136,8 @@ async def _run_on_device(
                 await db.flush()
                 result_id = tr.id
 
+        emit = lambda msg, level="info": _emit(task_id, msg, level)
+
         try:
             action_result = await action.execute(device, obj, params, emit, db=None)
         except Exception as exc:
@@ -161,6 +158,7 @@ async def _run_on_device(
         finished_at = datetime.now(timezone.utc)
         duration_ms = int((finished_at - started_at).total_seconds() * 1000)
 
+        # Session 2: update TaskResult + device status (ping) in one transaction
         async with AsyncSessionLocal() as db:
             async with db.begin():
                 tr = await db.get(TaskResult, result_id)
@@ -172,11 +170,8 @@ async def _run_on_device(
                     tr.data = {**(action_data or {}), "error": action_error} if action_error else (action_data or None)
                     tr.duration_ms = duration_ms
 
-        # For ping action: update device status, last_seen, last_ping_ms
+                if action.name == "ping":
-        if action.name == "ping":
+                    new_status = "online" if action_result_success else "offline"
-            new_status = "online" if action_result_success else "offline"
-            async with AsyncSessionLocal() as db:
-                async with db.begin():
                     device_row = await db.get(Device, device.id)
                     if device_row:
                         device_row.status = new_status
@@ -186,6 +181,10 @@ async def _run_on_device(
                             ping_ms = (action_data or {}).get("ping_ms")
                             if ping_ms is not None:
                                 device_row.last_ping_ms = ping_ms
+
+        # For ping action: publish SSE device_status after DB commit
+        if action.name == "ping":
+            new_status = "online" if action_result_success else "offline"
             await sse_manager.publish(SSEEvent(
                 type="device_status",
                 task_id="__monitor__",
@@ -217,6 +216,17 @@ async def _run_on_device(
         return action_result_success
 
 
+async def _heartbeat_writer(task_id: str, interval: int = 30) -> None:
+    """Write task.heartbeat_at every `interval` seconds while task is running."""
+    while True:
+        await asyncio.sleep(interval)
+        async with AsyncSessionLocal() as db:
+            async with db.begin():
+                task = await db.get(Task, task_id)
+                if task and task.status == "running":
+                    task.heartbeat_at = datetime.now(timezone.utc)
+
+
 async def _run_task(task_id: str) -> None:
     await _emit(task_id, "Task started", "info")
 
@@ -227,6 +237,7 @@ async def _run_task(task_id: str) -> None:
                 return
             task.status = "running"
             task.started_at = datetime.now(timezone.utc)
+            task.heartbeat_at = datetime.now(timezone.utc)
             scope = dict(task.target_scope)
             action_type = task.type
             params = dict(task.params)
@@ -275,10 +286,20 @@ async def _run_task(task_id: str) -> None:
     await _emit(task_id, f"Running on {len(targets)} device(s)...", "info")
     sem = asyncio.Semaphore(_TASK_DEVICE_CONCURRENCY)
 
-    results = await asyncio.gather(
+    # Start heartbeat writer
-        *[_run_on_device(task_id, device, obj, action, params, sem) for device, obj in targets],
+    heartbeat_task = asyncio.create_task(_heartbeat_writer(task_id))
-        return_exceptions=True,
+
-    )
+    try:
+        results = await asyncio.gather(
+            *[_run_on_device(task_id, device, obj, action, params, sem) for device, obj in targets],
+            return_exceptions=True,
+        )
+    finally:
+        heartbeat_task.cancel()
+        try:
+            await heartbeat_task
+        except asyncio.CancelledError:
+            pass
 
     ok = sum(1 for r in results if r is True)
     skipped = sum(1 for r in results if r is None)

backend/app/workers/startup.py

@@ -6,6 +6,7 @@ Runs once at app startup:
 
 from datetime import datetime, timedelta, timezone
 
+import sqlalchemy as sa
 from sqlalchemy import update
 
 from app.models.task import Task
@@ -20,7 +21,15 @@ async def recover_stale_tasks() -> int:
         async with db.begin():
             result = await db.execute(
                 update(Task)
-                .where(Task.status == "running", Task.started_at < cutoff)
+                .where(
+                    Task.status == "running",
+                    # Use heartbeat_at when available (more accurate for long tasks),
+                    # fall back to started_at for tasks that never wrote a heartbeat.
+                    sa.or_(
+                        sa.and_(Task.heartbeat_at.isnot(None), Task.heartbeat_at < cutoff),
+                        sa.and_(Task.heartbeat_at.is_(None), Task.started_at < cutoff),
+                    ),
+                )
                 .values(status="stale")
                 .returning(Task.id)
             )

frontend/src/api/alerts.ts

@@ -50,3 +50,16 @@ export const useResolveAlert = () => {
     onSuccess: () => qc.invalidateQueries({ queryKey: ['alerts'] }),
   })
 }
+
+export const useAcknowledgeByObject = () => {
+  const qc = useQueryClient()
+  return useMutation({
+    mutationFn: (objectId: number) =>
+      api
+        .post<AlertItem[]>(`/api/v1/alerts/acknowledge-by-object`, null, {
+          params: { object_id: objectId },
+        })
+        .then((r) => r.data),
+    onSuccess: () => qc.invalidateQueries({ queryKey: ['alerts'] }),
+  })
+}

frontend/src/api/tasks.ts

@@ -1,5 +1,6 @@
 import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'
 import { api } from './client'
+import { useSSE } from '../hooks/useSSE'
 
 export interface TaskEvent {
   id: number
@@ -78,11 +79,50 @@ export const useTask = (id: string | undefined) =>
     queryKey: ['tasks', id],
     queryFn: () => api.get<TaskDetail>(`/api/v1/tasks/${id}`).then((r) => r.data),
     enabled: !!id,
-    refetchInterval: (query) => {
+  })
-      const status = query.state.data?.status
+
-      return status === 'running' || status === 'pending' ? 2000 : false
+// Subscribe to SSE for a running task — merges results in cache, invalidates on completion.
+export const useTaskSSE = (id: string | undefined) => {
+  const qc = useQueryClient()
+  useSSE(id, {
+    onTaskResult: (data) => {
+      qc.setQueryData<TaskDetail>(['tasks', id], (prev) => {
+        if (!prev) return prev
+        const existing = prev.device_results.find((r) => r.device_id === data.device_id)
+        const updated: TaskResult = existing
+          ? { ...existing, status: data.status, stdout: data.stdout, stderr: data.stderr, duration_ms: data.duration_ms }
+          : {
+              id: 0,
+              device_id: data.device_id,
+              device_ip: data.device_ip,
+              device_hostname: data.device_hostname,
+              object_id: null,
+              object_name: null,
+              rack_id: null,
+              rack_name: null,
+              status: data.status,
+              stdout: data.stdout,
+              stderr: data.stderr,
+              exit_code: null,
+              data: null,
+              duration_ms: data.duration_ms,
+              executed_at: null,
+            }
+        return {
+          ...prev,
+          device_results: existing
+            ? prev.device_results.map((r) => (r.device_id === data.device_id ? updated : r))
+            : [...prev.device_results, updated],
+        }
+      })
+    },
+    onTaskStatus: () => {
+      // Task finished — do a full refetch to get final status and complete results
+      qc.invalidateQueries({ queryKey: ['tasks', id] })
+      qc.invalidateQueries({ queryKey: ['tasks'] })
     },
   })
+}
 
 export const useActions = () =>
   useQuery({
@@ -133,4 +173,5 @@ export const useDeviceTasks = (deviceId: number) =>
       api
         .get<Task[]>('/api/v1/tasks', { params: { device_id: deviceId, limit: 100 } })
         .then((r) => r.data),
+    refetchInterval: 10_000,
   })

frontend/src/hooks/useSSE.ts

@@ -39,8 +39,9 @@ export function useSSE(taskId: string | null | undefined, handlers: SSEHandlers)
           // ignore parse errors
         }
       },
-      onerror() {
+      onerror(err) {
         ctrl.abort()
+        throw err
       },
     })
 

frontend/src/pages/Alerts.tsx

@@ -18,13 +18,14 @@ import {
   Typography,
   message,
 } from 'antd'
-import { useState } from 'react'
+import { useState, useMemo } from 'react'
 import { useQueryClient } from '@tanstack/react-query'
 import { useNavigate } from 'react-router-dom'
 import {
   useAlerts,
   useAcknowledgeAlert,
   useResolveAlert,
+  useAcknowledgeByObject,
   type AlertItem,
 } from '../api/alerts'
 import { useMonitorSSE } from '../hooks/useMonitorSSE'
@@ -48,6 +49,12 @@ function AlertStatusTag({ status }: { status: string }) {
   )
 }
 
+interface ObjectGroup {
+  object_id: number | null
+  object_name: string | null
+  alerts: AlertItem[]
+}
+
 export function AlertsPage() {
   const navigate = useNavigate()
   const qc = useQueryClient()
@@ -61,13 +68,28 @@ export function AlertsPage() {
 
   const acknowledge = useAcknowledgeAlert()
   const resolve = useResolveAlert()
+  const acknowledgeByObject = useAcknowledgeByObject()
 
-  // Live updates via monitor SSE
   useMonitorSSE({
     onAlertCreated: () => qc.invalidateQueries({ queryKey: ['alerts'] }),
     onAlertUpdated: () => qc.invalidateQueries({ queryKey: ['alerts'] }),
   })
 
+  const groups = useMemo<ObjectGroup[]>(() => {
+    if (!alerts) return []
+    const map = new Map<number | null, ObjectGroup>()
+    for (const a of alerts) {
+      const key = a.object_id ?? null
+      if (!map.has(key)) {
+        map.set(key, { object_id: key, object_name: a.object_name, alerts: [] })
+      }
+      map.get(key)!.alerts.push(a)
+    }
+    return Array.from(map.values()).sort((a, b) =>
+      (a.object_name ?? '').localeCompare(b.object_name ?? '', 'ru')
+    )
+  }, [alerts])
+
   const handleAcknowledge = async (id: number) => {
     try {
       await acknowledge.mutateAsync(id)
@@ -86,6 +108,15 @@ export function AlertsPage() {
     }
   }
 
+  const handleAcknowledgeByObject = async (objectId: number) => {
+    try {
+      await acknowledgeByObject.mutateAsync(objectId)
+      message.success('Все алерты объекта приняты')
+    } catch {
+      message.error('Ошибка')
+    }
+  }
+
   const columns = [
     {
       title: 'Статус',
@@ -108,17 +139,6 @@ export function AlertsPage() {
         </Space>
       ),
     },
-    {
-      title: 'Объект',
-      dataIndex: 'object_name',
-      key: 'object_name',
-      render: (v: string | null, row: AlertItem) =>
-        v ? (
-          <a onClick={() => navigate(`/objects/${row.object_id}`)}>{v}</a>
-        ) : (
-          '—'
-        ),
-    },
     {
       title: 'Сообщение',
       dataIndex: 'message',
@@ -128,6 +148,7 @@ export function AlertsPage() {
       title: 'Создан',
       dataIndex: 'created_at',
       key: 'created_at',
+      width: 110,
       render: (v: string) => (
         <Tooltip title={dayjs(v).format('DD.MM.YYYY HH:mm:ss')}>
           {dayjs(v).format('DD.MM HH:mm')}
@@ -137,6 +158,7 @@ export function AlertsPage() {
     {
       title: 'Принят',
       key: 'acknowledged',
+      width: 130,
       render: (_: unknown, row: AlertItem) =>
         row.acknowledged_at ? (
           <Space direction="vertical" size={0}>
@@ -205,9 +227,7 @@ export function AlertsPage() {
             <Typography.Title level={4} style={{ margin: 0 }}>
               Алерты
             </Typography.Title>
-            {openCount > 0 && (
+            {openCount > 0 && <Badge count={openCount} color="red" />}
-              <Badge count={openCount} color="red" />
-            )}
           </Space>
         </Col>
         <Col>
@@ -225,15 +245,60 @@ export function AlertsPage() {
         </Col>
       </Row>
 
-      <Table
+      {isLoading ? (
-        dataSource={alerts ?? []}
+        <Table loading columns={columns} dataSource={[]} rowKey="id" />
-        columns={columns}
+      ) : (
-        rowKey="id"
+        groups.map((group) => {
-        loading={isLoading}
+          const openInGroup = group.alerts.filter((a) => a.status === 'open').length
-        size="middle"
+          return (
-        pagination={{ pageSize: 50 }}
+            <div key={group.object_id ?? 'unknown'} style={{ marginBottom: 24 }}>
-        rowClassName={(row) => (row.status === 'open' ? 'alert-row-open' : '')}
+              <Row justify="space-between" align="middle" style={{ marginBottom: 8 }}>
-      />
+                <Col>
+                  <Space align="center">
+                    {group.object_id ? (
+                      <Typography.Link
+                        strong
+                        onClick={() => navigate(`/work/objects/${group.object_id}`)}
+                      >
+                        {group.object_name ?? `Объект #${group.object_id}`}
+                      </Typography.Link>
+                    ) : (
+                      <Typography.Text strong>Без объекта</Typography.Text>
+                    )}
+                    <Badge count={group.alerts.length} color="blue" />
+                  </Space>
+                </Col>
+                {openInGroup > 0 && group.object_id != null && (
+                  <Col>
+                    <Popconfirm
+                      title={`Принять все ${openInGroup} открытых алертов объекта?`}
+                      okText="Принять все"
+                      cancelText="Отмена"
+                      onConfirm={() => handleAcknowledgeByObject(group.object_id!)}
+                    >
+                      <Button
+                        size="small"
+                        icon={<ClockCircleOutlined />}
+                        loading={acknowledgeByObject.isPending}
+                      >
+                        Принять все ({openInGroup})
+                      </Button>
+                    </Popconfirm>
+                  </Col>
+                )}
+              </Row>
+              <Table
+                dataSource={group.alerts}
+                columns={columns}
+                rowKey="id"
+                size="small"
+                pagination={false}
+                rowClassName={(row) => (row.status === 'open' ? 'alert-row-open' : '')}
+              />
+            </div>
+          )
+        })
+      )}
     </div>
   )
 }

frontend/src/pages/ObjectDetail.tsx

@@ -45,7 +45,7 @@ import {
 } from '../api/objects'
 import { useRacks, useCreateRack, useUpdateRack, useDeleteRack, type RackItem } from '../api/racks'
 import { useZones, useCreateZone, useDeleteZone, type ZoneItem } from '../api/zones'
-import { useActions, useCreateTask, useTask } from '../api/tasks'
+import { useActions, useCreateTask, useTask, useTaskSSE } from '../api/tasks'
 import { CategoryTag } from '../components/StatusBadge'
 import { DeviceEditModal } from '../components/DeviceEditModal'
 import { DiscoveryDrawer } from '../components/DiscoveryDrawer'
@@ -90,8 +90,11 @@ function PingButton({
   const [taskId, setTaskId] = useState<string | null>(null)
   const queryClient = useQueryClient()
   const createTask = useCreateTask()
-  const { data: task } = useTask(taskId ?? undefined)
 
+  useTaskSSE(taskId ?? undefined)
+
+  // When task finishes (SSE invalidates ['tasks', taskId]), refresh devices and clear taskId
+  const { data: task } = useTask(taskId ?? undefined)
   useEffect(() => {
     if (task && ['success', 'failed', 'partial', 'stale', 'cancelled'].includes(task.status)) {
       queryClient.invalidateQueries({ queryKey: ['devices', objId] })