diff --git a/.gitignore b/.gitignore
index 2eea525..a1b9cbb 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1 +1,2 @@
-.env
\ No newline at end of file
+.env
+__pycache__/
\ No newline at end of file
diff --git a/backend/alembic/env.py b/backend/alembic/env.py
index e81ae9d..e84a3be 100644
--- a/backend/alembic/env.py
+++ b/backend/alembic/env.py
@@ -1,4 +1,5 @@
 import asyncio
+import os
 from logging.config import fileConfig
 
 from alembic import context
@@ -6,6 +7,7 @@ from sqlalchemy.ext.asyncio import async_engine_from_config
 from sqlalchemy import pool
 
 from app.models import Base
+from app.config import settings
 
 config = context.config
 if config.config_file_name is not None:
@@ -13,6 +15,10 @@ if config.config_file_name is not None:
 
 target_metadata = Base.metadata
 
+# Override URL from settings/.env, taking precedence over alembic.ini
+_db_url = os.environ.get("DATABASE_URL", settings.DATABASE_URL)
+config.set_main_option("sqlalchemy.url", _db_url)
+
 
 def run_migrations_offline() -> None:
     url = config.get_main_option("sqlalchemy.url")
diff --git a/backend/alembic/versions/0002_tasks.py b/backend/alembic/versions/0002_tasks.py
new file mode 100644
index 0000000..400a6d8
--- /dev/null
+++ b/backend/alembic/versions/0002_tasks.py
@@ -0,0 +1,70 @@
+"""Tasks, task events and task results
+
+Revision ID: 0002
+Revises: 0001
+Create Date: 2026-04-07
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import JSONB
+from alembic import op
+
+revision: str = "0002"
+down_revision: Union[str, None] = "0001"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "tasks",
+        sa.Column("id", sa.Uuid(as_uuid=False), primary_key=True),
+        sa.Column("type", sa.String(100), nullable=False),
+        sa.Column("status", sa.String(20), nullable=False, server_default="pending"),
+        sa.Column("target_scope", JSONB, nullable=False),
+        sa.Column("params", JSONB, nullable=False, server_default="{}"),
+        sa.Column("result", JSONB, nullable=True),
+        sa.Column("error_message", sa.Text, nullable=True),
+        sa.Column("created_by", sa.Integer, sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("heartbeat_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.create_index("ix_tasks_status", "tasks", ["status"])
+    op.create_index("ix_tasks_created_by", "tasks", ["created_by"])
+    op.create_index("ix_tasks_created_at", "tasks", ["created_at"])
+
+    op.create_table(
+        "task_events",
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column("task_id", sa.Uuid(as_uuid=False), sa.ForeignKey("tasks.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("message", sa.Text, nullable=False),
+        sa.Column("level", sa.String(10), nullable=False, server_default="info"),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+    )
+    op.create_index("ix_task_events_task_id", "task_events", ["task_id"])
+
+    op.create_table(
+        "task_results",
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column("task_id", sa.Uuid(as_uuid=False), sa.ForeignKey("tasks.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("device_id", sa.Integer, sa.ForeignKey("devices.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("status", sa.String(20), nullable=False, server_default="pending"),
+        sa.Column("stdout", sa.Text, nullable=True),
+        sa.Column("stderr", sa.Text, nullable=True),
+        sa.Column("exit_code", sa.Integer, nullable=True),
+        sa.Column("data", JSONB, nullable=True),
+        sa.Column("duration_ms", sa.Integer, nullable=True),
+        sa.Column("executed_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.create_index("ix_task_results_task_id", "task_results", ["task_id"])
+    op.create_index("ix_task_results_device_id", "task_results", ["device_id"])
+
+
+def downgrade() -> None:
+    op.drop_table("task_results")
+    op.drop_table("task_events")
+    op.drop_table("tasks")
diff --git a/backend/app/main.py b/backend/app/main.py
index e389e46..25b1a8c 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -2,16 +2,19 @@ from contextlib import asynccontextmanager
 
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.openapi.utils import get_openapi
 
 from app.middleware.auth import AuthMiddleware
-from app.routers import auth, devices, objects, tags
+from app.plugins.loader import load_plugins
+from app.routers import auth, devices, events, objects, tags, tasks
+from app.workers.startup import recover_stale_tasks
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    # Startup: future workers (monitor, stale task recovery) will be launched here in Этап 2+
+    load_plugins()
+    await recover_stale_tasks()
     yield
-    # Shutdown
 
 
 def create_app() -> FastAPI:
@@ -37,7 +40,35 @@ def create_app() -> FastAPI:
     app.include_router(objects.router)
     app.include_router(devices.router)
     app.include_router(tags.router)
+    app.include_router(tasks.router)
+    app.include_router(events.router)
 
+    def custom_openapi():
+        if app.openapi_schema:
+            return app.openapi_schema
+        schema = get_openapi(
+            title=app.title,
+            version=app.version,
+            description=app.description,
+            routes=app.routes,
+        )
+        schema.setdefault("components", {}).setdefault("securitySchemes", {})["BearerAuth"] = {
+            "type": "http",
+            "scheme": "bearer",
+            "bearerFormat": "JWT",
+        }
+        for path in schema.get("paths", {}).values():
+            for operation in path.values():
+                operation.setdefault("security", [{"BearerAuth": []}])
+        # Login and refresh don't need auth
+        for no_auth in ["/api/v1/auth/login", "/api/v1/auth/refresh"]:
+            if no_auth in schema.get("paths", {}):
+                for op in schema["paths"][no_auth].values():
+                    op["security"] = []
+        app.openapi_schema = schema
+        return schema
+
+    app.openapi = custom_openapi
     return app
 
 
diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py
index 01d23d2..bfd8f6a 100644
--- a/backend/app/models/__init__.py
+++ b/backend/app/models/__init__.py
@@ -2,6 +2,7 @@ from .base import Base, TimestampMixin
 from .device import Device
 from .object import Object, object_tags
 from .tag import Tag
+from .task import Task, TaskEvent, TaskResult
 from .user import User, UserSession
 
 __all__ = [
@@ -13,4 +14,7 @@ __all__ = [
     "Object",
     "object_tags",
     "Device",
+    "Task",
+    "TaskEvent",
+    "TaskResult",
 ]
diff --git a/backend/app/models/task.py b/backend/app/models/task.py
new file mode 100644
index 0000000..81dde01
--- /dev/null
+++ b/backend/app/models/task.py
@@ -0,0 +1,76 @@
+from datetime import datetime
+from typing import Any, Optional
+from uuid import uuid4
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String, Text
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.types import Uuid
+
+from .base import Base
+
+TASK_STATUSES = ("pending", "running", "success", "partial", "failed", "stale", "cancelled")
+
+
+class Task(Base):
+    __tablename__ = "tasks"
+
+    id: Mapped[str] = mapped_column(
+        Uuid(as_uuid=False), primary_key=True, default=lambda: str(uuid4())
+    )
+    type: Mapped[str] = mapped_column(String(100), nullable=False)
+    status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending")
+    target_scope: Mapped[dict] = mapped_column(JSONB, nullable=False)
+    params: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
+    result: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True)
+    error_message: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+
+    created_by: Mapped[Optional[int]] = mapped_column(
+        Integer, ForeignKey("users.id", ondelete="SET NULL"), nullable=True
+    )
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    started_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+    finished_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+    heartbeat_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+
+    events: Mapped[list["TaskEvent"]] = relationship(
+        "TaskEvent", back_populates="task", cascade="all, delete-orphan"
+    )
+    device_results: Mapped[list["TaskResult"]] = relationship(
+        "TaskResult", back_populates="task", cascade="all, delete-orphan"
+    )
+
+
+class TaskEvent(Base):
+    __tablename__ = "task_events"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    task_id: Mapped[str] = mapped_column(
+        Uuid(as_uuid=False), ForeignKey("tasks.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    message: Mapped[str] = mapped_column(Text, nullable=False)
+    level: Mapped[str] = mapped_column(String(10), nullable=False, default="info")
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+
+    task: Mapped["Task"] = relationship("Task", back_populates="events")
+
+
+class TaskResult(Base):
+    __tablename__ = "task_results"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    task_id: Mapped[str] = mapped_column(
+        Uuid(as_uuid=False), ForeignKey("tasks.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    device_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("devices.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending")
+    stdout: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    stderr: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    exit_code: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
+    data: Mapped[Optional[dict[str, Any]]] = mapped_column(JSONB, nullable=True)
+    duration_ms: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
+    executed_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+
+    task: Mapped["Task"] = relationship("Task", back_populates="device_results")
diff --git a/backend/app/plugins/__init__.py b/backend/app/plugins/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/backend/app/plugins/base.py b/backend/app/plugins/base.py
new file mode 100644
index 0000000..4d2f152
--- /dev/null
+++ b/backend/app/plugins/base.py
@@ -0,0 +1,52 @@
+"""
+BaseAction — interface every action plugin must implement.
+
+execute() receives the device, its parent object, params from the request,
+and an emit() callback to write progress messages to the SSE stream.
+"""
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Any, Awaitable, Callable
+
+from app.models.device import Device
+from app.models.object import Object
+
+# Type alias for the progress callback
+Emitter = Callable[[str, str], Awaitable[None]]  # (message, level)
+
+
+@dataclass
+class ActionResult:
+    success: bool
+    stdout: str = ""
+    stderr: str = ""
+    exit_code: int = 0
+    data: dict[str, Any] = field(default_factory=dict)
+    error: str = ""
+
+
+class BaseAction(ABC):
+    # Unique identifier used in POST /api/v1/tasks {"type": "<name>"}
+    name: str
+
+    # Human-readable name shown in UI
+    display_name: str
+
+    # Which device categories this action supports. Empty list = all categories.
+    compatible_categories: list[str] = []
+
+    # Parameter definitions for UI form generation
+    params_schema: list[dict] = []
+
+    @abstractmethod
+    async def execute(
+        self,
+        device: Device,
+        obj: Object,
+        params: dict,
+        emit: Emitter,
+    ) -> ActionResult: ...
+
+    def is_compatible(self, category: str) -> bool:
+        return not self.compatible_categories or category in self.compatible_categories
diff --git a/backend/app/plugins/builtin/__init__.py b/backend/app/plugins/builtin/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/backend/app/plugins/builtin/common/__init__.py b/backend/app/plugins/builtin/common/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/backend/app/plugins/builtin/common/ping.py b/backend/app/plugins/builtin/common/ping.py
new file mode 100644
index 0000000..d368305
--- /dev/null
+++ b/backend/app/plugins/builtin/common/ping.py
@@ -0,0 +1,40 @@
+import asyncio
+import time
+
+from app.models.device import Device
+from app.models.object import Object
+from app.plugins.base import ActionResult, BaseAction, Emitter
+
+
+class PingAction(BaseAction):
+    name = "ping"
+    display_name = "Ping"
+    compatible_categories = []  # works for all device types
+    params_schema = []
+
+    async def execute(self, device: Device, obj: Object, params: dict, emit: Emitter) -> ActionResult:
+        await emit(f"Pinging {device.ip}...", "info")
+        start = time.monotonic()
+        try:
+            proc = await asyncio.create_subprocess_exec(
+                "ping", "-c", "1", "-W", "2", device.ip,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=5)
+            duration = int((time.monotonic() - start) * 1000)
+            success = proc.returncode == 0
+            msg = f"{device.ip}: {'reachable' if success else 'unreachable'} ({duration}ms)"
+            await emit(msg, "info" if success else "warn")
+            return ActionResult(
+                success=success,
+                stdout=stdout.decode(errors="replace").strip(),
+                exit_code=proc.returncode or 0,
+                data={"ping_ms": duration if success else None},
+            )
+        except asyncio.TimeoutError:
+            await emit(f"{device.ip}: ping timed out", "error")
+            return ActionResult(success=False, error="Timeout")
+        except Exception as e:
+            await emit(f"{device.ip}: error — {e}", "error")
+            return ActionResult(success=False, error=str(e))
diff --git a/backend/app/plugins/builtin/ssh/__init__.py b/backend/app/plugins/builtin/ssh/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/backend/app/plugins/builtin/ssh/get_timedatectl.py b/backend/app/plugins/builtin/ssh/get_timedatectl.py
new file mode 100644
index 0000000..4edff0a
--- /dev/null
+++ b/backend/app/plugins/builtin/ssh/get_timedatectl.py
@@ -0,0 +1,40 @@
+from app.models.device import Device
+from app.models.object import Object
+from app.plugins.base import ActionResult, BaseAction, Emitter
+from app.services import ssh as ssh_service
+
+
+class GetTimedatectlAction(BaseAction):
+    name = "get_timedatectl"
+    display_name = "Get Time/NTP Info"
+    compatible_categories = ["raspberry", "server"]
+    params_schema = []
+
+    async def execute(self, device: Device, obj: Object, params: dict, emit: Emitter) -> ActionResult:
+        await emit(f"[{device.ip}] Running timedatectl...", "info")
+        result = await ssh_service.run_command(device, obj, "timedatectl show --no-pager 2>/dev/null || timedatectl")
+
+        if result.error:
+            await emit(f"[{device.ip}] Error: {result.error}", "error")
+            return ActionResult(success=False, error=result.error)
+
+        parsed = _parse_timedatectl(result.stdout)
+        await emit(f"[{device.ip}] timezone={parsed.get('Timezone', '?')} ntp={parsed.get('NTP', '?')}", "info")
+
+        return ActionResult(
+            success=result.success,
+            stdout=result.stdout,
+            data=parsed,
+        )
+
+
+def _parse_timedatectl(output: str) -> dict:
+    data: dict = {}
+    for line in output.splitlines():
+        if "=" in line:
+            key, _, value = line.partition("=")
+            data[key.strip()] = value.strip()
+        elif ":" in line:
+            key, _, value = line.partition(":")
+            data[key.strip()] = value.strip()
+    return data
diff --git a/backend/app/plugins/builtin/ssh/ssh_command.py b/backend/app/plugins/builtin/ssh/ssh_command.py
new file mode 100644
index 0000000..82a55e7
--- /dev/null
+++ b/backend/app/plugins/builtin/ssh/ssh_command.py
@@ -0,0 +1,38 @@
+from app.models.device import Device
+from app.models.object import Object
+from app.plugins.base import ActionResult, BaseAction, Emitter
+from app.services import ssh as ssh_service
+
+
+class SSHCommandAction(BaseAction):
+    name = "ssh_command"
+    display_name = "Run SSH Command"
+    compatible_categories = ["raspberry", "server", "embedded", "other"]
+    params_schema = [
+        {"name": "command", "type": "string", "label": "Command", "required": True},
+    ]
+
+    async def execute(self, device: Device, obj: Object, params: dict, emit: Emitter) -> ActionResult:
+        command = params.get("command", "").strip()
+        if not command:
+            return ActionResult(success=False, error="No command specified")
+
+        await emit(f"[{device.ip}] $ {command}", "info")
+        result = await ssh_service.run_command(device, obj, command)
+
+        if result.error:
+            await emit(f"[{device.ip}] Error: {result.error}", "error")
+        else:
+            if result.stdout:
+                await emit(f"[{device.ip}] {result.stdout}", "info")
+            if result.stderr:
+                await emit(f"[{device.ip}] stderr: {result.stderr}", "warn")
+            await emit(f"[{device.ip}] exit_code={result.exit_code} ({result.duration_ms}ms)", "info")
+
+        return ActionResult(
+            success=result.success,
+            stdout=result.stdout,
+            stderr=result.stderr,
+            exit_code=result.exit_code,
+            error=result.error,
+        )
diff --git a/backend/app/plugins/loader.py b/backend/app/plugins/loader.py
new file mode 100644
index 0000000..75c3a4b
--- /dev/null
+++ b/backend/app/plugins/loader.py
@@ -0,0 +1,23 @@
+"""
+PluginLoader — discovers and registers action plugins at startup.
+
+Etap 2: loads only hardcoded builtins (all enabled).
+Etap 5: will scan plugins/custom/ and cross-reference with the plugins DB table.
+"""
+
+from app.plugins.base import BaseAction
+from app.plugins.builtin.common.ping import PingAction
+from app.plugins.builtin.ssh.get_timedatectl import GetTimedatectlAction
+from app.plugins.builtin.ssh.ssh_command import SSHCommandAction
+from app.plugins.registry import action_registry
+
+_BUILTIN_ACTIONS: list[BaseAction] = [
+    PingAction(),
+    SSHCommandAction(),
+    GetTimedatectlAction(),
+]
+
+
+def load_plugins() -> None:
+    for action in _BUILTIN_ACTIONS:
+        action_registry.register(action)
diff --git a/backend/app/plugins/registry.py b/backend/app/plugins/registry.py
new file mode 100644
index 0000000..bf8c80f
--- /dev/null
+++ b/backend/app/plugins/registry.py
@@ -0,0 +1,22 @@
+from app.plugins.base import BaseAction
+
+
+class ActionRegistry:
+    def __init__(self) -> None:
+        self._actions: dict[str, BaseAction] = {}
+
+    def register(self, action: BaseAction) -> None:
+        self._actions[action.name] = action
+
+    def get(self, name: str) -> BaseAction | None:
+        return self._actions.get(name)
+
+    def all(self) -> list[BaseAction]:
+        return list(self._actions.values())
+
+    def for_category(self, category: str) -> list[BaseAction]:
+        return [a for a in self._actions.values() if a.is_compatible(category)]
+
+
+# Module-level singleton
+action_registry = ActionRegistry()
diff --git a/backend/app/routers/events.py b/backend/app/routers/events.py
new file mode 100644
index 0000000..9476672
--- /dev/null
+++ b/backend/app/routers/events.py
@@ -0,0 +1,52 @@
+"""
+SSE endpoint — clients connect here to receive live task progress.
+
+GET /api/v1/events?task_id=<uuid>
+
+The connection stays open; the server pushes events as the task runs.
+Nginx must have `proxy_buffering off` for this to work through a reverse proxy.
+"""
+
+import asyncio
+
+from fastapi import APIRouter, Depends, Request
+from fastapi.responses import StreamingResponse
+
+from app.dependencies import get_current_user
+from app.models.user import User
+from app.services.sse import sse_manager
+
+router = APIRouter(prefix="/api/v1/events", tags=["events"])
+
+
+@router.get("")
+async def event_stream(
+    request: Request,
+    task_id: str,
+    _: User = Depends(get_current_user),
+):
+    queue: asyncio.Queue = asyncio.Queue(maxsize=200)
+    await sse_manager.subscribe(task_id, queue)
+
+    async def generator():
+        try:
+            while True:
+                if await request.is_disconnected():
+                    break
+                try:
+                    event = await asyncio.wait_for(queue.get(), timeout=25.0)
+                    yield event.encode()
+                except asyncio.TimeoutError:
+                    yield ": keepalive\n\n"
+        finally:
+            await sse_manager.unsubscribe(task_id, queue)
+
+    return StreamingResponse(
+        generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",   # tells Nginx to disable buffering
+            "Connection": "keep-alive",
+        },
+    )
diff --git a/backend/app/routers/tasks.py b/backend/app/routers/tasks.py
new file mode 100644
index 0000000..4c0cdde
--- /dev/null
+++ b/backend/app/routers/tasks.py
@@ -0,0 +1,73 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select, desc
+from sqlalchemy.orm import selectinload
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.dependencies import get_current_user, get_db
+from app.models.task import Task
+from app.models.user import User
+from app.plugins.registry import action_registry
+from app.schemas.task import ActionInfo, TaskCreate, TaskDetail, TaskRead
+from app.services import task as task_service
+
+router = APIRouter(prefix="/api/v1/tasks", tags=["tasks"])
+
+
+@router.get("/actions", response_model=list[ActionInfo])
+async def list_actions(_: User = Depends(get_current_user)):
+    return [
+        ActionInfo(
+            name=a.name,
+            display_name=a.display_name,
+            compatible_categories=a.compatible_categories,
+            params_schema=a.params_schema,
+        )
+        for a in action_registry.all()
+    ]
+
+
+@router.post("", response_model=TaskRead, status_code=status.HTTP_202_ACCEPTED)
+async def create_task(
+    body: TaskCreate,
+    current_user: User = Depends(get_current_user),
+):
+    try:
+        task = await task_service.create_task(
+            action_type=body.type,
+            target_scope=body.target_scope,
+            params=body.params,
+            user_id=current_user.id,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+    return task
+
+
+@router.get("", response_model=list[TaskRead])
+async def list_tasks(
+    limit: int = 50,
+    offset: int = 0,
+    db: AsyncSession = Depends(get_db),
+    _: User = Depends(get_current_user),
+):
+    result = await db.execute(
+        select(Task).order_by(desc(Task.created_at)).offset(offset).limit(limit)
+    )
+    return result.scalars().all()
+
+
+@router.get("/{task_id}", response_model=TaskDetail)
+async def get_task(
+    task_id: str,
+    db: AsyncSession = Depends(get_db),
+    _: User = Depends(get_current_user),
+):
+    result = await db.execute(
+        select(Task)
+        .options(selectinload(Task.events), selectinload(Task.device_results))
+        .where(Task.id == task_id)
+    )
+    task = result.scalar_one_or_none()
+    if task is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Task not found")
+    return task
diff --git a/backend/app/schemas/task.py b/backend/app/schemas/task.py
new file mode 100644
index 0000000..1fe3889
--- /dev/null
+++ b/backend/app/schemas/task.py
@@ -0,0 +1,61 @@
+from datetime import datetime
+from typing import Any, Optional
+
+from pydantic import BaseModel
+
+
+class TaskCreate(BaseModel):
+    type: str
+    target_scope: dict   # {"type": "device", "id": 42} | {"type": "object", "id": 5} | etc.
+    params: dict[str, Any] = {}
+
+
+class TaskEventRead(BaseModel):
+    id: int
+    message: str
+    level: str
+    created_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class TaskResultRead(BaseModel):
+    id: int
+    device_id: int
+    status: str
+    stdout: Optional[str] = None
+    stderr: Optional[str] = None
+    exit_code: Optional[int] = None
+    data: Optional[dict] = None
+    duration_ms: Optional[int] = None
+    executed_at: Optional[datetime] = None
+
+    model_config = {"from_attributes": True}
+
+
+class TaskRead(BaseModel):
+    id: str
+    type: str
+    status: str
+    target_scope: dict
+    params: dict
+    result: Optional[dict] = None
+    error_message: Optional[str] = None
+    created_by: Optional[int] = None
+    created_at: datetime
+    started_at: Optional[datetime] = None
+    finished_at: Optional[datetime] = None
+
+    model_config = {"from_attributes": True}
+
+
+class TaskDetail(TaskRead):
+    events: list[TaskEventRead] = []
+    device_results: list[TaskResultRead] = []
+
+
+class ActionInfo(BaseModel):
+    name: str
+    display_name: str
+    compatible_categories: list[str]
+    params_schema: list[dict]
diff --git a/backend/app/services/auth.py b/backend/app/services/auth.py
index 3091be4..0344c50 100644
--- a/backend/app/services/auth.py
+++ b/backend/app/services/auth.py
@@ -1,8 +1,8 @@
 import secrets
 from datetime import datetime, timedelta, timezone
 
+import bcrypt
 from jose import JWTError, jwt
-from passlib.context import CryptContext
 from sqlalchemy import delete, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -10,15 +10,14 @@ from app.config import settings
 from app.models.user import User, UserSession
 
 ALGORITHM = "HS256"
-pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
 
 def hash_password(password: str) -> str:
-    return pwd_context.hash(password)
+    return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
 
 
 def verify_password(plain: str, hashed: str) -> bool:
-    return pwd_context.verify(plain, hashed)
+    return bcrypt.checkpw(plain.encode(), hashed.encode())
 
 
 def create_access_token(user_id: int) -> str:
diff --git a/backend/app/services/sse.py b/backend/app/services/sse.py
new file mode 100644
index 0000000..af6e0d9
--- /dev/null
+++ b/backend/app/services/sse.py
@@ -0,0 +1,58 @@
+"""
+SSEManager — pub/sub hub for Server-Sent Events.
+
+Task runner calls publish() → all connected SSE clients watching that task_id receive the event.
+"""
+
+import asyncio
+import json
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+
+
+@dataclass
+class SSEEvent:
+    type: str        # "task_event" | "task_status" | "task_result"
+    task_id: str
+    data: dict
+
+    def encode(self) -> str:
+        payload = json.dumps({"task_id": self.task_id, **self.data})
+        return f"event: {self.type}\ndata: {payload}\n\n"
+
+
+class SSEManager:
+    def __init__(self) -> None:
+        # task_id -> list[Queue]
+        self._subscribers: dict[str, list[asyncio.Queue]] = {}
+        self._lock = asyncio.Lock()
+
+    async def subscribe(self, task_id: str, queue: asyncio.Queue) -> None:
+        async with self._lock:
+            self._subscribers.setdefault(task_id, []).append(queue)
+
+    async def unsubscribe(self, task_id: str, queue: asyncio.Queue) -> None:
+        async with self._lock:
+            subs = self._subscribers.get(task_id, [])
+            try:
+                subs.remove(queue)
+            except ValueError:
+                pass
+            if not subs:
+                self._subscribers.pop(task_id, None)
+
+    async def publish(self, event: SSEEvent) -> None:
+        async with self._lock:
+            queues = list(self._subscribers.get(event.task_id, []))
+        for queue in queues:
+            try:
+                queue.put_nowait(event)
+            except asyncio.QueueFull:
+                pass  # slow client, drop event
+
+    def subscriber_count(self, task_id: str) -> int:
+        return len(self._subscribers.get(task_id, []))
+
+
+# Module-level singleton shared across all requests
+sse_manager = SSEManager()
diff --git a/backend/app/services/ssh.py b/backend/app/services/ssh.py
new file mode 100644
index 0000000..c0cee68
--- /dev/null
+++ b/backend/app/services/ssh.py
@@ -0,0 +1,107 @@
+"""
+SSHService — asyncssh wrapper with per-device/per-object/global concurrency limits.
+
+Each call to run_command() opens a fresh connection (no pooling).
+Semaphores prevent overwhelming devices or the central server.
+"""
+
+import asyncio
+import time
+from dataclasses import dataclass
+
+import asyncssh
+
+from app.models.device import Device
+from app.models.object import Object
+from app.services.crypto import decrypt
+
+# Concurrency limits
+_GLOBAL_LIMIT = 50
+_OBJECT_LIMIT = 10
+_DEVICE_LIMIT = 1
+
+_global_sem = asyncio.Semaphore(_GLOBAL_LIMIT)
+_object_sems: dict[int, asyncio.Semaphore] = {}
+_device_sems: dict[int, asyncio.Semaphore] = {}
+
+
+def _get_object_sem(object_id: int) -> asyncio.Semaphore:
+    if object_id not in _object_sems:
+        _object_sems[object_id] = asyncio.Semaphore(_OBJECT_LIMIT)
+    return _object_sems[object_id]
+
+
+def _get_device_sem(device_id: int) -> asyncio.Semaphore:
+    if device_id not in _device_sems:
+        _device_sems[device_id] = asyncio.Semaphore(_DEVICE_LIMIT)
+    return _device_sems[device_id]
+
+
+@dataclass
+class SSHResult:
+    success: bool
+    stdout: str = ""
+    stderr: str = ""
+    exit_code: int = -1
+    duration_ms: int = 0
+    error: str = ""
+
+
+def resolve_ssh_credentials(device: Device, obj: Object) -> tuple[str, str, int]:
+    """Returns (username, password, port) — device overrides take priority."""
+    user = device.ssh_user_override or obj.ssh_user or "root"
+    port = device.ssh_port_override or obj.ssh_port or 22
+
+    if device.ssh_pass_override_enc:
+        password = decrypt(device.ssh_pass_override_enc)
+    elif obj.ssh_pass_enc:
+        password = decrypt(obj.ssh_pass_enc)
+    else:
+        password = ""
+
+    return user, password, port
+
+
+async def run_command(
+    device: Device,
+    obj: Object,
+    command: str,
+    timeout: int = 30,
+) -> SSHResult:
+    user, password, port = resolve_ssh_credentials(device, obj)
+    start = time.monotonic()
+
+    async with _global_sem:
+        async with _get_object_sem(device.object_id):
+            async with _get_device_sem(device.id):
+                try:
+                    conn = await asyncio.wait_for(
+                        asyncssh.connect(
+                            device.ip,
+                            port=port,
+                            username=user,
+                            password=password,
+                            known_hosts=None,
+                            connect_timeout=10,
+                        ),
+                        timeout=15,
+                    )
+                    async with conn:
+                        result = await asyncio.wait_for(
+                            conn.run(command, check=False),
+                            timeout=timeout,
+                        )
+                    duration = int((time.monotonic() - start) * 1000)
+                    return SSHResult(
+                        success=result.exit_status == 0,
+                        stdout=(result.stdout or "").strip(),
+                        stderr=(result.stderr or "").strip(),
+                        exit_code=result.exit_status or 0,
+                        duration_ms=duration,
+                    )
+                except asyncio.TimeoutError:
+                    return SSHResult(success=False, error="Connection timed out", duration_ms=int((time.monotonic() - start) * 1000))
+                except asyncssh.DisconnectError as e:
+                    return SSHResult(success=False, error=f"SSH disconnect: {e}", duration_ms=int((time.monotonic() - start) * 1000))
+                except Exception as e:
+                    return SSHResult(success=False, error=str(e), duration_ms=int((time.monotonic() - start) * 1000))
diff --git a/backend/app/services/task.py b/backend/app/services/task.py
new file mode 100644
index 0000000..1517afe
--- /dev/null
+++ b/backend/app/services/task.py
@@ -0,0 +1,236 @@
+"""
+TaskService — creates tasks, dispatches them as asyncio background tasks,
+and persists every state change to the DB so restarts don't lose history.
+
+Flow:
+  create() → INSERT tasks (status=pending) → asyncio.create_task(run) → return task_id
+  run()    → UPDATE running → per-device execute → UPDATE success/failed
+"""
+
+import asyncio
+from datetime import datetime, timezone
+from uuid import uuid4
+
+from sqlalchemy import select
+from sqlalchemy.orm import selectinload
+
+from app.models.device import Device
+from app.models.object import Object
+from app.models.task import Task, TaskEvent, TaskResult
+from app.models.user import User
+from app.plugins.registry import ActionRegistry, action_registry
+from app.services.database import AsyncSessionLocal
+from app.services.sse import SSEEvent, sse_manager
+
+# Max parallel devices within a single task
+_TASK_DEVICE_CONCURRENCY = 20
+
+
+async def _emit(task_id: str, message: str, level: str = "info") -> None:
+    """Write a progress line to DB and push to SSE subscribers."""
+    now = datetime.now(timezone.utc)
+    async with AsyncSessionLocal() as db:
+        async with db.begin():
+            db.add(TaskEvent(task_id=task_id, message=message, level=level, created_at=now))
+
+    await sse_manager.publish(SSEEvent(
+        type="task_event",
+        task_id=task_id,
+        data={"message": message, "level": level, "ts": now.isoformat()},
+    ))
+
+
+async def _resolve_scope(scope: dict) -> list[tuple[Device, Object]]:
+    """Expand target_scope to a list of (device, parent_object) pairs."""
+    async with AsyncSessionLocal() as db:
+        query = select(Device).options(selectinload(Device.object)).where(Device.is_active == True)
+
+        scope_type = scope.get("type")
+        if scope_type == "device":
+            query = query.where(Device.id == scope["id"])
+        elif scope_type == "object":
+            query = query.where(Device.object_id == scope["id"])
+        elif scope_type == "category":
+            query = query.where(Device.category == scope["value"])
+        elif scope_type == "role":
+            query = query.where(Device.role == scope["value"])
+            if "object_id" in scope:
+                query = query.where(Device.object_id == scope["object_id"])
+        else:
+            return []
+
+        result = await db.execute(query)
+        devices = result.scalars().all()
+        return [(d, d.object) for d in devices]
+
+
+async def _run_on_device(
+    task_id: str,
+    device: Device,
+    obj: Object,
+    action,
+    params: dict,
+    sem: asyncio.Semaphore,
+) -> bool:
+    async with sem:
+        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)
+        async with AsyncSessionLocal() as db:
+            async with db.begin():
+                tr = TaskResult(
+                    task_id=task_id,
+                    device_id=device.id,
+                    status="running",
+                    executed_at=started_at,
+                )
+                db.add(tr)
+                await db.flush()
+                result_id = tr.id
+
+        try:
+            action_result = await action.execute(device, obj, params, emit)
+        except Exception as exc:
+            action_result_success = False
+            action_stdout = ""
+            action_stderr = ""
+            action_exit_code = -1
+            action_data = {}
+            action_error = str(exc)
+        else:
+            action_result_success = action_result.success
+            action_stdout = action_result.stdout
+            action_stderr = action_result.stderr
+            action_exit_code = action_result.exit_code
+            action_data = action_result.data
+            action_error = action_result.error
+
+        finished_at = datetime.now(timezone.utc)
+        duration_ms = int((finished_at - started_at).total_seconds() * 1000)
+
+        async with AsyncSessionLocal() as db:
+            async with db.begin():
+                tr = await db.get(TaskResult, result_id)
+                if tr:
+                    tr.status = "success" if action_result_success else "failed"
+                    tr.stdout = action_stdout
+                    tr.stderr = action_stderr
+                    tr.exit_code = action_exit_code
+                    tr.data = action_data or None
+                    tr.duration_ms = duration_ms
+
+        await sse_manager.publish(SSEEvent(
+            type="task_result",
+            task_id=task_id,
+            data={
+                "device_id": device.id,
+                "device_ip": device.ip,
+                "status": "success" if action_result_success else "failed",
+                "duration_ms": duration_ms,
+                "error": action_error or None,
+            },
+        ))
+
+        return action_result_success
+
+
+async def _run_task(task_id: str) -> None:
+    await _emit(task_id, "Task started", "info")
+
+    async with AsyncSessionLocal() as db:
+        async with db.begin():
+            task = await db.get(Task, task_id)
+            if task is None:
+                return
+            task.status = "running"
+            task.started_at = datetime.now(timezone.utc)
+            scope = dict(task.target_scope)
+            action_type = task.type
+            params = dict(task.params)
+
+    action = action_registry.get(action_type)
+    if action is None:
+        async with AsyncSessionLocal() as db:
+            async with db.begin():
+                task = await db.get(Task, task_id)
+                if task:
+                    task.status = "failed"
+                    task.error_message = f"Unknown action: {action_type}"
+                    task.finished_at = datetime.now(timezone.utc)
+        await _emit(task_id, f"Unknown action: {action_type}", "error")
+        return
+
+    targets = await _resolve_scope(scope)
+    if not targets:
+        async with AsyncSessionLocal() as db:
+            async with db.begin():
+                task = await db.get(Task, task_id)
+                if task:
+                    task.status = "success"
+                    task.result = {"total": 0, "ok": 0, "failed": 0}
+                    task.finished_at = datetime.now(timezone.utc)
+        await _emit(task_id, "No devices matched the scope", "warn")
+        return
+
+    await _emit(task_id, f"Running on {len(targets)} device(s)...", "info")
+    sem = asyncio.Semaphore(_TASK_DEVICE_CONCURRENCY)
+
+    results = await asyncio.gather(
+        *[_run_on_device(task_id, device, obj, action, params, sem) for device, obj in targets],
+        return_exceptions=True,
+    )
+
+    ok = sum(1 for r in results if r is True)
+    failed = len(results) - ok
+    final_status = "success" if failed == 0 else ("failed" if ok == 0 else "partial")
+
+    async with AsyncSessionLocal() as db:
+        async with db.begin():
+            task = await db.get(Task, task_id)
+            if task:
+                task.status = final_status
+                task.result = {"total": len(results), "ok": ok, "failed": failed}
+                task.finished_at = datetime.now(timezone.utc)
+
+    await _emit(task_id, f"Done: {ok}/{len(results)} succeeded", "info")
+    await sse_manager.publish(SSEEvent(
+        type="task_status",
+        task_id=task_id,
+        data={"status": final_status, "total": len(results), "ok": ok, "failed": failed},
+    ))
+
+
+async def create_task(
+    action_type: str,
+    target_scope: dict,
+    params: dict,
+    user_id: int | None = None,
+) -> Task:
+    if action_registry.get(action_type) is None:
+        raise ValueError(f"Unknown action: {action_type}")
+
+    now = datetime.now(timezone.utc)
+    task_id = str(uuid4())
+
+    async with AsyncSessionLocal() as db:
+        async with db.begin():
+            task = Task(
+                id=task_id,
+                type=action_type,
+                status="pending",
+                target_scope=target_scope,
+                params=params,
+                created_by=user_id,
+                created_at=now,
+            )
+            db.add(task)
+
+    # Dispatch — fire and forget; state is already persisted
+    asyncio.create_task(_run_task(task_id))
+
+    # Return a fresh copy to the caller
+    async with AsyncSessionLocal() as db:
+        result = await db.get(Task, task_id)
+        return result
diff --git a/backend/app/workers/__init__.py b/backend/app/workers/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/backend/app/workers/startup.py b/backend/app/workers/startup.py
new file mode 100644
index 0000000..e57455b
--- /dev/null
+++ b/backend/app/workers/startup.py
@@ -0,0 +1,30 @@
+"""
+Runs once at app startup:
+  - Marks tasks stuck in 'running' with stale heartbeat as 'stale'.
+    These are tasks that were running when the app last crashed or restarted.
+"""
+
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import update
+
+from app.models.task import Task
+from app.services.database import AsyncSessionLocal
+
+STALE_AFTER_MINUTES = 15
+
+
+async def recover_stale_tasks() -> int:
+    cutoff = datetime.now(timezone.utc) - timedelta(minutes=STALE_AFTER_MINUTES)
+    async with AsyncSessionLocal() as db:
+        async with db.begin():
+            result = await db.execute(
+                update(Task)
+                .where(Task.status == "running", Task.started_at < cutoff)
+                .values(status="stale")
+                .returning(Task.id)
+            )
+            stale_ids = result.scalars().all()
+    if stale_ids:
+        print(f"[startup] Marked {len(stale_ids)} stale task(s): {stale_ids}")
+    return len(stale_ids)
diff --git a/backend/entrypoint.sh b/backend/entrypoint.sh
new file mode 100644
index 0000000..e32f259
--- /dev/null
+++ b/backend/entrypoint.sh
@@ -0,0 +1,6 @@
+#!/bin/bash
+set -e
+echo "Running database migrations..."
+alembic upgrade head
+echo "Starting application..."
+exec uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
diff --git a/backend/pyproject.toml b/backend/pyproject.toml
index ff18171..219dc8e 100644
--- a/backend/pyproject.toml
+++ b/backend/pyproject.toml
@@ -12,11 +12,12 @@ dependencies = [
     "pydantic>=2.10",
     "pydantic-settings>=2.7",
     "python-jose[cryptography]>=3.3",
-    "passlib[bcrypt]>=1.7",
+    "bcrypt>=4.0",
     "cryptography>=43",
-    "psycopg[async]>=3.2",
+    "psycopg[binary,async]>=3.2",
     "python-multipart>=0.0.20",
     "httpx>=0.28",
+    "asyncssh>=2.18",
 ]
 
 [project.optional-dependencies]
diff --git a/backend/scripts/create_admin.py b/backend/scripts/create_admin.py
new file mode 100644
index 0000000..fcca40a
--- /dev/null
+++ b/backend/scripts/create_admin.py
@@ -0,0 +1,40 @@
+"""
+Usage: python scripts/create_admin.py <username> <password>
+"""
+
+import asyncio
+import sys
+import os
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from sqlalchemy import select
+from app.services.database import AsyncSessionLocal
+from app.services.auth import hash_password
+from app.models.user import User
+
+
+async def create_admin(username: str, password: str) -> None:
+    async with AsyncSessionLocal() as db:
+        async with db.begin():
+            exists = await db.execute(select(User).where(User.username == username))
+            if exists.scalar_one_or_none():
+                print(f"User '{username}' already exists.")
+                return
+
+            user = User(
+                username=username,
+                password_hash=hash_password(password),
+                role="admin",
+                is_active=True,
+            )
+            db.add(user)
+
+    print(f"Admin user '{username}' created.")
+
+
+if __name__ == "__main__":
+    if len(sys.argv) != 3:
+        print("Usage: python scripts/create_admin.py <username> <password>")
+        sys.exit(1)
+    asyncio.run(create_admin(sys.argv[1], sys.argv[2]))
diff --git a/docker/Dockerfile.backend b/docker/Dockerfile.backend
index 94a94e4..0669062 100644
--- a/docker/Dockerfile.backend
+++ b/docker/Dockerfile.backend
@@ -4,11 +4,16 @@ WORKDIR /app
 
 RUN apt-get update && apt-get install -y --no-install-recommends \
     iputils-ping \
+    libpq-dev \
     && rm -rf /var/lib/apt/lists/*
 
 COPY pyproject.toml .
 RUN pip install --no-cache-dir -e .
 
+# Run migrations then start the app
+COPY entrypoint.sh /entrypoint.sh
+RUN chmod +x /entrypoint.sh
+
 COPY . .
 
-CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
+CMD ["/entrypoint.sh"]