feat(pipeline): уведомления CI/CD — PipelineWatcher, CI загрузок на Gitea Actions

- app/pipeline: протокол CIProvider, клиент Gitea Actions (прогон по head_sha, упавший
  шаг, лог джоба, ретраи), PipelineWatcher — опрос только при незавершённых push,
  сообщение о push обновляется CI ⏳ → ✅/❌ → CD 🚀/⛔, автору — отдельные уведомления
- деплой сервисов — только после зелёного CI; старый push не откатывает сервис,
  который развернёт более новый; после рестарта бот продолжает с того же места
- CI_PROVIDER=none — прежнее поведение (деплой сразу после push)
- шаблон tgit-uploads: .gitea/workflows/check.yml; .gitea/ защищён от правок из бота
- миграция 0002: старые push помечены skipped, чтобы не слать им «таймаут CI»
- ADR-016, ARCHITECTURE §2.5/§5.2, README, DEPLOY, .env.example

Refs #11

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/bot/test_commit_push.py

@@ -53,7 +53,7 @@ async def test_message_then_push_lands_in_remote(tg: TgClient, texts: Texts, rem
     pushing, done = tg.session.sent(EditMessageText)[-2:]
     assert pushing.text == t.pushing
     assert done.text.startswith("✅ Запушено: ") and "docs: заметки" in done.text
-    assert "https://example.test/commit/" in done.text and done.text.endswith(t.ci_pending)
+    assert "https://example.test/commit/" in done.text and "CI" not in done.text  # no CI configured
 
     clone = tmp_path / "verify"
     git(tmp_path, "clone", "-q", str(remote), str(clone))

tests/pipeline/test_gitea.py

@@ -0,0 +1,130 @@
+from typing import Any
+
+import httpx
+import pytest
+import respx
+
+from app.config import CISettings
+from app.db.models import PipelineStatus
+from app.pipeline.gitea import GiteaCI, GiteaError
+from app.pipeline.provider import CIRun
+
+API = "https://gitea.example.test"
+RUNS = f"{API}/api/v1/repos/HSE/up/actions/runs"
+SHA = "eda569427093acb3b39ad5b301216cb5b7400ab8"
+
+
+def run(
+    id: int, path: str = "check.yml@refs/heads/main", status: str = "completed", conclusion: Any = "success"
+) -> Any:
+    """Shape of a real Gitea 1.27 run (fields the client doesn't read trimmed)."""
+    run = {"id": id, "path": path, "status": status, "head_sha": SHA, "html_url": f"{API}/HSE/up/actions/runs/{id}"}
+    return run | ({"conclusion": conclusion} if conclusion else {})  # in-progress runs have no conclusion key
+
+
+def job(id: int, conclusion: str, steps: list[tuple[str, str]]) -> Any:
+    return {
+        "id": id,
+        "name": "check",
+        "status": "completed",
+        "conclusion": conclusion,
+        "steps": [{"name": n, "number": i, "status": "completed", "conclusion": c} for i, (n, c) in enumerate(steps)],
+    }
+
+
+async def no_sleep(_: float) -> None:
+    pass
+
+
+@pytest.fixture
+async def ci() -> Any:
+    c = GiteaCI(CISettings(token="secret", api_url=API, repo="HSE/up", max_attempts=3), sleep=no_sleep)  # type: ignore[arg-type]
+    yield c
+    await c.aclose()
+
+
+@respx.mock
+async def test_finds_newest_run_of_the_workflow(ci: GiteaCI) -> None:
+    other = run(9, path="other.yml@refs/heads/main")
+    route = respx.get(RUNS).respond(
+        json={"workflow_runs": [run(5), other, run(7, status="in_progress", conclusion=None)]}
+    )
+
+    found = await ci.find(SHA)
+
+    assert found == CIRun("7", PipelineStatus.RUNNING, f"{API}/HSE/up/actions/runs/7")
+    assert route.calls.last.request.url.params["head_sha"] == SHA
+    assert route.calls.last.request.headers["Authorization"] == "token secret"
+
+
+@respx.mock
+async def test_no_run_yet(ci: GiteaCI) -> None:
+    respx.get(RUNS).respond(json={"workflow_runs": [], "total_count": 0})
+
+    assert await ci.find(SHA) is None
+
+
+@pytest.mark.parametrize(
+    ("status", "conclusion", "expected"),
+    [
+        ("queued", None, PipelineStatus.PENDING),
+        ("waiting", None, PipelineStatus.PENDING),
+        ("in_progress", None, PipelineStatus.RUNNING),
+        ("completed", "success", PipelineStatus.SUCCESS),
+        ("completed", "skipped", PipelineStatus.SKIPPED),
+    ],
+)
+@respx.mock
+async def test_status_mapping(ci: GiteaCI, status: str, conclusion: str | None, expected: PipelineStatus) -> None:
+    respx.get(RUNS).respond(json={"workflow_runs": [run(1, status=status, conclusion=conclusion)]})
+
+    found = await ci.find(SHA)
+
+    assert found is not None and found.status is expected
+
+
+@respx.mock
+async def test_failure_names_the_step_and_reads_its_log(ci: GiteaCI) -> None:
+    respx.get(RUNS).respond(json={"workflow_runs": [run(3, conclusion="failure")]})
+    steps = [("Run actions/checkout@v4", "success"), ("lint", "failure"), ("tests", "skipped")]
+    respx.get(f"{RUNS}/3/jobs").respond(json={"jobs": [job(31, "failure", steps)], "total_count": 1})
+    log = "2026-10-18T20:44:56.7721431Z ruff check .\n2026-10-18T20:44:57.1Z F821 undefined name 'x'\n"
+    respx.get(f"{API}/api/v1/repos/HSE/up/actions/jobs/31/logs").respond(text=log)
+
+    found = await ci.find(SHA)
+    assert found is not None
+    text = await ci.failure_log(found)
+
+    assert (found.status, found.failed_step) == (PipelineStatus.FAILED, "check / lint")
+    assert text == "ruff check .\nF821 undefined name 'x'\n"
+
+
+@respx.mock
+async def test_cancelled_run_is_a_failure(ci: GiteaCI) -> None:
+    respx.get(RUNS).respond(json={"workflow_runs": [run(4, conclusion="cancelled")]})
+    respx.get(f"{RUNS}/4/jobs").respond(json={"jobs": [job(41, "cancelled", [("lint", "success")])]})
+
+    found = await ci.find(SHA)
+
+    assert found is not None and (found.status, found.failed_step) == (PipelineStatus.FAILED, "check / cancelled")
+
+
+@respx.mock
+async def test_retries_transport_errors_and_5xx(ci: GiteaCI) -> None:
+    route = respx.get(RUNS)
+    route.side_effect = [
+        httpx.ConnectError("reset"),
+        httpx.Response(502),
+        httpx.Response(200, json={"workflow_runs": []}),
+    ]
+
+    assert await ci.find(SHA) is None
+    assert route.call_count == 3
+
+
+@respx.mock
+async def test_gives_up_with_a_typed_error(ci: GiteaCI) -> None:
+    respx.get(RUNS).respond(404, json={"message": "not found"})
+
+    with pytest.raises(GiteaError, match="404"):
+        await ci.find(SHA)

tests/pipeline/test_watcher.py

@@ -0,0 +1,337 @@
+"""PipelineWatcher with a scripted CI, a fake deployer, an offline Telegram and a hand-turned clock."""
+
+import asyncio
+from collections.abc import Callable
+from datetime import datetime, timedelta
+from typing import Any
+
+import pytest
+from aiogram import Bot
+from aiogram.methods import EditMessageText, SendDocument, SendMessage
+from sqlalchemy import select
+
+from app.access.service import AccessService
+from app.db.models import PipelineRun, PipelineStatus, Push, Stage, utcnow
+from app.db.session import SessionMaker
+from app.pipeline.gitea import GiteaError
+from app.pipeline.provider import CIRun
+from app.pipeline.watcher import PipelineWatcher, initial_status
+from app.runner.client import RunnerResult
+from app.runner.deployer import Deployer, Outcome, Report
+from app.texts import Texts
+from tests.conftest import ADMIN
+from tests.fakes import BOT_TOKEN, FakeSession
+
+S = PipelineStatus
+URL = "https://ci.example/runs/1"
+
+
+class FakeCI:
+    def __init__(self) -> None:
+        self.runs: dict[str, CIRun | None] = {}
+        self.log = "ruff check .\nF821 undefined name 'x'"
+        self.errors = 0  # the next N calls fail like a flaky API
+
+    def set(self, sha: str, status: PipelineStatus, step: str | None = None) -> None:
+        self.runs[sha] = CIRun("1", status, URL, step)
+
+    async def find(self, sha: str) -> CIRun | None:
+        if self.errors:
+            self.errors -= 1
+            raise GiteaError("HTTP 502")
+        return self.runs.get(sha)
+
+    async def failure_log(self, run: CIRun) -> str:
+        return self.log
+
+    async def aclose(self) -> None:
+        pass
+
+
+class FakeDeployer(Deployer):
+    def __init__(self, fail: frozenset[str] = frozenset()) -> None:  # no repo, no runner
+        self.calls: list[tuple[str, list[str]]] = []
+        self.fail = fail
+
+    async def deploy(self, sha: str, services: list[str], report: Report) -> None:
+        self.calls.append((sha, services))
+        for svc in services:
+            ok = svc not in self.fail
+            await report(Outcome(svc, RunnerResult(ok, 0 if ok else 3, "ok" if ok else "not healthy", "boom\nFAIL")))
+
+
+class Clock:
+    def __init__(self) -> None:
+        self.now = utcnow()
+
+    def __call__(self) -> datetime:
+        return self.now
+
+
+@pytest.fixture
+def tg() -> tuple[Bot, FakeSession]:
+    session = FakeSession()
+    return Bot(BOT_TOKEN, session=session), session
+
+
+@pytest.fixture
+def ci() -> FakeCI:
+    return FakeCI()
+
+
+@pytest.fixture
+def clock() -> Clock:
+    return Clock()
+
+
+Make = Callable[..., PipelineWatcher]
+
+
+@pytest.fixture
+def make(sessions: SessionMaker, texts: Texts, tg: tuple[Bot, FakeSession], ci: FakeCI, clock: Clock) -> Make:
+    def build(deployer: Deployer | None = None, with_ci: bool = True) -> PipelineWatcher:
+        return PipelineWatcher(
+            sessions, tg[0], texts, ci if with_ci else None, deployer, poll_interval=0.01, clock=clock
+        )
+
+    return build
+
+
+async def add_push(
+    sessions: SessionMaker, sha: str, paths: list[str], ci: PipelineStatus = S.PENDING, cd: PipelineStatus = S.PENDING
+) -> Push:
+    push = Push(
+        user_id=ADMIN, sha=sha, message="feat: x", paths=paths, chat_id=ADMIN, message_id=7, ci_status=ci, cd_status=cd
+    )
+    async with sessions.begin() as s:
+        s.add(push)
+    return push
+
+
+async def state(sessions: SessionMaker, sha: str) -> tuple[PipelineStatus, PipelineStatus]:
+    async with sessions() as s:
+        push = await s.scalar(select(Push).where(Push.sha == sha))
+        assert push is not None
+        return push.ci_status, push.cd_status
+
+
+async def settle(watcher: PipelineWatcher, sessions: SessionMaker, sha: str) -> tuple[PipelineStatus, PipelineStatus]:
+    """CD runs in a background task. Wait for it, not for the DB: in tests SQLite is one shared connection, and a
+    read racing the watcher's write could roll that write back."""
+    async with asyncio.timeout(5):
+        await watcher.drain()
+    return await state(sessions, sha)
+
+
+def texts_of(session: FakeSession, method: Any) -> list[str]:
+    return [m.text if hasattr(m, "text") else m.caption for m in session.sent(method)]
+
+
+SHA_A = "a" * 40
+SHA_B = "b" * 40
+HELLO = ["services/hello/main.py"]
+
+
+def test_initial_status() -> None:
+    assert initial_status(True, True, HELLO) == (S.PENDING, S.PENDING)
+    assert initial_status(True, True, ["README.md"]) == (S.PENDING, S.SKIPPED)  # no service touched
+    assert initial_status(False, True, HELLO) == (S.SKIPPED, S.PENDING)
+    assert initial_status(True, False, HELLO) == (S.PENDING, S.SKIPPED)
+
+
+async def test_ci_green_then_deploy(
+    access: AccessService, sessions: SessionMaker, make: Make, ci: FakeCI, tg: tuple[Bot, FakeSession], texts: Texts
+) -> None:
+    t, session = texts.pipeline, tg[1]
+    deployer = FakeDeployer()
+    watcher = make(deployer)
+    await add_push(sessions, SHA_A, HELLO)
+
+    assert await watcher.tick() is True  # no run yet: nothing to say, keep polling
+    assert session.requests == []
+
+    ci.set(SHA_A, S.RUNNING)
+    await watcher.tick()
+    assert texts_of(session, EditMessageText)[-1].endswith(
+        f"{t.ci_running}\n{URL}\n{t.cd_waiting.format(services='hello')}"
+    )
+    await watcher.tick()  # same status — no edit
+    assert len(session.sent(EditMessageText)) == 1
+
+    ci.set(SHA_A, S.SUCCESS)
+    assert await watcher.tick() is False
+    assert await settle(watcher, sessions, SHA_A) == (S.SUCCESS, S.SUCCESS)
+
+    assert deployer.calls == [(SHA_A, ["hello"])]
+    edits = texts_of(session, EditMessageText)
+    assert edits[-1].endswith(f"{t.ci_success}\n{URL}\n{t.cd_success.format(services='hello')}")
+    assert any(t.cd_running.format(services="hello") in e for e in edits)
+    sent = texts_of(session, SendMessage)
+    assert sent == [
+        t.notify_ci_ok.format(sha="aaaaaaa", url=URL),
+        t.notify_cd_ok.format(services="hello", sha="aaaaaaa"),
+    ]
+    assert all(m.reply_parameters.message_id == 7 for m in session.sent(SendMessage))
+    assert await watcher.tick() is False  # nothing left: the loop would sleep until wake()
+
+
+async def test_ci_red_reports_step_and_log_and_skips_deploy(
+    access: AccessService, sessions: SessionMaker, make: Make, ci: FakeCI, tg: tuple[Bot, FakeSession], texts: Texts
+) -> None:
+    t, session = texts.pipeline, tg[1]
+    deployer = FakeDeployer()
+    watcher = make(deployer)
+    await add_push(sessions, SHA_A, HELLO)
+    ci.set(SHA_A, S.FAILED, "check / lint")
+
+    await watcher.tick()
+
+    assert await state(sessions, SHA_A) == (S.FAILED, S.SKIPPED)
+    assert deployer.calls == []
+    assert texts_of(session, EditMessageText)[-1].endswith(
+        f"{t.ci_failed.format(step='check / lint')}\n{URL}\n{t.cd_not_run}"
+    )
+    [note] = texts_of(session, SendMessage)
+    assert note.startswith("❌ CI упал на шаге <b>check / lint</b> · aaaaaaa")
+    assert "<pre>ruff check .\nF821 undefined name &#x27;x&#x27;</pre>" in note
+    assert session.sent(SendDocument) == []  # short log: inline only
+    async with sessions() as s:
+        row = await s.scalar(select(PipelineRun).where(PipelineRun.stage == Stage.CI))
+        assert row is not None and (row.run_slug, row.status, row.failed_cube, row.url) == (
+            "1",
+            S.FAILED,
+            "check / lint",
+            URL,
+        )
+
+
+async def test_long_log_also_goes_as_a_file(
+    access: AccessService, sessions: SessionMaker, make: Make, ci: FakeCI, tg: tuple[Bot, FakeSession]
+) -> None:
+    ci.log = "\n".join(f"line {i}" for i in range(2000))
+    await add_push(sessions, SHA_A, HELLO)
+    ci.set(SHA_A, S.FAILED, "check / tests")
+
+    await make(FakeDeployer()).tick()
+
+    [note] = texts_of(tg[1], SendMessage)
+    assert "line 1999" in note and "line 0\n" not in note and len(note) < 4096
+    [doc] = tg[1].sent(SendDocument)
+    assert doc.document.filename == "ci-aaaaaaa.log"
+
+
+async def test_no_run_for_too_long_times_out(
+    access: AccessService, sessions: SessionMaker, make: Make, clock: Clock, tg: tuple[Bot, FakeSession], texts: Texts
+) -> None:
+    deployer = FakeDeployer()
+    watcher = make(deployer)
+    await add_push(sessions, SHA_A, HELLO)
+    clock.now += timedelta(minutes=29)
+    await watcher.tick()
+    assert await state(sessions, SHA_A) == (S.PENDING, S.PENDING)
+
+    clock.now += timedelta(minutes=2)
+    await watcher.tick()
+
+    assert await state(sessions, SHA_A) == (S.TIMEOUT, S.SKIPPED)
+    assert deployer.calls == []
+    assert texts_of(tg[1], SendMessage) == [texts.pipeline.notify_ci_timeout.format(sha="aaaaaaa", minutes=30)]
+
+
+async def test_failed_deploy(
+    access: AccessService, sessions: SessionMaker, make: Make, ci: FakeCI, tg: tuple[Bot, FakeSession], texts: Texts
+) -> None:
+    t = texts.pipeline
+    paths = ["services/hello/main.py", "services/weather/main.py"]
+    await add_push(sessions, SHA_A, paths)
+    ci.set(SHA_A, S.SUCCESS)
+
+    watcher = make(FakeDeployer(fail=frozenset({"weather"})))
+    await watcher.tick()
+
+    assert await settle(watcher, sessions, SHA_A) == (S.SUCCESS, S.FAILED)
+    assert texts_of(tg[1], EditMessageText)[-1].endswith(t.cd_failed.format(services="weather"))
+    sent = texts_of(tg[1], SendMessage)
+    assert sent[1].startswith("⛔ <b>weather</b> не развёрнут: not healthy") and "<pre>boom" in sent[1]
+    assert sent[2] == t.notify_cd_ok.format(services="hello", sha="aaaaaaa")
+
+
+async def test_newer_push_wins_when_ci_finishes_out_of_order(
+    access: AccessService, sessions: SessionMaker, make: Make, ci: FakeCI, tg: tuple[Bot, FakeSession], texts: Texts
+) -> None:
+    deployer = FakeDeployer()
+    watcher = make(deployer)
+    await add_push(sessions, SHA_A, HELLO)
+    await add_push(sessions, SHA_B, [*HELLO, "services/weather/main.py"])
+    ci.set(SHA_B, S.SUCCESS)
+    await watcher.tick()
+    await settle(watcher, sessions, SHA_B)
+
+    ci.set(SHA_A, S.SUCCESS)  # the older push turns green later: deploying it would roll hello back
+    await watcher.tick()
+
+    assert await settle(watcher, sessions, SHA_A) == (S.SUCCESS, S.SKIPPED)
+    assert deployer.calls == [(SHA_B, ["hello", "weather"])]
+    assert any(texts.pipeline.cd_superseded in e for e in texts_of(tg[1], EditMessageText))
+
+
+async def test_restart_picks_up_an_interrupted_deploy(
+    access: AccessService, sessions: SessionMaker, make: Make
+) -> None:
+    await add_push(sessions, SHA_A, HELLO, ci=S.SUCCESS, cd=S.RUNNING)  # the bot died mid-deploy
+    deployer = FakeDeployer()
+    watcher = make(deployer)
+
+    await watcher.tick()
+
+    assert await settle(watcher, sessions, SHA_A) == (S.SUCCESS, S.SUCCESS)
+    assert deployer.calls == [(SHA_A, ["hello"])]
+
+
+async def test_without_ci_deploys_right_away(access: AccessService, sessions: SessionMaker, make: Make) -> None:
+    await add_push(sessions, SHA_A, HELLO, ci=S.SKIPPED)
+    deployer = FakeDeployer()
+
+    watcher = make(deployer, with_ci=False)
+
+    assert await watcher.tick() is False
+    assert await settle(watcher, sessions, SHA_A) == (S.SKIPPED, S.SUCCESS)
+
+
+async def test_ci_switched_off_does_not_leave_pushes_waiting(
+    access: AccessService, sessions: SessionMaker, make: Make
+) -> None:
+    await add_push(sessions, SHA_A, ["README.md"], cd=S.SKIPPED)
+
+    await make(FakeDeployer(), with_ci=False).tick()
+
+    assert await state(sessions, SHA_A) == (S.SKIPPED, S.SKIPPED)
+
+
+async def test_loop_survives_api_errors_and_wakes_on_push(
+    access: AccessService,
+    sessions: SessionMaker,
+    make: Make,
+    ci: FakeCI,
+    tg: tuple[Bot, FakeSession],
+    texts: Texts,
+) -> None:
+    watcher = make(FakeDeployer())
+    done = texts.pipeline.notify_cd_ok.format(services="hello", sha="aaaaaaa")
+    await add_push(sessions, SHA_B, ["README.md"], ci=S.SKIPPED, cd=S.SKIPPED)  # finished: nothing to watch
+    loop = asyncio.create_task(watcher.run())
+    try:
+        await asyncio.sleep(0.05)
+        assert tg[1].requests == []  # idle until woken
+        ci.errors = 2
+        ci.set(SHA_A, S.SUCCESS)
+        await add_push(sessions, SHA_A, HELLO)
+        watcher.wake()
+        async with asyncio.timeout(5):
+            while done not in texts_of(tg[1], SendMessage):  # noqa: ASYNC110 - fake Telegram, no event
+                await asyncio.sleep(0.01)
+            await watcher.drain()
+        assert ci.errors == 0
+    finally:
+        loop.cancel()
+    assert await state(sessions, SHA_A) == (S.SUCCESS, S.SUCCESS)

tests/repo/test_publisher.py

@@ -5,7 +5,7 @@ import pytest
 from sqlalchemy import select
 
 from app.access.service import AccessService
-from app.db.models import ChangeOp, Push, User, utcnow
+from app.db.models import ChangeOp, PipelineStatus, Push, User, utcnow
 from app.db.session import SessionMaker
 from app.repo.drafts import Change, DraftView
 from app.repo.git import UploadsRepo
@@ -129,11 +129,13 @@ async def test_rate_limit_and_record(repo: UploadsRepo, sessions: SessionMaker,
     publisher = Publisher(repo, sessions)
     draft = await draft_on(repo, Change("x.txt", ChangeOp.ADD, b"x"))
     result = await publisher.push(member, draft, "x")
-    await publisher.record(member, result, "x", draft, chat_id=ADMIN, message_id=77)
+    status = (PipelineStatus.PENDING, PipelineStatus.SKIPPED)
+    await publisher.record(member, result, "x", draft, chat_id=ADMIN, message_id=77, status=status)
 
     async with sessions() as s:
         push = await s.scalar(select(Push))
         assert push is not None and push.sha == result.sha and push.paths == ["x.txt"] and push.message_id == 77
+        assert (push.ci_status, push.cd_status) == status
         for i in range(PUSHES_PER_HOUR):
             s.add(
                 Push(user_id=ADMIN, sha=f"{i:040d}", message="m", chat_id=1, created_at=utcnow() - timedelta(minutes=1))

tests/repo/test_validate.py

@@ -27,6 +27,7 @@ def test_bad_names(name: str) -> None:
     ("path", "protected"),
     [
         (".sourcecraft/ci.yaml", True),
+        (".gitea/workflows/check.yml", True),
         (".tgit/x.py", True),
         ("ruff.toml", True),
         (".sourcecraft", True),

tests/runner/test_deploy.py

@@ -4,7 +4,7 @@ import asyncio
 import base64
 import json
 import subprocess
-from collections.abc import Callable
+from collections.abc import AsyncIterator, Callable
 from dataclasses import replace
 from datetime import UTC, datetime
 from pathlib import Path
@@ -24,6 +24,7 @@ from app.config import RunnerSettings
 from app.db.models import Alert, AlertKind
 from app.db.session import SessionMaker
 from app.main import build_dispatcher
+from app.pipeline.watcher import PipelineWatcher
 from app.repo.git import UploadsRepo
 from app.runner.client import RunnerClient, RunnerResult
 from app.runner.deployer import Deployer, Outcome, changed_services
@@ -168,14 +169,23 @@ async def test_one_broken_service_does_not_stop_others(repo: UploadsRepo) -> Non
 
 
 @pytest.fixture
-def tg(texts: Texts, sessions: SessionMaker, access: AccessService, repo: UploadsRepo, fake_host: Path) -> TgClient:
+async def tg(
+    texts: Texts, sessions: SessionMaker, access: AccessService, repo: UploadsRepo, fake_host: Path
+) -> AsyncIterator[TgClient]:
+    loops: list[asyncio.Task[None]] = []
+
     def dispatcher(bot: Bot) -> Dispatcher:
         alerts = AlertService(
             sessions, TelegramNotifier(bot, access, sessions, texts), load_alerts_config(ROOT / "config")
         )
-        return build_dispatcher(texts, sessions, access, repo, alerts=alerts, runner=client())
+        # no CI configured: the watcher deploys right after push
+        watcher = PipelineWatcher(sessions, bot, texts, None, Deployer(repo, client(), alerts), poll_interval=0.05)
+        loops.append(asyncio.create_task(watcher.run()))
+        return build_dispatcher(texts, sessions, access, repo, alerts=alerts, runner=client(), pipeline=watcher)
 
-    return TgClient(dispatcher)
+    yield TgClient(dispatcher)
+    for loop in loops:
+        loop.cancel()
 
 
 async def test_push_deploys_touched_service(tg: TgClient, texts: Texts, fake_host: Path, remote: Path) -> None:
@@ -189,12 +199,15 @@ async def test_push_deploys_touched_service(tg: TgClient, texts: Texts, fake_hos
     await tg.send(ADMIN, "add util")
     await tg.press(ADMIN, CommitCb(a=Act.PUSH).pack())
     await tg.press(ADMIN, CommitCb(a=Act.PUSH_GO).pack())
-    await until(lambda: any("работает" in (m.text or "") for m in tg.session.sent(SendMessage)))
+    await until(lambda: any("развёрнуты" in (m.text or "") for m in tg.session.sent(SendMessage)))
 
-    pushed = [e for e in tg.session.sent(EditMessageText) if "Запушено" in (e.text or "")]
-    assert pushed and "🚀 Разворачиваю: hello…" in pushed[-1].text
-    [done] = [m for m in tg.session.sent(SendMessage) if "работает" in (m.text or "")]
-    assert done.text.startswith("🚀 <b>hello</b> работает · ")
+    pushed = [e.text for e in tg.session.sent(EditMessageText) if "Запушено" in (e.text or "")]
+    assert "CI" not in pushed[0] and pushed[0].endswith(texts.pipeline.cd_running.format(services="hello"))
+    await until(
+        lambda: tg.session.sent(EditMessageText)[-1].text.endswith(texts.pipeline.cd_success.format(services="hello"))
+    )
+    [done] = [m for m in tg.session.sent(SendMessage) if "развёрнуты" in (m.text or "")]
+    assert done.text.startswith("🚀 CD: сервисы hello развёрнуты, ревизия ")
     assert "tgit-svc-hello" in docker_state(fake_host)["containers"]  # type: ignore[operator]