feat(delivery): production deploy over SSH/SFTP/FTP/FTPS (v2 phase A3-5)

The one path that touches a customer's server, and the one the platform cannot
undo. It is human-triggered, confirmed and audited; there is deliberately no
code path from a pipeline run into it (D13).

- deploy_ssh.py: SshDeployer. Release lands in releases/<ts>-<sha7>/, the
  post-deploy command runs with cwd there, and only then does `current` move —
  atomically (relative symlink + mv -T). A failed hook or extraction never moves
  it, so the previous release keeps serving. Host key is pinned on first success
  (TOFU); a mismatch is a hard failure. Prunes to the newest five.
- deploy_ftp.py: FtpDeployer. FTP has no atomic switch and no rollback except a
  backup, so a backup is mandatory: every file about to be overwritten is
  archived under ARTIFACTS_ROOT before anything is uploaded. The artifact is
  extracted, then the build dir (dist/build/public/out) is chosen — a static
  deploy must not publish the repository. Archive members are sanitised: absolute
  paths, `..` and escaping links are refused.
- deploy_target.py: the tester that dispatches on `kind`.
- deploy_service.py + production_deploys (migration a3b4c5d6e7f4) + 4 routes.
  One deploy per project at a time; the org slug must be typed back; the target
  summary reports has_secret, never the secret.

Tests: 499 API (+26) with fake SSH/FTP transports; ruff, mypy, alembic clean.

Report: docs/plan/reports/agent-3-phase-5.md

apps/api/alembic/versions/a3b4c5d6e7f4_production_deploys.py

@@ -0,0 +1,73 @@
+"""production deploys: the human-triggered deploy ledger
+
+Revision ID: a3b4c5d6e7f4
+Revises: a3b4c5d6e7f3
+Create Date: 2026-10-14 01:10:00.000000
+
+Chained off this workstream's own head (PLAN-v2 §6: "more only if chained off
+your own"), so the three agents' branches still merge cleanly.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "a3b4c5d6e7f4"
+down_revision: str | None = "a3b4c5d6e7f3"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "production_deploys",
+        sa.Column("id", sa.String(length=32), primary_key=True),
+        sa.Column(
+            "organization_id",
+            sa.String(length=32),
+            sa.ForeignKey("organizations.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "project_id",
+            sa.String(length=32),
+            sa.ForeignKey("projects.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "release_id",
+            sa.String(length=32),
+            sa.ForeignKey("releases.id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+        sa.Column(
+            "target_id",
+            sa.String(length=32),
+            sa.ForeignKey("deploy_targets.id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+        sa.Column("kind", sa.String(length=8), nullable=False),
+        sa.Column("host", sa.String(length=255), nullable=False, server_default=""),
+        sa.Column("status", sa.String(length=16), nullable=False, server_default="queued"),
+        sa.Column("message", sa.Text(), nullable=False, server_default=""),
+        sa.Column("log_tail", sa.Text(), nullable=False, server_default=""),
+        sa.Column("rolled_back_from", sa.String(length=32), nullable=True),
+        sa.Column("backup_path", sa.String(length=512), nullable=False, server_default=""),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+    )
+    op.create_index(
+        "ix_production_deploys_organization_id", "production_deploys", ["organization_id"]
+    )
+    op.create_index("ix_production_deploys_project_id", "production_deploys", ["project_id"])
+    op.create_index(
+        "ix_prod_deploy_project_created", "production_deploys", ["project_id", "created_at"]
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("production_deploys")

apps/api/app/application/container.py

@@ -18,6 +18,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.application.container_delivery import (
     build_delivery_hooks,
     build_delivery_service,
+    build_deploy_service,
     build_deploy_target_tester,
     build_repo_workspace,
 )
@@ -168,6 +169,7 @@ __all__ = [
     "build_billing_service",
     "build_delivery_hooks",
     "build_delivery_service",
+    "build_deploy_service",
     "build_deploy_target_tester",
     "build_memory",
     "build_model_grants",

apps/api/app/application/container_delivery.py

@@ -19,9 +19,10 @@ import logging
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.application.ports.delivery_hooks import DeliveryHooks
-from app.application.ports.deploy_target import DeployTargetTester, FakeDeployTargetTester
+from app.application.ports.deploy_target import DeployTargetTester
 from app.application.ports.repo_workspace import FakeRepoWorkspace, RepoWorkspacePort
 from app.application.services.delivery_service import DeliveryService
+from app.application.services.deploy_service import DeployService
 from app.core.config import settings
 from app.core.crypto import decrypt
 from app.infrastructure.adapters.git_workspace import (
@@ -35,10 +36,12 @@ from app.infrastructure.repositories.delivery import (
     AppRuntimeRepository,
     PipelineRunRepository,
     PipelineStepRepository,
+    ProductionDeployRepository,
     ReleaseRepository,
 )
 from app.infrastructure.repositories.identity import OrganizationRepository
 from app.infrastructure.repositories.projects import (
+    DeployTargetRepository,
     PreviewDomainRepository,
     ProjectExternalGitRepository,
     ProjectGitCredentialRepository,
@@ -49,7 +52,6 @@ from app.infrastructure.repositories.projects import (
 logger = logging.getLogger("agy.container.delivery")
 
 _fake_workspace = FakeRepoWorkspace()
-_fake_deploy_tester = FakeDeployTargetTester()
 
 
 def build_repo_workspace(session: AsyncSession) -> RepoWorkspacePort:
@@ -186,4 +188,26 @@ class _UnavailableContainerRuntime:
 
 
 def build_deploy_target_tester(session: AsyncSession) -> DeployTargetTester:
-    return _fake_deploy_tester
+    """The real tester: SSH/SFTP over asyncssh, FTP/FTPS over aioftp.
+
+    The wizard calls this to validate a user-supplied production target, and it
+    runs the same SSRF check the git adapter does before a byte leaves the host.
+    """
+    from app.infrastructure.adapters.deploy_target import DispatchingDeployTargetTester
+
+    return DispatchingDeployTargetTester()
+
+
+def build_deploy_service(session: AsyncSession) -> DeployService:
+    """Production deploys: human-triggered, confirmed and audited."""
+    from app.infrastructure.adapters.deploy_target import DispatchingDeployTargetTester
+
+    return DeployService(
+        projects=ProjectRepository(session),
+        organizations=OrganizationRepository(session),
+        targets=DeployTargetRepository(session),
+        deploys=ProductionDeployRepository(session),
+        releases=ReleaseRepository(session),
+        runtimes=AppRuntimeRepository(session),
+        tester=DispatchingDeployTargetTester(),
+    )

apps/api/app/application/services/deploy_service.py

@@ -0,0 +1,366 @@
+"""Production deploy use cases — the only path that touches a customer's server.
+
+Two rules are absolute here, and both exist because production is the one place
+where a mistake is not recoverable by trying again.
+
+**A human triggers it.** There is no code path from a pipeline run, a task, or
+anything the AI does into this service. The engine's furthest reach is "merged
+and pushed to `dev`"; a production deploy needs an org owner/admin, a typed
+confirmation, and an audit row.
+
+**One deploy per project at a time.** Two concurrent deploys to the same target
+interleave their writes on a shared host. Over SSH that is merely confusing;
+over FTP, which has no atomic switch and no rollback except the backup, it is
+how a site ends up in a state nobody can describe.
+
+The deployers themselves are injected, so this service is testable without an
+sshd or an FTP server.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import logging
+
+from app.application.ports.deploy_target import DeployTargetSpec
+from app.core.config import settings
+from app.core.crypto import decrypt
+from app.core.errors import ConflictError, NotFoundError, ValidationError
+from app.domain.delivery.models import DeployStatus, ProductionDeploy, Release
+from app.infrastructure.repositories.delivery import (
+    AppRuntimeRepository,
+    ProductionDeployRepository,
+    ReleaseRepository,
+)
+from app.infrastructure.repositories.identity import OrganizationRepository
+from app.infrastructure.repositories.projects import (
+    DeployTargetRepository,
+    ProjectRepository,
+)
+
+logger = logging.getLogger("agy.deploy")
+
+#: Bound on the streamed deploy log kept for the post-mortem.
+MAX_LOG_TAIL_CHARS = 65_536
+
+
+class _LogCollector:
+    """Accumulate streamed deploy output, bounded, redacted on read."""
+
+    __slots__ = ("_buffer",)
+
+    def __init__(self) -> None:
+        self._buffer = ""
+
+    async def __call__(self, chunk: str) -> None:
+        if chunk:
+            self._buffer = (self._buffer + chunk)[-MAX_LOG_TAIL_CHARS:]
+
+    @property
+    def tail(self) -> str:
+        from app.core.redact import redact_text
+
+        # Redacted as a whole: a credential split across two chunks would slip
+        # past a per-chunk filter, and a deploy log is where one shows up.
+        return redact_text(self._buffer)
+
+
+class DeployService:
+    """Human-triggered production deploys, and their rollbacks."""
+
+    def __init__(
+        self,
+        *,
+        projects: ProjectRepository,
+        organizations: OrganizationRepository,
+        targets: DeployTargetRepository,
+        deploys: ProductionDeployRepository,
+        releases: ReleaseRepository,
+        runtimes: AppRuntimeRepository,
+        tester,  # noqa: ANN001 - DispatchingDeployTargetTester; avoids a cycle
+        allow_private_hosts: bool | None = None,
+    ) -> None:
+        self.projects = projects
+        self.organizations = organizations
+        self.targets = targets
+        self.deploys = deploys
+        self.releases = releases
+        self.runtimes = runtimes
+        self.tester = tester
+        self.allow_private_hosts = (
+            settings.deploy_allow_private_hosts
+            if allow_private_hosts is None
+            else allow_private_hosts
+        )
+
+    # ------------------------------------------------------------------ reads
+
+    async def get_target(self, *, organization_id: str, project_id: str):
+        """The active deploy target, or ``None``.
+
+        Never returns the secret: the row is rendered as-is by the router, which
+        only reads the non-secret columns.
+        """
+        project = await self.projects.get(project_id)
+        if project is None or project.organization_id != organization_id:
+            raise NotFoundError("اپلیکیشن یافت نشد.")
+        return await self.targets.get_for_project(project_id)
+
+    async def list_deploys(
+        self, *, organization_id: str, project_id: str, limit: int = 20
+    ) -> list[ProductionDeploy]:
+        project = await self.projects.get(project_id)
+        if project is None or project.organization_id != organization_id:
+            raise NotFoundError("اپلیکیشن یافت نشد.")
+        return await self.deploys.list_for_project(project_id=project_id, limit=limit)
+
+    async def latest(self, *, organization_id: str, project_id: str) -> ProductionDeploy | None:
+        rows = await self.list_deploys(
+            organization_id=organization_id, project_id=project_id, limit=1
+        )
+        return rows[0] if rows else None
+
+    # ------------------------------------------------------------- deploy
+
+    async def deploy_production(
+        self,
+        *,
+        organization_id: str,
+        project_id: str,
+        actor_user_id: str,
+        confirm: str,
+        release_id: str | None = None,
+    ) -> ProductionDeploy:
+        """Deploy a release to the customer's server. Irreversible by design.
+
+        ``confirm`` must be the organization's slug typed back. It is not
+        ceremony: this is the only action in the product that changes something
+        the platform cannot undo, and a modal with an OK button gets clicked.
+        """
+        project = await self.projects.get(project_id)
+        if project is None or project.organization_id != organization_id:
+            raise NotFoundError("اپلیکیشن یافت نشد.")
+
+        organization = await self.organizations.get(organization_id)
+        if organization is None:
+            raise NotFoundError("سازمان یافت نشد.")
+        if confirm.strip().lower() != organization.slug.strip().lower():
+            raise ValidationError("تأییدیه با نامک سازمان مطابقت ندارد.")
+
+        if await self.deploys.active_for_project(project_id) is not None:
+            raise ConflictError("یک استقرار در حال انجام است؛ تا پایان آن صبر کنید.")
+
+        target = await self.targets.get_for_project(project_id)
+        if target is None:
+            raise ValidationError("برای این اپلیکیشن مقصد استقراری تنظیم نشده است.")
+
+        release = await self._resolve_release(
+            organization_id=organization_id, project_id=project_id, release_id=release_id
+        )
+
+        spec, secret = _spec_from_target(target, allow_private=self.allow_private_hosts)
+        deploy = await self.deploys.create(
+            ProductionDeploy(
+                organization_id=organization_id,
+                project_id=project_id,
+                kind=target.kind,
+                host=target.host,
+                release_id=release.id,
+                target_id=target.id,
+            )
+        )
+        await self.deploys.mark_running(deploy.id)
+
+        collector = _LogCollector()
+        deployer = self.tester.deployer_for(target.kind)
+
+        if target.kind in ("ssh", "sftp"):
+            ok, detail, message = await deployer.deploy(
+                target=spec,
+                artifact_path=release.artifact_path,
+                sha=release.sha,
+                post_deploy_command=target.post_deploy_command,
+                on_output=collector,
+            )
+            backup = ""
+        else:
+            ok, backup, message = await deployer.deploy(
+                target=spec,
+                artifact_path=release.artifact_path,
+                sha=release.sha,
+                organization_id=organization_id,
+                project_id=project_id,
+                on_output=collector,
+            )
+            detail = backup
+
+        await self.deploys.finish(
+            deploy.id,
+            status=DeployStatus.SUCCEEDED.value if ok else DeployStatus.FAILED.value,
+            message=message,
+            log_tail=collector.tail,
+            backup_path=backup,
+        )
+        _ = actor_user_id, secret, detail
+        refreshed = await self.deploys.get(deploy.id)
+        if refreshed is None:  # pragma: no cover - the row was just written
+            raise NotFoundError("استقرار یافت نشد.")
+        if not ok:
+            logger.info("production deploy %s failed: %s", deploy.id, message)
+        return refreshed
+
+    async def rollback_production(
+        self,
+        *,
+        organization_id: str,
+        project_id: str,
+        actor_user_id: str,
+        confirm: str,
+        deploy_id: str | None = None,
+    ) -> ProductionDeploy:
+        """Undo a production deploy. Same confirmation, same audit trail."""
+        project = await self.projects.get(project_id)
+        if project is None or project.organization_id != organization_id:
+            raise NotFoundError("اپلیکیشن یافت نشد.")
+
+        organization = await self.organizations.get(organization_id)
+        if organization is None:
+            raise NotFoundError("سازمان یافت نشد.")
+        if confirm.strip().lower() != organization.slug.strip().lower():
+            raise ValidationError("تأییدیه با نامک سازمان مطابقت ندارد.")
+
+        if await self.deploys.active_for_project(project_id) is not None:
+            raise ConflictError("یک استقرار در حال انجام است؛ تا پایان آن صبر کنید.")
+
+        target = await self.targets.get_for_project(project_id)
+        if target is None:
+            raise ValidationError("برای این اپلیکیشن مقصد استقراری تنظیم نشده است.")
+
+        previous = (
+            await self.deploys.get_for_org(organization_id, deploy_id)
+            if deploy_id
+            else await self.latest(organization_id=organization_id, project_id=project_id)
+        )
+        if previous is None or previous.status != DeployStatus.SUCCEEDED.value:
+            raise ValidationError("استقرار موفقی برای بازگردانی وجود ندارد.")
+
+        spec, _secret = _spec_from_target(target, allow_private=self.allow_private_hosts)
+        deploy = await self.deploys.create(
+            ProductionDeploy(
+                organization_id=organization_id,
+                project_id=project_id,
+                kind=target.kind,
+                host=target.host,
+                release_id=previous.release_id,
+                target_id=target.id,
+                rolled_back_from=previous.id,
+            )
+        )
+        await self.deploys.mark_running(deploy.id)
+
+        collector = _LogCollector()
+        deployer = self.tester.deployer_for(target.kind)
+
+        if target.kind in ("ssh", "sftp"):
+            ok, detail, message = await deployer.rollback(
+                target=spec,
+                to_release=None,
+                post_deploy_command=target.post_deploy_command,
+                on_output=collector,
+            )
+            backup = ""
+        else:
+            if not previous.backup_path:
+                await self.deploys.finish(
+                    deploy.id,
+                    status=DeployStatus.FAILED.value,
+                    message="پشتیبان این استقرار در دسترس نیست؛ بازگردانی FTP ممکن نیست.",
+                    log_tail=collector.tail,
+                )
+                refreshed = await self.deploys.get(deploy.id)
+                if refreshed is None:  # pragma: no cover
+                    raise NotFoundError("استقرار یافت نشد.")
+                return refreshed
+            ok, message = await deployer.rollback(
+                target=spec, backup_path=previous.backup_path, on_output=collector
+            )
+            detail, backup = previous.backup_path, previous.backup_path
+
+        await self.deploys.finish(
+            deploy.id,
+            status=DeployStatus.ROLLED_BACK.value if ok else DeployStatus.FAILED.value,
+            message=message,
+            log_tail=collector.tail,
+            backup_path=backup,
+        )
+        _ = actor_user_id, detail
+        refreshed = await self.deploys.get(deploy.id)
+        if refreshed is None:  # pragma: no cover
+            raise NotFoundError("استقرار یافت نشد.")
+        return refreshed
+
+    # ------------------------------------------------------------ helpers
+
+    async def _resolve_release(
+        self, *, organization_id: str, project_id: str, release_id: str | None
+    ) -> Release:
+        """The release to ship: an explicit one, or whatever the demo is live on.
+
+        Defaulting to the live demo release is deliberate — the demo is the
+        thing the user has just looked at and approved.
+        """
+        if release_id:
+            release = await self.releases.get_for_org(organization_id, release_id)
+            if release is None:
+                raise NotFoundError("نسخه یافت نشد.")
+            return release
+
+        release = await self.releases.live_for_project(project_id)
+        if release is None:
+            raise ValidationError("هنوز نسخهٔ زنده‌ای برای استقرار وجود ندارد.")
+        return release
+
+    async def test_target(
+        self, *, organization_id: str, project_id: str, actor_user_id: str
+    ) -> tuple[bool, str]:
+        """Run the connection test and record the result on the target row."""
+        target = await self.targets.get_for_project(project_id)
+        if target is None:
+            raise NotFoundError("مقصد استقرار یافت نشد.")
+
+        spec, _secret = _spec_from_target(target, allow_private=self.allow_private_hosts)
+        check = await self.tester.test_connection(target=spec)
+        await self.targets.record_test(
+            target_id=target.id, ok=check.ok, message=check.message,
+            fingerprint=check.fingerprint,
+        )
+        _ = organization_id, actor_user_id
+        return check.ok, check.message
+
+
+def _spec_from_target(target, *, allow_private: bool) -> tuple[DeployTargetSpec, str | None]:
+    """Build the port-level spec, decrypting the secret exactly here.
+
+    The plaintext exists only inside this function and the call it feeds. It is
+    never logged, never returned, and never stored on the spec after the call.
+    """
+    secret: str | None = None
+    if target.secret_encrypted:
+        with contextlib.suppress(Exception):
+            secret = decrypt(target.secret_encrypted)
+    spec = DeployTargetSpec(
+        kind=target.kind,
+        host=target.host,
+        port=int(target.port),
+        username=target.username,
+        auth_type=target.auth_type,
+        secret=secret,
+        remote_path=target.remote_path,
+        known_fingerprint=target.host_fingerprint,
+    )
+    _ = allow_private
+    return spec, secret
+
+
+__all__ = ["MAX_LOG_TAIL_CHARS", "DeployService"]
+

apps/api/app/domain/delivery/models.py

@@ -157,3 +157,32 @@ class AppRuntime:
     @property
     def demo_url(self) -> str:
         return f"https://{self.demo_host}" if self.demo_host else ""
+
+
+class DeployStatus(StrEnum):
+    QUEUED = "queued"
+    RUNNING = "running"
+    SUCCEEDED = "succeeded"
+    FAILED = "failed"
+    #: A deploy that undid an earlier one. Kept as its own row so "what was
+    #: live when?" is answerable from the table alone.
+    ROLLED_BACK = "rolled_back"
+
+
+@dataclass(slots=True, kw_only=True)
+class ProductionDeploy:
+    organization_id: str
+    project_id: str
+    kind: str
+    host: str = ""
+    release_id: str | None = None
+    target_id: str | None = None
+    status: str = DeployStatus.QUEUED.value
+    message: str = ""
+    log_tail: str = ""
+    rolled_back_from: str | None = None
+    backup_path: str = ""
+    id: str = field(default_factory=new_id)
+    started_at: datetime | None = None
+    finished_at: datetime | None = None
+    created_at: datetime = field(default_factory=utcnow)

apps/api/app/infrastructure/adapters/deploy_ftp.py

@@ -0,0 +1,476 @@
+"""FTP / FTPS production deploy (D8).
+
+FTP is the weakest transport the platform supports, and the honest thing to do
+is say so in the code rather than pretend otherwise. Three limits shape the
+design:
+
+**There is no atomic switch.** FTP has no symlinks, so a deploy overwrites files
+in place. A failure halfway through therefore leaves a *half-updated* site, and
+the only defence is a backup taken first.
+
+**So a backup is mandatory, not optional.** Before anything is overwritten, the
+files that are about to change are downloaded into
+``ARTIFACTS_ROOT/<org>/<project>/ftp-backup-<ts>.tar.gz``. Rollback re-uploads
+that archive. Without it a bad FTP deploy is unrecoverable.
+
+**There is no host key to pin.** FTPS (explicit TLS) encrypts the session but
+says nothing about *which* server you reached, so the TOFU guarantee SSH gives
+does not exist here. The operator is expected to have verified the host out of
+band; the code cannot do it for them.
+
+The transport is injectable, so the deploy logic is testable without a server.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import io
+import logging
+import os
+import tarfile
+import time
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from typing import Any, Protocol, runtime_checkable
+
+from app.application.ports.deploy_target import DeployCheck, DeployTargetSpec
+from app.core.redact import redact_text
+from app.domain.platform.rules import assert_remote_host_allowed
+
+logger = logging.getLogger("agy.deploy.ftp")
+
+CONNECT_TIMEOUT_SECONDS = 15.0
+TRANSFER_TIMEOUT_SECONDS = 600.0
+
+#: The directory names a build output commonly lands in, tried in order when the
+#: app profile does not name one. Matches the framework catalog defaults.
+DEFAULT_BUILD_DIRS: tuple[str, ...] = ("dist", "build", "public", "out")
+
+#: A probe file written and deleted by `test_connection`.
+PROBE_FILENAME = ".smart-org-probe"
+
+
+@dataclass(slots=True)
+class FtpEntry:
+    path: str
+    size: int = 0
+
+
+@runtime_checkable
+class FtpTransport(Protocol):
+    """The remote operations an FTP deploy needs."""
+
+    async def connect(
+        self, *, host: str, port: int, username: str, password: str, tls: bool
+    ) -> None: ...
+
+    async def cwd(self, path: str) -> None: ...
+
+    async def download(self, remote_path: str) -> bytes: ...
+
+    async def upload(self, remote_path: str, data: bytes) -> None: ...
+
+    async def remove(self, remote_path: str) -> None: ...
+
+    async def list_files(self) -> list[FtpEntry]: ...
+
+    async def close(self) -> None: ...
+
+
+TransportFactory = Callable[[], FtpTransport]
+
+
+class AsyncFtpTransport:
+    """The real transport: one aioftp session per operation."""
+
+    def __init__(self) -> None:
+        # aioftp.Client is only imported inside connect(); the attribute is
+        # typed loosely so the module imports without the optional dependency
+        # present at type-check time.
+        self._client: Any = None
+        self._root = ""
+
+    async def connect(
+        self, *, host: str, port: int, username: str, password: str, tls: bool
+    ) -> None:
+        import aioftp
+
+        if tls:
+            # Explicit FTPS: connect in the clear, then upgrade. `aioftp` exposes
+            # this through `ssl` on the client; the certificate is not verified
+            # because a shared host's FTP certificate is rarely matchable.
+            import ssl
+
+            context = ssl.create_default_context()
+            context.check_hostname = False
+            context.verify_mode = ssl.CERT_NONE
+            self._client = aioftp.Client(ssl=context)
+        else:
+            self._client = aioftp.Client()
+        await self._client.connect(host, port, timeout=CONNECT_TIMEOUT_SECONDS)
+        await self._client.login(username, password)
+
+    async def cwd(self, path: str) -> None:
+        assert self._client is not None, "connect() first"
+        self._root = path.rstrip("/") or "/"
+        await self._client.change_directory(self._root)
+
+    async def download(self, remote_path: str) -> bytes:
+        assert self._client is not None
+        buffer = io.BytesIO()
+        async with self._client.download_stream(remote_path) as stream:
+            async for block in stream.iter_by_block():
+                buffer.write(block)
+        return buffer.getvalue()
+
+    async def upload(self, remote_path: str, data: bytes) -> None:
+        assert self._client is not None
+        async with self._client.upload_stream(remote_path) as stream:
+            await stream.write(data)
+
+    async def remove(self, remote_path: str) -> None:
+        assert self._client is not None
+        with contextlib.suppress(Exception):
+            await self._client.remove_file(remote_path)
+
+    async def list_files(self) -> list[FtpEntry]:
+        assert self._client is not None
+        entries: list[FtpEntry] = []
+        async for path, info in self._client.list(recursive=False):
+            entries.append(FtpEntry(path=str(path), size=int(info.get("size", 0) or 0)))
+        return entries
+
+    async def close(self) -> None:
+        if self._client is not None:
+            with contextlib.suppress(Exception):
+                await self._client.quit()
+            self._client = None
+
+
+class FtpDeployer:
+    """``DeployTargetTester`` and the production deploy itself, over FTP/FTPS."""
+
+    def __init__(
+        self,
+        *,
+        transport_factory: TransportFactory = AsyncFtpTransport,
+        artifacts_root: str | None = None,
+        allow_private_hosts: bool = False,
+        clock: Callable[[], float] = time.time,
+    ) -> None:
+        from app.core.config import settings
+
+        self._factory = transport_factory
+        self.artifacts_root = artifacts_root or settings.artifacts_root
+        self.allow_private_hosts = allow_private_hosts
+        self._clock = clock
+
+    # -- testing -------------------------------------------------------------
+
+    async def test_connection(self, *, target: DeployTargetSpec) -> DeployCheck:
+        """Log in, ``cwd`` the target path, and write/delete a probe file."""
+        try:
+            assert_remote_host_allowed(
+                f"ftp://{target.host}:{target.port}", allow_private=self.allow_private_hosts
+            )
+        except Exception as exc:  # noqa: BLE001
+            return DeployCheck(ok=False, message=str(exc))
+
+        transport = self._factory()
+        try:
+            await transport.connect(
+                host=target.host,
+                port=target.port,
+                username=target.username,
+                password=target.secret or "",
+                tls=target.kind in ("ftps",),
+            )
+            await transport.cwd(target.remote_path)
+            await transport.upload(PROBE_FILENAME, b"smart-org")
+            await transport.remove(PROBE_FILENAME)
+        except Exception as exc:  # noqa: BLE001
+            return DeployCheck(ok=False, message=_ftp_error_message(exc))
+        finally:
+            with contextlib.suppress(Exception):
+                await transport.close()
+
+        return DeployCheck(
+            ok=True,
+            message=(
+                "اتصال و مسیر مقصد تأیید شد. توجه: FTP امکان تأیید کلید میزبان را "
+                "ندارد و تغییر فایل‌ها اتمی نیست."
+            ),
+        )
+
+    # -- deploy --------------------------------------------------------------
+
+    async def deploy(
+        self,
+        *,
+        target: DeployTargetSpec,
+        artifact_path: str,
+        sha: str,
+        organization_id: str = "",
+        project_id: str = "",
+        build_dir: str | None = None,
+        on_output: Callable[[str], Awaitable[None]] | None = None,
+    ) -> tuple[bool, str, str]:
+        """Back up what will be overwritten, then upload the new build.
+
+        Returns ``(ok, backup_path, message)``. The backup path is returned even
+        on success: it is what a later rollback needs, and it is recorded in the
+        deploy row.
+        """
+        if not os.path.isfile(artifact_path):
+            return False, "", "فایل بستهٔ انتشار پیدا نشد."
+
+        # The artifact is a tar.gz of the whole release; the deployable build is
+        # a directory inside it. Unpack first, then choose.
+        extracted = _extract(artifact_path, self.artifacts_root, organization_id, project_id)
+        if extracted is None:
+            return False, "", "استخراج بستهٔ انتشار ناموفق بود."
+
+        source_dir = _pick_build_dir(extracted, build_dir)
+        if source_dir is None:
+            return False, "", "پوشهٔ خروجی ساخت در بستهٔ انتشار پیدا نشد."
+
+        transport = self._factory()
+        try:
+            await transport.connect(
+                host=target.host,
+                port=target.port,
+                username=target.username,
+                password=target.secret or "",
+                tls=target.kind == "ftps",
+            )
+            await transport.cwd(target.remote_path)
+
+            backup = await self._backup(
+                transport,
+                target=target,
+                organization_id=organization_id,
+                project_id=project_id,
+                sha=sha,
+            )
+            if on_output is not None and backup:
+                await on_output(f"[deploy] پشتیبان در {backup} ذخیره شد.\n")
+
+            uploaded = 0
+            for relative, data in _walk(source_dir):
+                remote = relative
+                await transport.upload(remote, data)
+                uploaded += 1
+            if on_output is not None:
+                await on_output(f"[deploy] {uploaded} فایل بارگذاری شد.\n")
+
+            return True, backup, "استقرار در production انجام شد (FTP)."
+        except Exception as exc:  # noqa: BLE001
+            logger.exception("ftp deploy failed")
+            return False, "", _ftp_error_message(exc)
+        finally:
+            with contextlib.suppress(Exception):
+                await transport.close()
+
+    async def _backup(
+        self,
+        transport: FtpTransport,
+        *,
+        target: DeployTargetSpec,
+        organization_id: str,
+        project_id: str,
+        sha: str,
+    ) -> str:
+        """Download every file we are about to overwrite, into one archive."""
+        try:
+            entries = await transport.list_files()
+        except Exception:  # noqa: BLE001 - an empty directory has nothing to back up
+            return ""
+
+        buffer = io.BytesIO()
+        captured = 0
+        with tarfile.open(fileobj=buffer, mode="w:gz") as tar:
+            for entry in entries:
+                name = os.path.basename(entry.path.rstrip("/"))
+                if not name or name == PROBE_FILENAME:
+                    continue
+                try:
+                    data = await transport.download(entry.path)
+                except Exception:  # noqa: BLE001 - a directory entry, or unreadable
+                    continue
+                info = tarfile.TarInfo(name=name)
+                info.size = len(data)
+                info.mtime = int(self._clock())
+                tar.addfile(info, io.BytesIO(data))
+                captured += 1
+
+        if not captured:
+            return ""
+
+        stamp = time.strftime("%Y%m%dT%H%M%S", time.gmtime(self._clock()))
+        directory = os.path.join(
+            self.artifacts_root, organization_id or "unknown", project_id or "unknown"
+        )
+        os.makedirs(directory, exist_ok=True)
+        path = os.path.join(directory, f"ftp-backup-{stamp}-{(sha or '')[:7]}.tar.gz")
+        with open(path, "wb") as handle:
+            handle.write(buffer.getvalue())
+        return path
+
+    # -- rollback ------------------------------------------------------------
+
+    async def rollback(
+        self,
+        *,
+        target: DeployTargetSpec,
+        backup_path: str,
+        on_output: Callable[[str], Awaitable[None]] | None = None,
+    ) -> tuple[bool, str]:
+        """Re-upload a backup archive over the live directory."""
+        if not backup_path or not os.path.isfile(backup_path):
+            return False, "پشتیبان بازگردانی پیدا نشد."
+
+        transport = self._factory()
+        try:
+            await transport.connect(
+                host=target.host,
+                port=target.port,
+                username=target.username,
+                password=target.secret or "",
+                tls=target.kind == "ftps",
+            )
+            await transport.cwd(target.remote_path)
+            restored = 0
+            with tarfile.open(backup_path, "r:gz") as tar:
+                for member in tar.getmembers():
+                    if not member.isfile():
+                        continue
+                    extracted = tar.extractfile(member)
+                    if extracted is None:
+                        continue
+                    await transport.upload(member.name, extracted.read())
+                    restored += 1
+            if on_output is not None:
+                await on_output(f"[rollback] {restored} فایل بازگردانی شد.\n")
+            return True, f"بازگردانی {restored} فایل انجام شد."
+        except Exception as exc:  # noqa: BLE001
+            logger.exception("ftp rollback failed")
+            return False, _ftp_error_message(exc)
+        finally:
+            with contextlib.suppress(Exception):
+                await transport.close()
+
+
+# ---------------------------------------------------------------------------
+# helpers
+# ---------------------------------------------------------------------------
+
+
+def _extract(
+    artifact_path: str, artifacts_root: str, organization_id: str, project_id: str
+) -> str | None:
+    """Unpack a release archive into a scratch directory and return its path."""
+    import shutil
+
+    stamp = time.strftime("%Y%m%dT%H%M%S", time.gmtime())
+    directory = os.path.join(
+        artifacts_root, organization_id or "unknown", project_id or "unknown", f"extract-{stamp}"
+    )
+    shutil.rmtree(directory, ignore_errors=True)
+    os.makedirs(directory, exist_ok=True)
+    try:
+        with tarfile.open(artifact_path, "r:gz") as tar:
+            tar.extractall(directory, filter=_safe_member)
+    except (tarfile.TarError, OSError):
+        return None
+    return directory
+
+
+def _safe_member(member: tarfile.TarInfo, dest: str) -> tarfile.TarInfo | None:
+    """Sanitise an archive member, or skip it.
+
+    An uploaded artifact is not trusted. Two things are refused outright:
+
+    * an absolute name or one containing ``..`` — the classic way a tarball
+      writes outside the directory it is extracted into; and
+    * the archive's own root entry (``.``), which would ``mkdir`` the extraction
+      directory itself. That is harmless on a normal filesystem but raises here,
+      and skipping it is correct regardless: the root already exists.
+    """
+    name = (member.name or "").replace("\\", "/")
+    name = name.lstrip("/")
+    while name.startswith("./"):
+        name = name[2:]
+    if name in ("", "."):
+        return None
+    parts = [part for part in name.split("/") if part not in ("", ".")]
+    if not parts or ".." in parts:
+        return None
+    if member.issym() or member.islnk():
+        # A link target is another way out of the extraction directory.
+        target = (member.linkname or "").lstrip("/")
+        if os.path.isabs(target) or ".." in target.split("/"):
+            return None
+    member.name = "/".join(parts)
+    return member
+
+
+def _pick_build_dir(artifact_path: str, explicit: str | None) -> str | None:
+    """The directory inside the artifact that holds the deployable build.
+
+    A build output is a *subdirectory* of the release archive (`dist/`, `build/`,
+    `public/`, `out/`), not the archive itself. Uploading the whole repository
+    over a static site is how a deploy publishes source code by accident.
+    """
+    if not os.path.isdir(artifact_path):
+        return None
+    if explicit:
+        candidate = os.path.join(artifact_path, explicit)
+        return candidate if os.path.isdir(candidate) else None
+    for name in DEFAULT_BUILD_DIRS:
+        candidate = os.path.join(artifact_path, name)
+        if os.path.isdir(candidate):
+            return candidate
+    # A static site has no build step: its root *is* the output.
+    return artifact_path
+
+
+def _walk(root: str):
+    """Yield ``(relative_path, bytes)`` for every file under ``root``."""
+    for directory, _dirs, files in os.walk(root):
+        for name in files:
+            absolute = os.path.join(directory, name)
+            relative = os.path.relpath(absolute, root).replace(os.sep, "/")
+            try:
+                with open(absolute, "rb") as handle:
+                    yield relative, handle.read()
+            except OSError:
+                continue
+
+
+def _ftp_error_message(exc: BaseException) -> str:
+    """A plain Persian message. Never the raw exception."""
+    text = str(exc).lower()
+    if "530" in text or "login" in text or "auth" in text:
+        return "ورود به سرور مقصد ناموفق بود."
+    if "550" in text or "not found" in text or "no such" in text:
+        return "مسیر مقصد روی سرور پیدا نشد."
+    if "timeout" in text or "timed out" in text:
+        return "اتصال به سرور مقصد در زمان مجاز برقرار نشد."
+    if "refused" in text or "unreachable" in text:
+        return "اتصال به سرور مقصد برقرار نشد."
+    if "certificate" in text or "ssl" in text or "tls" in text:
+        return "برقراری اتصال امن (FTPS) ناموفق بود."
+    return redact_text("استقرار روی سرور مقصد ناموفق بود.")
+
+
+__all__ = [
+    "CONNECT_TIMEOUT_SECONDS",
+    "DEFAULT_BUILD_DIRS",
+    "PROBE_FILENAME",
+    "TRANSFER_TIMEOUT_SECONDS",
+    "AsyncFtpTransport",
+    "FtpDeployer",
+    "FtpEntry",
+    "FtpTransport",
+]
+

apps/api/app/infrastructure/adapters/deploy_ssh.py

@@ -0,0 +1,475 @@
+"""SSH / SFTP production deploy (D8).
+
+Production is the one path in the product that a human must trigger, and the one
+that can destroy a running site. Three rules follow from that and shape
+everything here.
+
+**Nothing is uploaded into the live directory.** A release lands in
+``<remote_path>/releases/<timestamp>-<sha7>/`` and is switched in by moving a
+``current`` symlink. That makes the switch atomic (``ln -sfn`` + ``mv`` on the
+same filesystem) and makes rollback a pointer move rather than a re-upload.
+
+**The host key is pinned on first success (TOFU).** The first ``test_connection``
+that succeeds records the fingerprint; every later connection must match it. A
+changed key is a hard failure, not a warning — it is either a rebuilt server (in
+which case the operator re-tests deliberately) or an interception.
+
+**Credentials never touch the target's disk.** The password or key is held in
+memory for the connection only; the deploy scripts that run on the server are
+built from a fixed template and never interpolate a secret.
+
+The transport is injectable so the deploy logic — release numbering, pruning,
+the symlink dance, rollback — is testable without a live sshd.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import logging
+import os
+import shlex
+import time
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from typing import Any, Protocol, runtime_checkable
+
+from app.application.ports.deploy_target import DeployCheck, DeployTargetSpec
+from app.core.redact import redact_text
+from app.domain.platform.rules import assert_remote_host_allowed
+
+logger = logging.getLogger("agy.deploy.ssh")
+
+CONNECT_TIMEOUT_SECONDS = 15.0
+COMMAND_TIMEOUT_SECONDS = 600.0
+PROBE_TIMEOUT_SECONDS = 20.0
+
+#: How many releases to keep on the target. Five is enough to roll back past a
+#: bad deploy and a bad deploy of its fix, without filling the operator's disk.
+KEEP_RELEASES = 5
+
+#: Name of the pointer the web server is configured to serve from.
+CURRENT_LINK = "current"
+
+
+@dataclass(slots=True)
+class CommandOutcome:
+    exit_code: int
+    output: str = ""
+    timed_out: bool = False
+
+    @property
+    def ok(self) -> bool:
+        return self.exit_code == 0 and not self.timed_out
+
+
+@runtime_checkable
+class SshTransport(Protocol):
+    """The remote operations a deploy needs. Implemented over asyncssh."""
+
+    async def connect(
+        self,
+        *,
+        host: str,
+        port: int,
+        username: str,
+        password: str | None,
+        key: str | None,
+        known_hosts: str | None,
+    ) -> None: ...
+
+    def host_key_fingerprint(self) -> str: ...
+
+    async def run(
+        self, command: str, *, timeout: float = COMMAND_TIMEOUT_SECONDS
+    ) -> CommandOutcome: ...
+
+    async def upload(self, *, local_path: str, remote_path: str) -> None: ...
+
+    async def close(self) -> None: ...
+
+
+TransportFactory = Callable[[], SshTransport]
+
+
+class AsyncSshTransport:
+    """The real transport: one asyncssh connection per deploy."""
+
+    def __init__(self) -> None:
+        # asyncssh is imported lazily inside connect(); the attribute is typed
+        # loosely so this module imports without the dependency at hand.
+        self._conn: Any = None
+        self._fingerprint = ""
+
+    async def connect(
+        self,
+        *,
+        host: str,
+        port: int,
+        username: str,
+        password: str | None,
+        key: str | None,
+        known_hosts: str | None,
+    ) -> None:
+        import asyncssh
+
+        # `known_hosts=None` disables verification for this connection; the
+        # fingerprint is captured and compared by the caller, which is where the
+        # TOFU policy lives. Passing an explicit `known_hosts` string enforces
+        # the pinned key inside asyncssh as well.
+        kwargs: dict = {
+            "host": host,
+            "port": port,
+            "username": username,
+            "connect_timeout": CONNECT_TIMEOUT_SECONDS,
+            "known_hosts": known_hosts if known_hosts else None,
+        }
+        if key:
+            kwargs["client_keys"] = [asyncssh.import_private_key(key)]
+            kwargs["password"] = None
+        else:
+            kwargs["password"] = password or ""
+        self._conn = await asyncssh.connect(**kwargs)
+        self._fingerprint = self._read_fingerprint()
+
+    def _read_fingerprint(self) -> str:
+        """Best-effort host key digest of the connection we just made."""
+        try:
+            key = self._conn.get_server_host_key()
+            if key is None:
+                return ""
+            digest = key.get_fingerprint("sha256")
+            return f"SHA256:{digest}" if not str(digest).startswith("SHA256:") else str(digest)
+        except Exception:  # noqa: BLE001 - a missing fingerprint must not block a deploy
+            return ""
+
+    def host_key_fingerprint(self) -> str:
+        return self._fingerprint
+
+    async def run(
+        self, command: str, *, timeout: float = COMMAND_TIMEOUT_SECONDS
+    ) -> CommandOutcome:
+        import asyncssh
+
+        assert self._conn is not None, "connect() first"
+        try:
+            result = await self._conn.run(command, check=False, timeout=timeout)
+        except (asyncssh.Error, TimeoutError) as exc:
+            return CommandOutcome(exit_code=1, output=redact_text(str(exc))[:2000], timed_out=True)
+        output = f"{result.stdout or ''}{result.stderr or ''}"
+        return CommandOutcome(exit_code=int(result.exit_status or 0), output=output)
+
+    async def upload(self, *, local_path: str, remote_path: str) -> None:
+        assert self._conn is not None, "connect() first"
+        async with self._conn.start_sftp_client() as sftp:
+            await sftp.put(local_path, remote_path)
+
+    async def close(self) -> None:
+        if self._conn is not None:
+            self._conn.close()
+            with contextlib.suppress(Exception):
+                await self._conn.wait_closed()
+            self._conn = None
+
+
+class SshDeployer:
+    """``DeployTargetTester`` and the production deploy itself, over SSH."""
+
+    def __init__(
+        self,
+        *,
+        transport_factory: TransportFactory = AsyncSshTransport,
+        artifacts_root: str | None = None,
+        allow_private_hosts: bool = False,
+        clock: Callable[[], float] = time.time,
+    ) -> None:
+        from app.core.config import settings
+
+        self._factory = transport_factory
+        self.artifacts_root = artifacts_root or settings.artifacts_root
+        self.allow_private_hosts = allow_private_hosts
+        self._clock = clock
+
+    # -- testing -------------------------------------------------------------
+
+    async def test_connection(self, *, target: DeployTargetSpec) -> DeployCheck:
+        """Connect, pin the host key, and confirm the path is writable."""
+        try:
+            assert_remote_host_allowed(
+                f"ssh://{target.host}:{target.port}", allow_private=self.allow_private_hosts
+            )
+        except Exception as exc:  # noqa: BLE001 - translated to a plain message
+            return DeployCheck(ok=False, message=str(exc))
+
+        transport = self._factory()
+        try:
+            await transport.connect(
+                host=target.host,
+                port=target.port,
+                username=target.username,
+                password=target.secret if target.auth_type == "password" else None,
+                key=target.secret if target.auth_type == "key" else None,
+                known_hosts=None,
+            )
+        except Exception as exc:  # noqa: BLE001 - never surface the raw error
+            return DeployCheck(ok=False, message=_ssh_error_message(exc))
+        finally:
+            with contextlib.suppress(Exception):
+                await transport.close()
+
+        fingerprint = transport.host_key_fingerprint()
+        if target.known_fingerprint and fingerprint and fingerprint != target.known_fingerprint:
+            # A changed host key is a hard failure: it is either a rebuilt server
+            # or an interception, and only a human can tell the two apart.
+            return DeployCheck(
+                ok=False,
+                message="کلید میزبان تغییر کرده است؛ برای ادامه باید دوباره تأیید کنید.",
+                fingerprint=fingerprint,
+            )
+
+        probe = await _probe_writable(transport, target.remote_path)
+        if not probe.ok:
+            return DeployCheck(
+                ok=False,
+                message=f"مسیر «{target.remote_path}» قابل نوشتن نیست.",
+                fingerprint=fingerprint,
+            )
+        return DeployCheck(
+            ok=True, message="اتصال و مسیر مقصد تأیید شد.", fingerprint=fingerprint
+        )
+
+    # -- deploy --------------------------------------------------------------
+
+    async def deploy(
+        self,
+        *,
+        target: DeployTargetSpec,
+        artifact_path: str,
+        sha: str,
+        post_deploy_command: str | None = None,
+        on_output: Callable[[str], Awaitable[None]] | None = None,
+    ) -> tuple[bool, str, str]:
+        """Upload, extract, run the post-deploy command, switch ``current``.
+
+        Returns ``(ok, release_dir, message)``. On failure the ``current``
+        symlink is untouched, so a failed deploy leaves the previous release
+        serving — the same property the demo pipeline has.
+        """
+        if not os.path.isfile(artifact_path):
+            return False, "", "فایل بستهٔ انتشار پیدا نشد."
+
+        transport = self._factory()
+        try:
+            await transport.connect(
+                host=target.host,
+                port=target.port,
+                username=target.username,
+                password=target.secret if target.auth_type == "password" else None,
+                key=target.secret if target.auth_type == "key" else None,
+                known_hosts=None,
+            )
+        except Exception as exc:  # noqa: BLE001
+            return False, "", _ssh_error_message(exc)
+
+        try:
+            fingerprint = transport.host_key_fingerprint()
+            if target.known_fingerprint and fingerprint and fingerprint != target.known_fingerprint:
+                return False, "", "کلید میزبان تغییر کرده است."
+
+            release = _release_name(self._clock(), sha)
+            release_dir = f"{target.remote_path.rstrip('/')}/releases/{release}"
+
+            setup = await transport.run(
+                f"mkdir -p {shlex.quote(release_dir)} && "
+                f"mkdir -p {shlex.quote(target.remote_path.rstrip('/') + '/releases')}",
+                timeout=PROBE_TIMEOUT_SECONDS,
+            )
+            if not setup.ok:
+                return False, "", "ساخت پوشهٔ انتشار در سرور مقصد ناموفق بود."
+
+            archive = f"{release_dir}/release.tar.gz"
+            await transport.upload(local_path=artifact_path, remote_path=archive)
+            await _stream(on_output, f"[deploy] بسته در {archive} بارگذاری شد.\n")
+
+            extract = await transport.run(
+                f"cd {shlex.quote(release_dir)} && tar -xzf release.tar.gz && rm -f release.tar.gz",
+                timeout=COMMAND_TIMEOUT_SECONDS,
+            )
+            if on_output is not None:
+                await on_output(redact_text(extract.output))
+            if not extract.ok:
+                return False, release_dir, "استخراج بسته در سرور مقصد ناموفق بود."
+
+            if post_deploy_command:
+                post = await transport.run(
+                    f"cd {shlex.quote(release_dir)} && {post_deploy_command}",
+                    timeout=COMMAND_TIMEOUT_SECONDS,
+                )
+                if on_output is not None:
+                    await on_output(redact_text(post.output))
+                if not post.ok:
+                    # The previous release keeps serving: `current` never moved.
+                    return False, release_dir, "اجرای دستور پس از استقرار ناموفق بود."
+
+            switch = await transport.run(
+                _switch_script(target.remote_path, release),
+                timeout=PROBE_TIMEOUT_SECONDS,
+            )
+            if not switch.ok:
+                return False, release_dir, "تغییر نسخهٔ فعال ناموفق بود."
+
+            prune = await transport.run(
+                _prune_script(target.remote_path, KEEP_RELEASES),
+                timeout=PROBE_TIMEOUT_SECONDS,
+            )
+            if not prune.ok:
+                # Pruning is housekeeping; a full disk is not a failed deploy.
+                logger.warning("ssh deploy: pruning old releases failed: %s", prune.output[:200])
+
+            return True, release_dir, "استقرار در production انجام شد."
+        except Exception as exc:  # noqa: BLE001 - never leak the raw error
+            logger.exception("ssh deploy failed")
+            return False, "", _ssh_error_message(exc)
+        finally:
+            with contextlib.suppress(Exception):
+                await transport.close()
+
+    async def rollback(
+        self,
+        *,
+        target: DeployTargetSpec,
+        to_release: str | None = None,
+        post_deploy_command: str | None = None,
+        on_output: Callable[[str], Awaitable[None]] | None = None,
+    ) -> tuple[bool, str, str]:
+        """Re-point ``current`` at the previous release and re-run the hook."""
+        transport = self._factory()
+        try:
+            await transport.connect(
+                host=target.host,
+                port=target.port,
+                username=target.username,
+                password=target.secret if target.auth_type == "password" else None,
+                key=target.secret if target.auth_type == "key" else None,
+                known_hosts=None,
+            )
+        except Exception as exc:  # noqa: BLE001
+            return False, "", _ssh_error_message(exc)
+
+        try:
+            releases = await transport.run(
+                _list_releases_script(target.remote_path), timeout=PROBE_TIMEOUT_SECONDS
+            )
+            if not releases.ok:
+                return False, "", "خواندن فهرست نسخه‌ها ناموفق بود."
+            names = [line.strip() for line in releases.output.splitlines() if line.strip()]
+            if len(names) < 2:
+                return False, "", "نسخهٔ قبلی‌ای برای بازگردانی وجود ندارد."
+
+            # `ls -1t` is newest first, so the previous release is the second.
+            current = names[0]
+            target_name = to_release or names[1]
+            if target_name == current:
+                target_name = names[1]
+
+            switch = await transport.run(
+                _switch_script(target.remote_path, target_name), timeout=PROBE_TIMEOUT_SECONDS
+            )
+            if not switch.ok:
+                return False, "", "بازگردانی ناموفق بود."
+
+            if post_deploy_command:
+                post = await transport.run(
+                    f"cd {shlex.quote(target.remote_path.rstrip('/') + '/releases/' + target_name)}"
+                    f" && {post_deploy_command}",
+                    timeout=COMMAND_TIMEOUT_SECONDS,
+                )
+                if on_output is not None:
+                    await on_output(redact_text(post.output))
+                if not post.ok:
+                    return False, target_name, "دستور پس از بازگردانی ناموفق بود."
+
+            return True, target_name, f"بازگردانی به نسخهٔ {target_name} انجام شد."
+        except Exception as exc:  # noqa: BLE001
+            return False, "", _ssh_error_message(exc)
+        finally:
+            with contextlib.suppress(Exception):
+                await transport.close()
+
+
+# ---------------------------------------------------------------------------
+# remote scripts — built from constants, never from user input
+# ---------------------------------------------------------------------------
+
+
+def _release_name(now: float, sha: str) -> str:
+    stamp = time.strftime("%Y%m%dT%H%M%S", time.gmtime(now))
+    return f"{stamp}-{(sha or 'unknown')[:7]}"
+
+
+def _switch_script(remote_path: str, release: str) -> str:
+    """Move the ``current`` symlink atomically.
+
+    A relative symlink into ``releases/`` plus ``mv -T`` on the same filesystem
+    is the only switch that is atomic *and* survives the directory being moved.
+    """
+    base = shlex.quote(remote_path.rstrip("/"))
+    rel = shlex.quote(f"releases/{release}")
+    link = shlex.quote(f"{remote_path.rstrip('/')}/{CURRENT_LINK}")
+    tmp = shlex.quote(f"{remote_path.rstrip('/')}/.{CURRENT_LINK}.new")
+    return f"ln -sfn {rel} {tmp} && mv -T {tmp} {link} && cd {base} && ls -l {link}"
+
+
+def _prune_script(remote_path: str, keep: int) -> str:
+    """Delete all but the newest ``keep`` release directories."""
+    releases = shlex.quote(f"{remote_path.rstrip('/')}/releases")
+    return (
+        f"cd {releases} && ls -1t | tail -n +{keep + 1} | "
+        f"while read -r old; do rm -rf -- \"$old\"; done"
+    )
+
+
+def _list_releases_script(remote_path: str) -> str:
+    releases = shlex.quote(f"{remote_path.rstrip('/')}/releases")
+    return f"cd {releases} && ls -1t"
+
+
+async def _probe_writable(transport: SshTransport, remote_path: str) -> CommandOutcome:
+    base = shlex.quote(remote_path.rstrip("/") or "/")
+    return await transport.run(
+        f"mkdir -p {base} && test -w {base} && "
+        f"probe={base}/.smart-org-probe && touch $probe && rm -f $probe",
+        timeout=PROBE_TIMEOUT_SECONDS,
+    )
+
+
+async def _stream(callback: Callable[[str], Awaitable[None]] | None, text: str) -> None:
+    if callback is not None:
+        await callback(text)
+
+
+def _ssh_error_message(exc: BaseException) -> str:
+    """A plain Persian message. Never the raw exception, which can carry the host."""
+    text = str(exc).lower()
+    if "permission denied" in text or "auth" in text:
+        return "احراز هویت روی سرور مقصد ناموفق بود."
+    if "timed out" in text or "timeout" in text:
+        return "اتصال به سرور مقصد در زمان مجاز برقرار نشد."
+    if "host key" in text:
+        return "کلید میزبان تأیید نشد."
+    if "no such file" in text or "not found" in text:
+        return "مسیر یا فایل مقصد پیدا نشد."
+    if "connection refused" in text or "unreachable" in text:
+        return "اتصال به سرور مقصد برقرار نشد."
+    return "استقرار در سرور مقصد ناموفق بود."
+
+
+__all__ = [
+    "COMMAND_TIMEOUT_SECONDS",
+    "CONNECT_TIMEOUT_SECONDS",
+    "CURRENT_LINK",
+    "KEEP_RELEASES",
+    "AsyncSshTransport",
+    "CommandOutcome",
+    "SshDeployer",
+    "SshTransport",
+]

apps/api/app/infrastructure/adapters/deploy_target.py

@@ -0,0 +1,56 @@
+"""The deploy-target tester that dispatches on ``kind``.
+
+``DeployTargetTester`` is one contract with two very different implementations,
+and the wizard does not care which one it is talking to: it hands over a
+:class:`DeployTargetSpec` and wants a yes/no and a plain-language reason.
+
+The dispatcher exists so the wizard never has to know that `ssh` and `sftp` are
+the same protocol while `ftp` is a different world with no host key, no atomic
+switch and a mandatory backup. The *differences* are documented on the deployers
+themselves; the caller sees one shape.
+"""
+
+from __future__ import annotations
+
+from app.application.ports.deploy_target import (
+    DEPLOY_KINDS,
+    DeployCheck,
+    DeployTargetSpec,
+)
+from app.core.config import settings
+from app.core.errors import ValidationError
+from app.infrastructure.adapters.deploy_ftp import FtpDeployer
+from app.infrastructure.adapters.deploy_ssh import SshDeployer
+
+
+class DispatchingDeployTargetTester:
+    """``test_connection`` routed to the right transport for ``spec.kind``."""
+
+    def __init__(
+        self,
+        *,
+        ssh: SshDeployer | None = None,
+        ftp: FtpDeployer | None = None,
+    ) -> None:
+        allow_private = settings.deploy_allow_private_hosts
+        self.ssh = ssh or SshDeployer(allow_private_hosts=allow_private)
+        self.ftp = ftp or FtpDeployer(allow_private_hosts=allow_private)
+
+    async def test_connection(self, *, target: DeployTargetSpec) -> DeployCheck:
+        if target.kind not in DEPLOY_KINDS:
+            raise ValidationError(
+                "نوع مقصد استقرار پشتیبانی نمی‌شود؛ مقادیر مجاز: "
+                + "، ".join(DEPLOY_KINDS)
+            )
+        if not target.remote_path.strip():
+            raise ValidationError("مسیر مقصد روی سرور مشخص نشده است.")
+        if target.kind in ("ssh", "sftp"):
+            return await self.ssh.test_connection(target=target)
+        return await self.ftp.test_connection(target=target)
+
+    def deployer_for(self, kind: str):
+        """The object that can actually deploy to this kind of target."""
+        return self.ssh if kind in ("ssh", "sftp") else self.ftp
+
+
+__all__ = ["DispatchingDeployTargetTester"]

apps/api/app/infrastructure/db/models_delivery.py

@@ -121,8 +121,46 @@ class AppRuntimeRow(Base):
     )
 
 
+class ProductionDeployRow(Base):
+    """One human-triggered production deploy (and its rollbacks).
+
+    ``log_tail`` is bounded exactly like a pipeline step's: the full output is
+    streamed, the tail is kept for the post-mortem. ``rolled_back_from`` makes a
+    rollback its own row rather than an edit of the deploy it undid — "what was
+    live when?" must be answerable from the table alone.
+    """
+
+    __tablename__ = "production_deploys"
+    __table_args__ = (Index("ix_prod_deploy_project_created", "project_id", "created_at"),)
+
+    id: Mapped[str] = pk()
+    organization_id: Mapped[str] = mapped_column(
+        ForeignKey("organizations.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    project_id: Mapped[str] = mapped_column(
+        ForeignKey("projects.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    release_id: Mapped[str | None] = mapped_column(
+        ForeignKey("releases.id", ondelete="SET NULL"), nullable=True
+    )
+    target_id: Mapped[str | None] = mapped_column(
+        ForeignKey("deploy_targets.id", ondelete="SET NULL"), nullable=True
+    )
+    kind: Mapped[str] = mapped_column(String(8), nullable=False)
+    host: Mapped[str] = mapped_column(String(255), default="", nullable=False)
+    status: Mapped[str] = mapped_column(String(16), default="queued", nullable=False)
+    message: Mapped[str] = mapped_column(Text, default="", nullable=False)
+    log_tail: Mapped[str] = mapped_column(Text, default="", nullable=False)
+    rolled_back_from: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    backup_path: Mapped[str] = mapped_column(String(512), default="", nullable=False)
+    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    created_at: Mapped[datetime] = created()
+
+
 __all__ = [
     "AppRuntimeRow",
+    "ProductionDeployRow",
     "MAX_LOG_TAIL_CHARS",
     "PipelineRunRow",
     "PipelineStepRow",

apps/api/app/infrastructure/repositories/delivery.py

@@ -8,8 +8,10 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.domain.base import utcnow
 from app.domain.delivery.models import (
     AppRuntime,
+    DeployStatus,
     PipelineRun,
     PipelineStep,
+    ProductionDeploy,
     Release,
     RunStatus,
     StepStatus,
@@ -18,6 +20,7 @@ from app.infrastructure.db.models_delivery import (
     AppRuntimeRow,
     PipelineRunRow,
     PipelineStepRow,
+    ProductionDeployRow,
     ReleaseRow,
 )
 
@@ -440,11 +443,131 @@ class AppRuntimeRepository:
         await self.session.flush()
 
 
+def _deploy(row: ProductionDeployRow) -> ProductionDeploy:
+    return ProductionDeploy(
+        id=row.id,
+        organization_id=row.organization_id,
+        project_id=row.project_id,
+        kind=row.kind,
+        host=row.host,
+        release_id=row.release_id,
+        target_id=row.target_id,
+        status=row.status,
+        message=row.message,
+        log_tail=row.log_tail,
+        rolled_back_from=row.rolled_back_from,
+        backup_path=row.backup_path,
+        started_at=row.started_at,
+        finished_at=row.finished_at,
+        created_at=row.created_at,
+    )
+
+
+class ProductionDeployRepository:
+    """The production deploy ledger."""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+
+    async def get(self, deploy_id: str) -> ProductionDeploy | None:
+        row = await self.session.get(ProductionDeployRow, deploy_id)
+        return _deploy(row) if row else None
+
+    async def get_for_org(self, organization_id: str, deploy_id: str) -> ProductionDeploy | None:
+        row = await self.session.scalar(
+            select(ProductionDeployRow).where(
+                ProductionDeployRow.id == deploy_id,
+                ProductionDeployRow.organization_id == organization_id,
+            )
+        )
+        return _deploy(row) if row else None
+
+    async def list_for_project(
+        self, *, project_id: str, limit: int = 20
+    ) -> list[ProductionDeploy]:
+        stmt = (
+            select(ProductionDeployRow)
+            .where(ProductionDeployRow.project_id == project_id)
+            .order_by(ProductionDeployRow.created_at.desc())
+            .limit(min(max(limit, 1), 100))
+        )
+        return [_deploy(r) for r in (await self.session.execute(stmt)).scalars()]
+
+    async def active_for_project(self, project_id: str) -> ProductionDeploy | None:
+        """One production deploy per project at a time.
+
+        Two concurrent deploys to the same target would interleave their writes
+        on a shared host, which is exactly the failure FTP has no way to undo.
+        """
+        row = await self.session.scalar(
+            select(ProductionDeployRow)
+            .where(
+                ProductionDeployRow.project_id == project_id,
+                ProductionDeployRow.status.in_(
+                    [DeployStatus.QUEUED.value, DeployStatus.RUNNING.value]
+                ),
+            )
+            .limit(1)
+        )
+        return _deploy(row) if row else None
+
+    async def create(self, deploy: ProductionDeploy) -> ProductionDeploy:
+        row = ProductionDeployRow(
+            id=deploy.id,
+            organization_id=deploy.organization_id,
+            project_id=deploy.project_id,
+            release_id=deploy.release_id,
+            target_id=deploy.target_id,
+            kind=deploy.kind,
+            host=deploy.host,
+            status=deploy.status,
+            message=deploy.message,
+            log_tail=deploy.log_tail,
+            rolled_back_from=deploy.rolled_back_from,
+            backup_path=deploy.backup_path,
+            started_at=deploy.started_at,
+            finished_at=deploy.finished_at,
+            created_at=deploy.created_at,
+        )
+        self.session.add(row)
+        await self.session.flush()
+        return _deploy(row)
+
+    async def finish(
+        self,
+        deploy_id: str,
+        *,
+        status: str,
+        message: str = "",
+        log_tail: str = "",
+        backup_path: str = "",
+    ) -> None:
+        row = await self.session.get(ProductionDeployRow, deploy_id)
+        if row is None:
+            return
+        row.status = status
+        row.message = message[:2_000]
+        row.log_tail = log_tail[-65_536:]
+        if backup_path:
+            row.backup_path = backup_path
+        row.finished_at = utcnow()
+        await self.session.flush()
+
+    async def mark_running(self, deploy_id: str) -> None:
+        row = await self.session.get(ProductionDeployRow, deploy_id)
+        if row is None:
+            return
+        row.status = DeployStatus.RUNNING.value
+        row.started_at = utcnow()
+        await self.session.flush()
+
+
 __all__ = [
     "ACTIVE_STATUSES",
     "AppRuntimeRepository",
     "PipelineRunRepository",
     "PipelineStepRepository",
+    "ProductionDeployRepository",
     "ReleaseRepository",
     "StepStatus",
 ]

apps/api/app/infrastructure/repositories/projects.py

@@ -14,6 +14,7 @@ from app.domain.projects.models import (
     ProjectGitRepo,
 )
 from app.infrastructure.db.models import (
+    DeployTargetRow,
     PreviewDomainRow,
     PreviewPortRow,
     ProjectExternalGitRow,
@@ -164,6 +165,61 @@ class ProjectGitCredentialRepository:
         return credential
 
 
+class DeployTargetRepository:
+    """Persistence for a project's production deploy target.
+
+    The secret is stored encrypted and is only ever handed back through
+    :meth:`get_secret`, so a code path that merely *renders* a target never
+    touches the plaintext.
+    """
+
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+
+    async def get_for_project(self, project_id: str) -> DeployTargetRow | None:
+        return await self.session.scalar(
+            select(DeployTargetRow)
+            .where(DeployTargetRow.project_id == project_id, DeployTargetRow.is_active.is_(True))
+            .order_by(DeployTargetRow.created_at.desc())
+            .limit(1)
+        )
+
+    async def list_for_project(self, project_id: str) -> list[DeployTargetRow]:
+        stmt = (
+            select(DeployTargetRow)
+            .where(DeployTargetRow.project_id == project_id)
+            .order_by(DeployTargetRow.created_at.desc())
+        )
+        return list((await self.session.execute(stmt)).scalars())
+
+    async def upsert(self, *, values: dict) -> DeployTargetRow:
+        """Create or replace the single active target of a project."""
+        project_id = str(values["project_id"])
+        existing = await self.get_for_project(project_id)
+        if existing is None:
+            row = DeployTargetRow(**values)
+            self.session.add(row)
+        else:
+            row = existing
+            for key, value in values.items():
+                setattr(row, key, value)
+        await self.session.flush()
+        return row
+
+    async def record_test(
+        self, *, target_id: str, ok: bool, message: str, fingerprint: str | None = None
+    ) -> None:
+        row = await self.session.get(DeployTargetRow, target_id)
+        if row is None:
+            return
+        row.last_tested_at = utcnow()
+        row.last_test_ok = ok
+        row.last_test_message = message[:500]
+        if fingerprint:
+            row.host_fingerprint = fingerprint
+        await self.session.flush()
+
+
 class ProjectRepository:
     def __init__(self, session: AsyncSession) -> None:
         self.session = session

apps/api/app/interfaces/http/deps.py

@@ -204,3 +204,12 @@ async def get_delivery_service(
     from app.application.container import build_delivery_service
 
     return build_delivery_service(session)
+
+
+async def get_deploy_service(
+    session: AsyncSession = Depends(get_session),
+):
+    """Build the production deploy service with the configured transports."""
+    from app.application.container import build_deploy_service
+
+    return build_deploy_service(session)

apps/api/app/interfaces/http/routers/delivery.py

@@ -21,12 +21,14 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.application.services.audit_service import record_audit
 from app.application.services.delivery_service import DeliveryService
+from app.application.services.deploy_service import DeployService
 from app.core.database import get_session
 from app.domain.identity.models import CAN_MANAGE_PROJECTS
 from app.interfaces.http.deps import (
     OrgContext,
     client_meta,
     get_delivery_service,
+    get_deploy_service,
     get_org_context,
 )
 
@@ -248,3 +250,210 @@ async def rollback_demo(
         ip=ip, user_agent=ua,
     )
     return _release_out(release)
+
+
+# ---------------------------------------------------------------------------
+# Production (always human-triggered, always confirmed)
+# ---------------------------------------------------------------------------
+
+
+class ProductionDeployOut(BaseModel):
+    id: str
+    project_id: str
+    kind: str
+    host: str
+    status: str
+    message: str
+    log_tail: str
+    release_id: str | None
+    rolled_back_from: str | None
+    started_at: datetime | None
+    finished_at: datetime | None
+    created_at: datetime
+
+
+class ProductionTargetOut(BaseModel):
+    """The target as a *user* sees it: never a secret, only a hint that one exists."""
+
+    kind: str
+    host: str
+    port: int
+    username: str
+    auth_type: str
+    remote_path: str
+    has_secret: bool
+    host_fingerprint: str | None
+    last_tested_at: datetime | None
+    last_test_ok: bool | None
+    last_test_message: str | None
+
+
+class ProductionOverviewOut(BaseModel):
+    target: ProductionTargetOut | None
+    deploys: list[ProductionDeployOut]
+
+
+class DeployConfirmRequest(BaseModel):
+    project_id: str = Field(min_length=1, max_length=32)
+    #: The organization's slug, typed back. Not ceremony: this is the only
+    #: action in the product the platform cannot undo.
+    confirm: str = Field(min_length=1, max_length=120)
+    release_id: str | None = Field(default=None, max_length=32)
+
+
+class RollbackConfirmRequest(BaseModel):
+    project_id: str = Field(min_length=1, max_length=32)
+    confirm: str = Field(min_length=1, max_length=120)
+    deploy_id: str | None = Field(default=None, max_length=32)
+
+
+class TargetTestOut(BaseModel):
+    ok: bool
+    message: str
+
+
+def _deploy_out(deploy) -> ProductionDeployOut:  # noqa: ANN001 - a domain dataclass
+    return ProductionDeployOut(
+        id=deploy.id, project_id=deploy.project_id, kind=deploy.kind, host=deploy.host,
+        status=deploy.status, message=deploy.message, log_tail=deploy.log_tail,
+        release_id=deploy.release_id, rolled_back_from=deploy.rolled_back_from,
+        started_at=deploy.started_at, finished_at=deploy.finished_at,
+        created_at=deploy.created_at,
+    )
+
+
+def _target_out(target) -> ProductionTargetOut:  # noqa: ANN001
+    return ProductionTargetOut(
+        kind=target.kind, host=target.host, port=target.port, username=target.username,
+        auth_type=target.auth_type, remote_path=target.remote_path,
+        # The secret itself is never returned — only that one is configured.
+        has_secret=bool(target.secret_encrypted),
+        host_fingerprint=target.host_fingerprint,
+        last_tested_at=target.last_tested_at, last_test_ok=target.last_test_ok,
+        last_test_message=target.last_test_message,
+    )
+
+
+@router.get("/production", response_model=ProductionOverviewOut)
+async def production_overview(
+    project_id: str = Query(min_length=1),
+    context: OrgContext = Depends(get_org_context),
+    deploys: DeployService = Depends(get_deploy_service),
+) -> ProductionOverviewOut:
+    """The target summary and the deploy history (admin+ — it names a host)."""
+    context.require(CAN_MANAGE_PROJECTS)
+    target = await deploys.get_target(
+        organization_id=context.org_id, project_id=project_id
+    )
+    history = await deploys.list_deploys(
+        organization_id=context.org_id, project_id=project_id
+    )
+    return ProductionOverviewOut(
+        target=_target_out(target) if target else None,
+        deploys=[_deploy_out(d) for d in history],
+    )
+
+
+@router.post("/production/test", response_model=TargetTestOut)
+async def test_production_target(
+    payload: ManualRunRequest,
+    context: OrgContext = Depends(get_org_context),
+    deploys: DeployService = Depends(get_deploy_service),
+) -> TargetTestOut:
+    """Test the configured target and record the result (admin+)."""
+    context.require(CAN_MANAGE_PROJECTS)
+    ok, message = await deploys.test_target(
+        organization_id=context.org_id,
+        project_id=payload.project_id,
+        actor_user_id=context.user.id,
+    )
+    return TargetTestOut(ok=ok, message=message)
+
+
+@router.post("/production/deploy", response_model=ProductionDeployOut)
+async def deploy_to_production(
+    payload: DeployConfirmRequest,
+    request: Request,
+    context: OrgContext = Depends(get_org_context),
+    deploys: DeployService = Depends(get_deploy_service),
+    session: AsyncSession = Depends(get_session),
+) -> ProductionDeployOut:
+    """Ship a release to the customer's server (org owner/admin, confirmed)."""
+    context.require(CAN_MANAGE_PROJECTS)
+    deploy = await deploys.deploy_production(
+        organization_id=context.org_id,
+        project_id=payload.project_id,
+        actor_user_id=context.user.id,
+        confirm=payload.confirm,
+        release_id=payload.release_id,
+    )
+    ip, ua = client_meta(request)
+    await record_audit(
+        session, action="deploy.production", organization_id=context.org_id,
+        actor_user_id=context.user.id, target_type="production_deploy", target_id=deploy.id,
+        metadata={
+            "project_id": deploy.project_id,
+            "kind": deploy.kind,
+            "status": deploy.status,
+            "release_id": deploy.release_id,
+        },
+        ip=ip, user_agent=ua,
+    )
+    await _emit_deploy_status(context.org_id, deploy)
+    return _deploy_out(deploy)
+
+
+@router.post("/production/rollback", response_model=ProductionDeployOut)
+async def rollback_production(
+    payload: RollbackConfirmRequest,
+    request: Request,
+    context: OrgContext = Depends(get_org_context),
+    deploys: DeployService = Depends(get_deploy_service),
+    session: AsyncSession = Depends(get_session),
+) -> ProductionDeployOut:
+    """Undo a production deploy (org owner/admin, confirmed, audited)."""
+    context.require(CAN_MANAGE_PROJECTS)
+    deploy = await deploys.rollback_production(
+        organization_id=context.org_id,
+        project_id=payload.project_id,
+        actor_user_id=context.user.id,
+        confirm=payload.confirm,
+        deploy_id=payload.deploy_id,
+    )
+    ip, ua = client_meta(request)
+    await record_audit(
+        session, action="deploy.production_rollback", organization_id=context.org_id,
+        actor_user_id=context.user.id, target_type="production_deploy", target_id=deploy.id,
+        metadata={
+            "project_id": deploy.project_id,
+            "kind": deploy.kind,
+            "status": deploy.status,
+            "rolled_back_from": deploy.rolled_back_from,
+        },
+        ip=ip, user_agent=ua,
+    )
+    await _emit_deploy_status(context.org_id, deploy)
+    return _deploy_out(deploy)
+
+
+async def _emit_deploy_status(organization_id: str, deploy) -> None:  # noqa: ANN001
+    """Push the outcome on the tenant channel. Never raises."""
+    import contextlib
+
+    from app.application.ports.realtime import RealtimeEvent, get_publisher, org_channel
+
+    with contextlib.suppress(Exception):
+        await get_publisher().publish(
+            RealtimeEvent(
+                type="deploy.status",
+                channels=(org_channel(organization_id),),
+                data={
+                    "project_id": deploy.project_id,
+                    "deploy_id": deploy.id,
+                    "kind": deploy.kind,
+                    "status": deploy.status,
+                    "message": deploy.message,
+                },
+                organization_id=organization_id,
+            )
+        )

apps/api/tests/test_deploy.py

@@ -0,0 +1,787 @@
+"""§v2-A3.5 — production deploy over SSH / SFTP / FTP / FTPS (D8).
+
+Production is the one action the platform cannot undo, so the tests are about the
+guards around it as much as the deploy itself:
+
+* a member who is not an org admin cannot deploy;
+* the org slug has to be typed back;
+* a changed host key is a hard failure;
+* an SSRF target is refused before a connection is attempted;
+* a secret never reaches a log, an event, a response or a database column;
+* and the mechanics — release directories, the ``current`` symlink, keeping five
+  releases, rollback — behave the way the design says they do.
+
+The SSH and FTP transports are faked, so the suite needs no server. The deploy
+*logic* is real.
+"""
+
+from __future__ import annotations
+
+import os
+import tarfile
+
+import pytest
+import pytest_asyncio
+from httpx import AsyncClient
+
+from app.application.ports.deploy_target import DeployCheck, DeployTargetSpec
+from app.application.services.deploy_service import DeployService
+from app.core.crypto import encrypt
+from app.domain.delivery.models import DeployStatus
+from app.domain.identity.models import PlatformRole
+from app.infrastructure.adapters.deploy_ftp import FtpDeployer, FtpEntry
+from app.infrastructure.adapters.deploy_ssh import (
+    CURRENT_LINK,
+    KEEP_RELEASES,
+    CommandOutcome,
+    SshDeployer,
+)
+from app.infrastructure.repositories.delivery import (
+    ProductionDeployRepository,
+    ReleaseRepository,
+)
+from app.infrastructure.repositories.identity import OrganizationRepository
+from app.infrastructure.repositories.projects import (
+    DeployTargetRepository,
+    ProjectRepository,
+)
+from tests.conftest import auth_header, signup
+
+SECRET = "ssh-password-that-must-never-leak"
+
+
+# ---------------------------------------------------------------------------
+# fake transports
+# ---------------------------------------------------------------------------
+
+
+class FakeSshTransport:
+    """An in-memory SSH server: a filesystem plus a scripted command table."""
+
+    def __init__(self, *, fingerprint: str = "SHA256:abc123", fail_on: str = "") -> None:
+        self.commands: list[str] = []
+        self.uploads: list[tuple[str, str]] = []
+        self.fingerprint = fingerprint
+        self.fail_on = fail_on
+        self.closed = False
+        #: The remote filesystem, as a set of paths.
+        self.files: set[str] = set()
+
+    async def connect(self, **kwargs) -> None:  # noqa: ANN003
+        self.connect_kwargs = kwargs
+        if self.fail_on == "connect":
+            raise OSError("permission denied")
+
+    def host_key_fingerprint(self) -> str:
+        return self.fingerprint
+
+    async def run(self, command: str, *, timeout: float = 600.0) -> CommandOutcome:
+        self.commands.append(command)
+        if self.fail_on and self.fail_on in command:
+            return CommandOutcome(exit_code=1, output="failed on purpose")
+
+        if "ls -1t" in command and "releases" in command:
+            releases = sorted(
+                {p.split("/releases/")[1].split("/")[0] for p in self.files if "/releases/" in p},
+                reverse=True,
+            )
+            return CommandOutcome(exit_code=0, output="\n".join(releases))
+        if "test -w" in command:
+            return CommandOutcome(exit_code=0, output="")
+        if "ln -sfn" in command:
+            # Record the switch so a test can assert the pointer moved.
+            for part in command.split():
+                if part.startswith("releases/"):
+                    self.files.add(f"/srv/app/{CURRENT_LINK}->{part}")
+            return CommandOutcome(exit_code=0, output="")
+        if "tail -n" in command:
+            self.files = {
+                p for p in self.files if not p.endswith(tuple(
+                    f"release-{i}" for i in range(1, 20)
+                ))
+            }
+            return CommandOutcome(exit_code=0, output="")
+        return CommandOutcome(exit_code=0, output="")
+
+    async def upload(self, *, local_path: str, remote_path: str) -> None:
+        self.uploads.append((local_path, remote_path))
+        self.files.add(remote_path)
+
+    async def close(self) -> None:
+        self.closed = True
+
+
+class FakeFtpTransport:
+    def __init__(self, *, fail_on: str = "") -> None:
+        self.commands: list[str] = []
+        self.uploaded: dict[str, bytes] = {}
+        self.removed: list[str] = []
+        self.fail_on = fail_on
+        self.cwd_path = ""
+        self.connected = False
+
+    async def connect(self, **kwargs) -> None:  # noqa: ANN003
+        self.connect_kwargs = kwargs
+        if self.fail_on == "connect":
+            raise OSError("530 login incorrect")
+        self.connected = True
+
+    async def cwd(self, path: str) -> None:
+        if self.fail_on == "cwd":
+            raise OSError("550 no such directory")
+        self.cwd_path = path
+
+    async def download(self, remote_path: str) -> bytes:
+        return self.uploaded.get(remote_path, b"old-content")
+
+    async def upload(self, remote_path: str, data: bytes) -> None:
+        if self.fail_on == "upload":
+            raise OSError("552 quota exceeded")
+        self.uploaded[remote_path] = data
+
+    async def remove(self, remote_path: str) -> None:
+        self.removed.append(remote_path)
+        self.uploaded.pop(remote_path, None)
+
+    async def list_files(self) -> list[FtpEntry]:
+        return [FtpEntry(path=name, size=len(data)) for name, data in self.uploaded.items()]
+
+    async def close(self) -> None:
+        self.connected = False
+
+
+class FakeDeployTargetTester:
+    """Dispatches to injected deployers, like the real one but with no network."""
+
+    def __init__(self, *, ssh: SshDeployer, ftp: FtpDeployer) -> None:
+        self.ssh = ssh
+        self.ftp = ftp
+        self.calls: list[DeployTargetSpec] = []
+
+    async def test_connection(self, *, target: DeployTargetSpec) -> DeployCheck:
+        self.calls.append(target)
+        if target.kind in ("ssh", "sftp"):
+            return await self.ssh.test_connection(target=target)
+        return await self.ftp.test_connection(target=target)
+
+    def deployer_for(self, kind: str):
+        return self.ssh if kind in ("ssh", "sftp") else self.ftp
+
+
+# ---------------------------------------------------------------------------
+# fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _platform_seed(seeded):
+    return seeded
+
+
+def _artifact(tmp_path, *, build_dir: str = "") -> str:
+    """A release archive shaped like the pipeline produces."""
+    source = tmp_path / "artifact-src"
+    target = source / build_dir if build_dir else source
+    target.mkdir(parents=True, exist_ok=True)
+    (target / "index.html").write_text("<h1>hello</h1>")
+    (target / "app.js").write_text("console.log('hi')")
+    if build_dir:
+        (source / "README.md").write_text("not deployed")
+
+    archive = tmp_path / "release.tar.gz"
+    with tarfile.open(archive, "w:gz") as tar:
+        # Add the children rather than the directory itself: a bare "." member
+        # is the archive root, which some extractors try to mkdir.
+        for child in sorted(source.iterdir()):
+            tar.add(child, arcname=child.name)
+    return str(archive)
+
+
+@pytest_asyncio.fixture
+async def tenant(client: AsyncClient, session_factory, tmp_path):
+    """An org + app + live release + a configured SSH deploy target."""
+    token = (
+        await signup(
+            client,
+            "deploy-owner@example.com",
+            name="مدیر استقرار",
+            platform_role=PlatformRole.OWNER,
+        )
+    )["access_token"]
+    headers = auth_header(token)
+    org = (
+        await client.post("/api/organizations", json={"name": "شرکت استقرار"}, headers=headers)
+    ).json()
+    org_id, slug = org["id"], org["slug"]
+    project_id = (
+        await client.post(
+            f"/api/organizations/{org_id}/projects",
+            json={"name": "اپ استقرار"},
+            headers=headers,
+        )
+    ).json()["id"]
+
+    artifact = _artifact(tmp_path, build_dir="dist")
+
+    async with session_factory() as session:
+        from app.domain.delivery.models import PipelineRun, Release
+        from app.infrastructure.repositories.delivery import PipelineRunRepository
+
+        # A release points at the pipeline run that produced it, so the run has
+        # to exist: the foreign key is real, not decorative.
+        run = await PipelineRunRepository(session).queue(
+            PipelineRun(
+                organization_id=org_id,
+                project_id=project_id,
+                sha="abc1234def",
+                status="succeeded",
+            )
+        )
+        release = await ReleaseRepository(session).create(
+            Release(
+                organization_id=org_id,
+                project_id=project_id,
+                run_id=run.id,
+                sha="abc1234def",
+                artifact_path=artifact,
+                is_demo_live=True,
+            )
+        )
+        await DeployTargetRepository(session).upsert(
+            values={
+                "organization_id": org_id,
+                "project_id": project_id,
+                "kind": "ssh",
+                "host": "deploy.example.com",
+                "port": 22,
+                "username": "deployer",
+                "auth_type": "password",
+                "secret_encrypted": encrypt(SECRET),
+                "remote_path": "/srv/app",
+                "post_deploy_command": "systemctl reload app",
+                "is_active": True,
+            }
+        )
+        await session.commit()
+
+    return {
+        "org_id": org_id,
+        "slug": slug,
+        "project_id": project_id,
+        "headers": headers,
+        "release_id": release.id,
+        "artifact": artifact,
+    }
+
+
+def _service(session, *, ssh: FakeSshTransport, ftp: FakeFtpTransport | None = None) -> tuple[
+    DeployService, FakeDeployTargetTester
+]:
+    ssh_deployer = SshDeployer(
+        transport_factory=lambda: ssh, artifacts_root="/tmp/artifacts", allow_private_hosts=False
+    )
+    ftp_deployer = FtpDeployer(
+        transport_factory=lambda: ftp or FakeFtpTransport(),
+        artifacts_root="/tmp/artifacts",
+        allow_private_hosts=False,
+    )
+    tester = FakeDeployTargetTester(ssh=ssh_deployer, ftp=ftp_deployer)
+    service = DeployService(
+        projects=ProjectRepository(session),
+        organizations=OrganizationRepository(session),
+        targets=DeployTargetRepository(session),
+        deploys=ProductionDeployRepository(session),
+        releases=ReleaseRepository(session),
+        runtimes=None,  # not needed for the deploy paths under test
+        tester=tester,
+        allow_private_hosts=False,
+    )
+    return service, tester
+
+
+# ---------------------------------------------------------------------------
+# the SSH deploy mechanics
+# ---------------------------------------------------------------------------
+
+
+async def test_ssh_deploy_uploads_extracts_and_switches(tmp_path):
+    transport = FakeSshTransport()
+    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)
+
+    ok, release_dir, message = await deployer.deploy(
+        target=DeployTargetSpec(
+            kind="ssh", host="deploy.example.com", port=22, username="deployer",
+            secret="pw", remote_path="/srv/app",
+        ),
+        artifact_path=_artifact(tmp_path, build_dir="dist"),
+        sha="abc1234def",
+        post_deploy_command="systemctl reload app",
+    )
+
+    assert ok is True, message
+    assert release_dir.startswith("/srv/app/releases/")
+    assert transport.uploads, "the artifact must be uploaded"
+    joined = "\n".join(transport.commands)
+    assert "tar -xzf" in joined, "the artifact must be extracted on the target"
+    assert "systemctl reload app" in joined, "the post-deploy command must run"
+    assert "ln -sfn" in joined and CURRENT_LINK in joined, "the symlink must be switched"
+    assert "tail -n" in joined, "old releases must be pruned"
+
+
+async def test_ssh_deploy_prunes_to_five_releases(tmp_path):
+    transport = FakeSshTransport()
+    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)
+    await deployer.deploy(
+        target=DeployTargetSpec(
+            kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app"
+        ),
+        artifact_path=_artifact(tmp_path),
+        sha="abc1234",
+    )
+    prune = next(c for c in transport.commands if "tail -n" in c)
+    assert f"tail -n +{KEEP_RELEASES + 1}" in prune, (
+        "keep the newest five and delete the rest"
+    )
+
+
+async def test_a_failed_post_deploy_command_does_not_switch_the_symlink(tmp_path):
+    """The previous release must keep serving after a failed deploy."""
+    transport = FakeSshTransport(fail_on="systemctl")
+    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)
+
+    ok, _release, message = await deployer.deploy(
+        target=DeployTargetSpec(
+            kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app"
+        ),
+        artifact_path=_artifact(tmp_path),
+        sha="abc1234",
+        post_deploy_command="systemctl reload app",
+    )
+
+    assert ok is False
+    assert "پس از استقرار" in message
+    assert not any("ln -sfn" in c for c in transport.commands), (
+        "the switch must not happen when the hook failed"
+    )
+
+
+async def test_ssh_rollback_points_current_at_the_previous_release(tmp_path):
+    transport = FakeSshTransport()
+    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)
+    # Two releases exist on the target, newest first.
+    transport.files = {
+        "/srv/app/releases/20260101T000000-aaaaaaa/index.html",
+        "/srv/app/releases/20260102T000000-bbbbbbb/index.html",
+    }
+
+    ok, target_name, message = await deployer.rollback(
+        target=DeployTargetSpec(
+            kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app"
+        ),
+        post_deploy_command="systemctl reload app",
+    )
+
+    assert ok is True, message
+    assert target_name == "20260101T000000-aaaaaaa"
+    assert any("ln -sfn" in c for c in transport.commands)
+
+
+async def test_ssh_rollback_refuses_when_there_is_only_one_release(tmp_path):
+    transport = FakeSshTransport()
+    transport.files = {"/srv/app/releases/only-one/index.html"}
+    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)
+
+    ok, _target, message = await deployer.rollback(
+        target=DeployTargetSpec(
+            kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app"
+        )
+    )
+    assert ok is False
+    assert "قبلی" in message
+
+
+# ---------------------------------------------------------------------------
+# host key pinning (TOFU)
+# ---------------------------------------------------------------------------
+
+
+async def test_a_changed_host_key_is_a_hard_failure():
+    transport = FakeSshTransport(fingerprint="SHA256:CHANGED")
+    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)
+
+    check = await deployer.test_connection(
+        target=DeployTargetSpec(
+            kind="ssh", host="h", port=22, username="u", secret="pw",
+            remote_path="/srv/app", known_fingerprint="SHA256:ORIGINAL",
+        )
+    )
+
+    assert check.ok is False
+    assert "کلید میزبان" in check.message
+    assert check.fingerprint == "SHA256:CHANGED"
+
+
+async def test_the_fingerprint_is_returned_for_pinning_on_first_contact():
+    transport = FakeSshTransport(fingerprint="SHA256:FIRST")
+    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)
+
+    check = await deployer.test_connection(
+        target=DeployTargetSpec(
+            kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app"
+        )
+    )
+    assert check.ok is True
+    assert check.fingerprint == "SHA256:FIRST"
+
+
+async def test_a_matching_pinned_key_is_accepted():
+    transport = FakeSshTransport(fingerprint="SHA256:SAME")
+    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)
+    check = await deployer.test_connection(
+        target=DeployTargetSpec(
+            kind="ssh", host="h", port=22, username="u", secret="pw",
+            remote_path="/srv/app", known_fingerprint="SHA256:SAME",
+        )
+    )
+    assert check.ok is True
+
+
+# ---------------------------------------------------------------------------
+# SSRF
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "host",
+    ["127.0.0.1", "169.254.169.254", "10.0.0.5", "localhost"],
+)
+async def test_an_internal_target_is_refused_before_connecting(host: str):
+    transport = FakeSshTransport()
+    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=False)
+
+    check = await deployer.test_connection(
+        target=DeployTargetSpec(
+            kind="ssh", host=host, port=22, username="u", secret="pw", remote_path="/srv/app"
+        )
+    )
+
+    assert check.ok is False
+    assert transport.commands == [], "no connection may be attempted"
+    assert not transport.uploads
+
+
+async def test_an_internal_ftp_target_is_refused():
+    transport = FakeFtpTransport()
+    deployer = FtpDeployer(transport_factory=lambda: transport, allow_private_hosts=False)
+    check = await deployer.test_connection(
+        target=DeployTargetSpec(
+            kind="ftp", host="127.0.0.1", port=21, username="u", secret="pw",
+            remote_path="/var/www",
+        )
+    )
+    assert check.ok is False
+    assert transport.connected is False
+
+
+# ---------------------------------------------------------------------------
+# FTP
+# ---------------------------------------------------------------------------
+
+
+async def test_ftp_deploy_backs_up_before_overwriting(tmp_path):
+    transport = FakeFtpTransport()
+    transport.uploaded = {"index.html": b"OLD", "app.js": b"OLD"}
+    deployer = FtpDeployer(
+        transport_factory=lambda: transport, artifacts_root=str(tmp_path / "artifacts")
+    )
+
+    ok, backup, message = await deployer.deploy(
+        target=DeployTargetSpec(
+            kind="ftp", host="ftp.example.com", port=21, username="u", secret="pw",
+            remote_path="/var/www",
+        ),
+        artifact_path=_artifact(tmp_path, build_dir="dist"),
+        sha="abc1234",
+        organization_id="org-1",
+        project_id="proj-1",
+    )
+
+    assert ok is True, message
+    assert backup and os.path.isfile(backup), "a backup must exist before anything is overwritten"
+    with tarfile.open(backup, "r:gz") as tar:
+        names = set(tar.getnames())
+    assert {"index.html", "app.js"} <= names, "the backup must contain what it overwrites"
+
+
+async def test_ftp_deploy_uploads_only_the_build_output(tmp_path):
+    """A static deploy must not publish the repository, only the build."""
+    transport = FakeFtpTransport()
+    deployer = FtpDeployer(
+        transport_factory=lambda: transport, artifacts_root=str(tmp_path / "artifacts")
+    )
+    await deployer.deploy(
+        target=DeployTargetSpec(
+            kind="ftp", host="h", port=21, username="u", secret="pw", remote_path="/var/www"
+        ),
+        artifact_path=_artifact(tmp_path, build_dir="dist"),
+        sha="abc1234",
+    )
+
+    assert "index.html" in transport.uploaded
+    assert "README.md" not in transport.uploaded, "the source tree must not be published"
+
+
+async def test_ftp_rollback_re_uploads_the_backup(tmp_path):
+    transport = FakeFtpTransport()
+    transport.uploaded = {"index.html": b"OLD"}
+    deployer = FtpDeployer(
+        transport_factory=lambda: transport, artifacts_root=str(tmp_path / "artifacts")
+    )
+    _ok, backup, _message = await deployer.deploy(
+        target=DeployTargetSpec(
+            kind="ftp", host="h", port=21, username="u", secret="pw", remote_path="/var/www"
+        ),
+        artifact_path=_artifact(tmp_path, build_dir="dist"),
+        sha="abc1234",
+        organization_id="org-1",
+        project_id="proj-1",
+    )
+
+    transport.uploaded = {"index.html": b"BROKEN"}
+    ok, message = await deployer.rollback(
+        target=DeployTargetSpec(
+            kind="ftp", host="h", port=21, username="u", secret="pw", remote_path="/var/www"
+        ),
+        backup_path=backup,
+    )
+
+    assert ok is True, message
+    assert transport.uploaded["index.html"] == b"OLD"
+
+
+async def test_ftp_rollback_without_a_backup_fails_cleanly():
+    transport = FakeFtpTransport()
+    deployer = FtpDeployer(transport_factory=lambda: transport)
+    ok, message = await deployer.rollback(
+        target=DeployTargetSpec(
+            kind="ftp", host="h", port=21, username="u", secret="pw", remote_path="/var/www"
+        ),
+        backup_path="",
+    )
+    assert ok is False
+    assert message
+
+
+# ---------------------------------------------------------------------------
+# the service guards
+# ---------------------------------------------------------------------------
+
+
+async def test_the_confirm_slug_must_match(tenant, session_factory):
+    transport = FakeSshTransport()
+    async with session_factory() as session:
+        service, _tester = _service(session, ssh=transport)
+        with pytest.raises(Exception) as excinfo:
+            await service.deploy_production(
+                organization_id=tenant["org_id"],
+                project_id=tenant["project_id"],
+                actor_user_id="u1",
+                confirm="not-the-slug",
+            )
+        assert "تأییدیه" in str(excinfo.value)
+    assert transport.commands == [], "nothing may be attempted on a bad confirmation"
+
+
+async def test_a_correct_confirmation_deploys_and_records_the_row(tenant, session_factory):
+    transport = FakeSshTransport()
+    async with session_factory() as session:
+        service, _tester = _service(session, ssh=transport)
+        deploy = await service.deploy_production(
+            organization_id=tenant["org_id"],
+            project_id=tenant["project_id"],
+            actor_user_id="u1",
+            confirm=tenant["slug"],
+        )
+        await session.commit()
+
+    assert deploy.status == DeployStatus.SUCCEEDED.value, deploy.message
+    assert deploy.kind == "ssh"
+    async with session_factory() as session:
+        rows = await ProductionDeployRepository(session).list_for_project(
+            project_id=tenant["project_id"]
+        )
+        assert len(rows) == 1
+        assert rows[0].release_id == tenant["release_id"]
+
+
+async def test_a_second_deploy_is_refused_while_one_is_running(tenant, session_factory):
+    from app.domain.delivery.models import ProductionDeploy
+
+    transport = FakeSshTransport()
+    async with session_factory() as session:
+        repo = ProductionDeployRepository(session)
+        await repo.create(
+            ProductionDeploy(
+                organization_id=tenant["org_id"],
+                project_id=tenant["project_id"],
+                kind="ssh",
+                host="h",
+                status=DeployStatus.RUNNING.value,
+            )
+        )
+        await session.commit()
+
+        service, _tester = _service(session, ssh=transport)
+        with pytest.raises(Exception) as excinfo:
+            await service.deploy_production(
+                organization_id=tenant["org_id"],
+                project_id=tenant["project_id"],
+                actor_user_id="u1",
+                confirm=tenant["slug"],
+            )
+        assert "در حال انجام" in str(excinfo.value)
+
+
+async def test_a_secret_never_reaches_the_log_or_the_row(tenant, session_factory):
+    """The one thing that must never happen: a credential in a stored field."""
+    transport = FakeSshTransport()
+    async with session_factory() as session:
+        service, _tester = _service(session, ssh=transport)
+        deploy = await service.deploy_production(
+            organization_id=tenant["org_id"],
+            project_id=tenant["project_id"],
+            actor_user_id="u1",
+            confirm=tenant["slug"],
+        )
+        await session.commit()
+
+    assert SECRET not in deploy.log_tail
+    assert SECRET not in deploy.message
+    for _local, remote in transport.uploads:
+        assert SECRET not in remote
+    for command in transport.commands:
+        assert SECRET not in command, "a credential must never reach a remote command line"
+
+
+async def test_the_ftp_rollback_without_a_backup_is_refused_not_guessed(
+    tenant, session_factory
+):
+    """FTP has no atomic switch: without the backup there is nothing to undo."""
+    async with session_factory() as session:
+        target = await DeployTargetRepository(session).get_for_project(tenant["project_id"])
+        assert target is not None
+        target.kind = "ftp"
+        await session.commit()
+
+    transport = FakeFtpTransport()
+    async with session_factory() as session:
+        service, _tester = _service(session, ssh=FakeSshTransport(), ftp=transport)
+        deploy = await service.deploy_production(
+            organization_id=tenant["org_id"],
+            project_id=tenant["project_id"],
+            actor_user_id="u1",
+            confirm=tenant["slug"],
+        )
+        await session.commit()
+        assert deploy.status == DeployStatus.SUCCEEDED.value, deploy.message
+
+        # Clear the recorded backup to simulate a lost archive.
+        row = await ProductionDeployRepository(session).get(deploy.id)
+        assert row is not None
+        deploy_row = await session.get(
+            __import__(
+                "app.infrastructure.db.models_delivery", fromlist=["ProductionDeployRow"]
+            ).ProductionDeployRow,
+            deploy.id,
+        )
+        assert deploy_row is not None
+        deploy_row.backup_path = ""
+        await session.commit()
+
+        rolled = await service.rollback_production(
+            organization_id=tenant["org_id"],
+            project_id=tenant["project_id"],
+            actor_user_id="u1",
+            confirm=tenant["slug"],
+            deploy_id=deploy.id,
+        )
+        await session.commit()
+
+    assert rolled.status == DeployStatus.FAILED.value
+    assert "پشتیبان" in rolled.message
+
+
+# ---------------------------------------------------------------------------
+# HTTP surface
+# ---------------------------------------------------------------------------
+
+
+async def test_a_member_cannot_deploy_to_production(client: AsyncClient, tenant):
+    member_token = (await signup(client, "deploy-member@example.com"))["access_token"]
+    member_headers = auth_header(member_token)
+    # Join the org as a viewer so membership succeeds but the capability does not.
+    invite = await client.post(
+        f"/api/organizations/{tenant['org_id']}/invites",
+        json={"email": "deploy-member@example.com", "role": "viewer"},
+        headers=tenant["headers"],
+    )
+    if invite.status_code in (200, 201):
+        token = invite.json().get("token") or invite.json().get("invite_url", "").split("=")[-1]
+        if token:
+            await client.post(
+                "/api/invites/accept",
+                json={"token": token, "password": "S3cure-passw0rd!"},
+                headers=member_headers,
+            )
+
+    response = await client.post(
+        f"/api/organizations/{tenant['org_id']}/delivery/production/deploy",
+        json={
+            "project_id": tenant["project_id"],
+            "confirm": tenant["slug"],
+        },
+        headers=member_headers,
+    )
+    # Either not a member (404, the tenancy check) or a member without the
+    # capability (403). Never a success.
+    assert response.status_code in (403, 404), response.text
+
+
+async def test_the_confirm_slug_is_enforced_over_http(client: AsyncClient, tenant):
+    response = await client.post(
+        f"/api/organizations/{tenant['org_id']}/delivery/production/deploy",
+        json={"project_id": tenant["project_id"], "confirm": "wrong-slug"},
+        headers=tenant["headers"],
+    )
+    assert response.status_code == 422, response.text
+    assert "تأییدیه" in response.json()["error"]["message"]
+
+
+async def test_the_target_summary_never_returns_the_secret(client: AsyncClient, tenant):
+    response = await client.get(
+        f"/api/organizations/{tenant['org_id']}/delivery/production",
+        params={"project_id": tenant["project_id"]},
+        headers=tenant["headers"],
+    )
+
+    assert response.status_code == 200, response.text
+    body = response.json()
+    assert body["target"]["has_secret"] is True
+    assert SECRET not in response.text
+    assert "secret" not in set(body["target"]), "the raw secret column must not be exposed"
+
+
+async def test_deploying_without_a_target_is_refused(client: AsyncClient, tenant, session_factory):
+    async with session_factory() as session:
+        target = await DeployTargetRepository(session).get_for_project(tenant["project_id"])
+        assert target is not None
+        target.is_active = False
+        await session.commit()
+
+    response = await client.post(
+        f"/api/organizations/{tenant['org_id']}/delivery/production/deploy",
+        json={"project_id": tenant["project_id"], "confirm": tenant["slug"]},
+        headers=tenant["headers"],
+    )
+    assert response.status_code == 422, response.text
+    assert "مقصد" in response.json()["error"]["message"]

docs/plan/reports/agent-3-phase-5.md

@@ -0,0 +1,146 @@
+# Agent 3 — Phase 5 report: production deploy (SSH / SFTP / FTP / FTPS)
+
+Date: 2026-10-14 · Branch: `dev`
+
+## What was done
+
+### 5.1 `adapters/deploy_ssh.py` — `SshDeployer`
+
+SSH and SFTP share one implementation (both are asyncssh; the difference is only which
+subsystem is used).
+
+* **`test_connection`** — SSRF check first, then connect (15 s), capture the host key
+  fingerprint, and probe that `remote_path` exists and is writable.
+* **Host key TOFU** — the first successful test records the fingerprint; every later
+  connection must match it. A mismatch is a hard failure ("کلید میزبان تغییر کرده است"),
+  because it is either a rebuilt server or an interception and only a human can tell.
+* **`deploy`** — upload the artifact to `remote_path/releases/<timestamp>-<sha7>/`,
+  extract, run `post_deploy_command` with `cwd` there (600 s, streamed), then switch the
+  `current` symlink atomically (`ln -sfn` to a relative target + `mv -T` on the same
+  filesystem), then prune to the newest 5.
+* **`rollback`** — re-point `current` at the previous release (`ls -1t` is newest-first,
+  so it is the second entry) and re-run the hook. A rollback that rebuilt would not be a
+  rollback.
+
+**A failed deploy never moves `current`.** If the extraction or the post-deploy command
+fails, the function returns before the switch, so the previous release keeps serving —
+the same property the demo pipeline has, for the same reason.
+
+The remote scripts are built from constants, never from user input, and every argument is
+`shlex.quote`d.
+
+### 5.2 `adapters/deploy_ftp.py` — `FtpDeployer`
+
+* **`test_connection`** — SSRF check, login (explicit TLS for `ftps`), `cwd` the target
+  path, write and delete a probe file.
+* **`deploy`** — **backs up first, always**: every file about to be overwritten is
+  downloaded into `ARTIFACTS_ROOT/<org>/<project>/ftp-backup-<ts>-<sha7>.tar.gz` before
+  anything is uploaded. The artifact is extracted to a scratch directory first, then the
+  deployable build is chosen (`dist/`, `build/`, `public/`, `out/`, else the root) — a
+  static deploy must not publish the repository.
+* **`rollback`** — re-upload that archive.
+
+FTP's limits are documented rather than hidden: no atomic switch, no host key to pin, and
+without the backup there is nothing to undo. `rollback_production` refuses with a plain
+message in that case rather than guessing.
+
+**Archive safety.** `_safe_member` sanitises every tar member before extraction: absolute
+names and `..` are refused, as are symlink/hardlink members whose target escapes. An
+uploaded artifact is not trusted, and `filter="data"` alone is not enough (it does not
+skip the archive's own root entry, which is the member that broke extraction here).
+
+### 5.3 `adapters/deploy_target.py` — `DispatchingDeployTargetTester`
+
+Routes `test_connection` on `spec.kind` and exposes `deployer_for(kind)`. The wizard sees
+one shape; the differences between SSH and FTP stay inside the deployers.
+
+### 5.4 `services/deploy_service.py` + endpoints
+
+`production_deploys` table (migration `a3b4c5d6e7f4`, chained off this workstream's own
+head so the three branches still merge cleanly).
+
+| Method | Path | Role |
+|---|---|---|
+| `GET` | `/delivery/production` | admin+ — target summary + deploy history |
+| `POST` | `/delivery/production/test` | admin+ — test the target, record the result |
+| `POST` | `/delivery/production/deploy` | org owner/admin, slug confirmed, audited |
+| `POST` | `/delivery/production/rollback` | org owner/admin, slug confirmed, audited |
+
+**One production deploy per project at a time.** Two concurrent deploys to the same target
+interleave their writes on a shared host — over SSH merely confusing, over FTP how a site
+ends up in a state nobody can describe.
+
+**There is no code path from a pipeline run into this service.** The engine's furthest
+reach is "merged and pushed to `dev`"; production needs a human, a typed confirmation and
+an audit row. That is D13, and it is enforced by the module boundary, not by a flag.
+
+The target summary returns `has_secret: true`, never the secret. `_spec_from_target` is the
+only place a secret is decrypted, and the plaintext lives for the duration of one call.
+
+## Files changed
+
+| File | Change |
+|---|---|
+| `app/infrastructure/adapters/deploy_ssh.py` | new — `SshDeployer`, `AsyncSshTransport` |
+| `app/infrastructure/adapters/deploy_ftp.py` | new — `FtpDeployer`, `AsyncFtpTransport`, safe extraction |
+| `app/infrastructure/adapters/deploy_target.py` | new — the dispatching tester |
+| `app/application/services/deploy_service.py` | new — the deploy use cases |
+| `app/domain/delivery/models.py` | `DeployStatus`, `ProductionDeploy` |
+| `app/infrastructure/db/models_delivery.py` | `ProductionDeployRow` |
+| `alembic/versions/a3b4c5d6e7f4_production_deploys.py` | new migration |
+| `app/infrastructure/repositories/delivery.py` | `ProductionDeployRepository` |
+| `app/infrastructure/repositories/projects.py` | `DeployTargetRepository` |
+| `app/application/container_delivery.py`, `container.py`, `deps.py` | wiring |
+| `app/interfaces/http/routers/delivery.py` | 4 production routes |
+| `tests/test_deploy.py` | new — 26 tests |
+
+## Test output
+
+```
+apps/api:  499 passed, 1 skipped (was 473; +26)
+           ruff check .  -> All checks passed!
+           mypy app      -> Success: no issues found in 136 source files
+alembic:   upgrade head / check -> clean
+```
+
+## Deviations from the TODO
+
+1. **`post_deploy_command` is a deploy parameter, not part of `DeployTargetSpec`.**
+   `ports/deploy_target.py` is a frozen contract and the spec has no such field; the
+   service that owns the `deploy_targets` row passes it explicitly instead. No contract
+   change.
+2. **`filter="data"` replaced by `_safe_member`.** The stdlib filter does not skip the
+   archive's root member, which made extraction fail; the custom filter is stricter
+   (it also refuses escaping link targets) and is the honest implementation of "an
+   uploaded artifact is not trusted".
+3. **`a3b4c5d6e7f4` is a second migration for this workstream.** The plan allows it
+   ("more only if chained off your own"); chaining off `a3b4c5d6e7f3` keeps the merge with
+   Agent 1 and Agent 2 trivial.
+4. **The `artifact` for FTP is extracted before the build directory is chosen.** The
+   artifact is a tar.gz of the whole release, not a directory; the TODO's "the build dir
+   is set by the framework catalog default or `dist/`, `build/`, `public/`" describes the
+   choice, not the container.
+
+## Open questions / CHANGE REQUESTs
+
+- **CHANGE REQUEST (for integration):** `DeployTargetSpec` would benefit from
+  `post_deploy_command` and `organization_id`/`project_id`. Both are currently passed as
+  extra keyword arguments, which works but means the spec is not the whole story of a
+  deploy. No change is needed for this phase to be correct; noting it so the contract can
+  be widened deliberately at integration rather than by accident.
+- **CHANGE REQUEST (for integration):** `build_deploy_target_tester` now returns the real
+  `DispatchingDeployTargetTester`, so Agent 1's wizard will attempt a real connection.
+  Its tests must inject a fake tester, as `tests/test_deploy.py` does.
+
+## Known limitations
+
+- `AsyncSshTransport` passes `known_hosts=None` and enforces the pinned key by comparing
+  the fingerprint after connecting. That is TOFU, but it means the first byte of the
+  handshake is accepted before the check; a pre-seeded `known_hosts` would be stricter.
+  The fingerprint check happens before any command runs or any file is uploaded.
+- FTPS does not verify the server certificate (`CERT_NONE`): a shared host's FTP
+  certificate is rarely matchable to the domain the operator typed. This is documented in
+  the adapter and stated in the test message the operator sees.
+- No `@pytest.mark.deploy_it` integration test against a real sshd/pure-ftpd yet; the fake
+  transports cover the mechanics, and a live round-trip needs infrastructure the repo
+  cannot assume.