diff --git a/.env.example b/.env.example index 0204d02..8884702 100644 --- a/.env.example +++ b/.env.example @@ -24,6 +24,14 @@ GIT_SSH_KEY_B64= # (secret) для Docker: тот же ключ GIT_UPLOADS_URL=git@ssh.sourcecraft.dev:rogachevegorushka/tgit-uploads.git # GIT_CLONE_DIR=data/uploads # локальный клон репозитория загрузок (в Docker — volume /data/uploads) +# --- CI репозитория загрузок: деплой ждёт зелёный check (#11, ADR-016) --- +CI_PROVIDER=none # none — разворачивать сразу после push | gitea — ждать Gitea Actions +# CI_API_URL=https://gitea.24studbank.ru +CI_TOKEN= # (secret) токен Gitea с чтением репозитория загрузок (его Actions) +# CI_REPO=HSE/tgit-uploads +# CI_WORKFLOW=check.yml +# CI_RUN_TIMEOUT_MINUTES=30 + # --- Docker: раздел «Статус» (#14) --- # В Compose задан сам: http://docker-proxy:2375. Без него (make run) «Статус» отвечает «Docker недоступен». # DOCKER_URL= diff --git a/README.md b/README.md index 96b5ef4..e55c3f2 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ |---|---| | [`docs/SPEC.md`](docs/SPEC.md) | ТЗ: требования A-, M-, C-, S-, L-, AL- | | [`docs/ARCHITECTURE.md`](docs/ARCHITECTURE.md) | модули, потоки данных, хранилище | -| [`docs/DECISIONS.md`](docs/DECISIONS.md) | решения ADR-001…015 с альтернативами | +| [`docs/DECISIONS.md`](docs/DECISIONS.md) | решения ADR-001…016 с альтернативами | | [`docs/dev/DEPLOY.md`](docs/dev/DEPLOY.md) | CI/CD и деплой на сервер | | [`examples/README.md`](examples/README.md) | демо-сервисы и сценарии для проверки | @@ -41,10 +41,11 @@ | **📜 Логи** | хвост 50/200 строк или 5 мин/1 ч/24 ч; свой grep (`-i -v -E -c -C N`), выполняется в Python, не в shell; длинный ответ — файлом; журнал поисков | L-1…L-6 | | **🤖 Агент** | вопрос словами → фильтр → LLM-страж → LLM строит только `LogQuery` → валидация → grep; показывает построенную команду и строки; атака — отказ и 🛡 админам; 30 вопросов в сутки | L-10…L-17 | | **🚨 Алерты** | 💥 краш, 🧠 OOM, 🔁 цикл рестартов, 🩺 unhealthy, 🔥 перегрузка, 💾 мало места, 🔌 Docker недоступен, 🛡 атака; всем участникам, с хвостом лога и кнопками; повторы — счётчик, «✅ восстановлено» ответом; список за 7 дней | AL-1…AL-6 | -| **🚀 Деплой сервисов** | после push бот отправляет изменённые `services//` в `tgit-runner` на хосте: сборка образа, запуск в песочнице (uid 10001, read-only, без capabilities и сети, 128 МБ / 0.25 CPU / 64 процесса); старая версия заменяется только здоровой новой, иначе ⛔ с логом; «♻️ Перезапустить» в «Статусе» для админа | #13, S-5 | +| **⏳ CI/CD после push** | сообщение о push обновляется на месте: CI ⏳ → ✅/❌ → CD 🚀/⛔; CI репозитория загрузок — Gitea Actions (compile · ruff · bandit · pytest), бот находит прогон по sha и деплоит только после зелёного; автору — «✅ CI пройден» со ссылкой, упавший шаг с логом (длинный — файлом), «🚀 CD: сервисы … развёрнуты»; таймаут 30 мин; после рестарта бот продолжает с того же места | C-24, C-30…C-34, #11 | +| **🚀 Деплой сервисов** | после CI бот отправляет изменённые `services//` в `tgit-runner` на хосте: сборка образа, запуск в песочнице (uid 10001, read-only, без capabilities и сети, 128 МБ / 0.25 CPU / 64 процесса); старая версия заменяется только здоровой новой, иначе ⛔ с логом; «♻️ Перезапустить» в «Статусе» для админа | #13, S-5 | | **CI/CD бота** | Gitea Actions: ruff, mypy strict, миграции up/down/up на PostgreSQL, pytest с порогом покрытия 80 %; автодеплой `dev` на сервер с проверкой health | ADR-015 | -Качество: 296 тестов, покрытие 92 %, `mypy --strict`, `ruff`. Telegram в тестах — фейковая сессия aiogram, +Качество: 357 тестов, покрытие 92 %, `mypy --strict`, `ruff`. Telegram в тестах — фейковая сессия aiogram, Docker — фейковый Engine API на `httpx.MockTransport`, LLM — сценарный фейк. ## 3. Развёртывание и запуск @@ -107,6 +108,8 @@ CD из Gitea Actions разворачивает `dev` на сервер скр | `TELEGRAM_PROXY_URL` | нет | HTTP/SOCKS-прокси до Telegram (если сервер не видит api.telegram.org) | | `GIT_UPLOADS_URL` | да | репозиторий загрузок: путь, `ssh://…` или `git@…` | | `GIT_SSH_KEY_B64` / `GIT_SSH_KEY_PATH` | для SSH | deploy-ключ бота | +| `CI_PROVIDER`, `CI_TOKEN`, `CI_REPO` | нет | CI репозитория загрузок, которого ждёт деплой (`gitea`); при `none` сервисы разворачиваются сразу после push | +| `RUNNER_HOST`, `RUNNER_SSH_KEY_B64` | нет | `tgit-runner` на хосте; без него сервисы не разворачиваются | | `DATABASE_URL` | нет | по умолчанию SQLite; в Compose — PostgreSQL | | `DOCKER_URL` | нет | docker-socket-proxy; без него «Статус», «Логи» и алерты отвечают «Docker недоступен» | | `LLM_PROVIDER`, `LLM_BASE_URL`, `LLM_MODEL`, `LLM_API_KEY` | нет | LLM для агента; при `mock` агент выключен | @@ -137,7 +140,8 @@ make cov # pytest с покрытием (порог 80 %) - **В Телеграме**: экраны разделов, файлы (`.log` для длинных результатов, исходник для больших файлов), QR-приглашения PNG, алерты. -- **В репозитории загрузок**: коммиты от `tgit-bot` с автором из Телеграма; push в `main` запускает CI репозитория. +- **В репозитории загрузок**: коммиты от `tgit-bot` с автором из Телеграма; push в `main` запускает CI репозитория + (`.gitea/workflows/check.yml`). - **В БД** (13 таблиц, миграции Alembic): участники и приглашения, черновики, состояние диалогов, снапшоты контейнеров, алерты и их доставки, журналы поисков по логам и вопросов агенту. @@ -150,7 +154,8 @@ make cov # pytest с покрытием (порог 80 %) | Docker Engine API | контейнеры, stats, логи | только GET через `docker-socket-proxy` | | OpenAI-совместимый `/chat/completions` | страж и планировщик агента | httpx, `response_format: json_schema` | | SSH → `tgit-runner` | деплой и перезапуск сервисов на хосте | `ssh` подпроцессом, ключ с forced command | -| SourceCraft REST API | статус прогонов CI репозитория загрузок | httpx, клиент готов, в UI — #11 | +| Gitea Actions API | прогон CI по sha, упавший шаг, лог джоба | httpx с ретраями, `app/pipeline/gitea.py` | +| SourceCraft REST API | прогоны CI — на будущее возвращение (ADR-016) | httpx, клиент готов, к наблюдателю не подключён | ## 5. Ограничения и особенности @@ -178,9 +183,8 @@ make cov # pytest с покрытием (порог 80 %) | повтор алерта | в течение 10 мин — счётчик, а не новое сообщение | **Что пока не сделано** (открытые задачи): -- проверки CI репозитория загрузок (ruff, bandit, тесты) перед деплоем: сейчас бот разворачивает сервис сразу после - push, проверив только синтаксис; ожидание зелёного CI появится вместе с #11; -- уведомления о прогонах CI/CD репозитория загрузок и алерты 📦/⛔ — #11, ждут возврата к SourceCraft; +- CI репозитория загрузок в SourceCraft: сейчас — Gitea Actions (ADR-016); SourceCraft подключится второй + реализацией `CIProvider`, когда он примет ключ бота; - «🔕 Тишина на час» для алертов (P2); - prod-окружение: нужен отдельный бот, деплой `main` пропускается, пока его нет. @@ -195,7 +199,8 @@ app/ logs/ grep-движок и журнал agent/ фильтр, страж, планировщик, схема LogQuery alerts/ детекторы, сервис алертов, рассылка, монитор - runner/ SSH-клиент tgit-runner и деплой после push + pipeline/ наблюдатель CI/CD после push, клиент Gitea Actions + runner/ SSH-клиент tgit-runner и деплой сервисов llm/ провайдеры: OpenAI-совместимый, mock sourcecraft/ клиент API прогонов db/ модели, миграции Alembic, очистка diff --git a/app/bot/routers/commit.py b/app/bot/routers/commit.py index 8680983..b350dcf 100644 --- a/app/bot/routers/commit.py +++ b/app/bot/routers/commit.py @@ -4,7 +4,6 @@ import base64 import html import io import logging -from collections.abc import Awaitable, Callable from dataclasses import dataclass from enum import StrEnum from typing import Any @@ -19,6 +18,7 @@ from aiogram.utils.keyboard import InlineKeyboardBuilder from app.bot.menu import Nav, NavAction, free_input from app.db.models import ChangeOp, User +from app.pipeline.watcher import PipelineWatcher, initial_status, render from app.repo.drafts import DraftFull, DraftService, DraftView from app.repo.git import Entry, GitError, UploadsRepo from app.repo.publisher import Publisher, PushResult, PushStatus @@ -34,8 +34,6 @@ from app.repo.validate import ( python_syntax, strip_fence, ) -from app.runner.deployer import Deployer -from app.runner.deployer import Outcome as DeployOutcome from app.texts import CommitTexts, Texts log = logging.getLogger(__name__) @@ -220,7 +218,7 @@ def build_router( drafts: DraftService, publisher: Publisher, commit_url_template: str = "", - deployer: Deployer | None = None, + pipeline: PipelineWatcher | None = None, ) -> Router: router = Router(name="commit") t = texts.commit @@ -283,21 +281,6 @@ def build_router( ops = {ChangeOp.ADD: t.op_add, ChangeOp.MODIFY: t.op_modify, ChangeOp.DELETE: t.op_delete} return "\n".join(f"{ops[c.op]} /{c.path}" for c in draft.changes) - def deploy_report(msg: Message, sha: str) -> Callable[[DeployOutcome], Awaitable[None]]: - async def report(o: DeployOutcome) -> None: - name = html.escape(o.service) - if o.result.ok and o.removed: - text = t.removed.format(service=name) - elif o.result.ok: - text = t.deployed.format(service=name, sha=sha[:7]) - else: - text = t.deploy_failed.format(service=name, reason=html.escape(o.result.summary)) - if o.result.tail: - text += f"\n
{html.escape(o.result.tail[-3000:])}
" - await msg.answer(text, parse_mode=ParseMode.HTML) - - return report - async def do_push(msg: Message, state: FSMContext, member: User, *, overwrite: bool) -> None: draft = await drafts.get(member.tg_id) if draft is None or not draft.changes or not draft.message: @@ -308,28 +291,20 @@ def build_router( result: PushResult = await publisher.push(member, draft, draft.message, overwrite=overwrite) match result.status: case PushStatus.OK: - assert result.sha is not None - first_line = draft.message.splitlines()[0] - lines = [t.pushed.format(sha=result.sha[:7], message=first_line)] - if commit_url_template: - lines.append(commit_url_template.format(sha=result.sha)) - lines.append(t.ci_pending) - if deployer is not None: - services = deployer.schedule( - result.sha, [c.path for c in draft.changes], deploy_report(msg, result.sha) - ) - if services: - lines.append(t.deploying.format(services=", ".join(services))) - await msg.edit_text("\n".join(lines)) - await publisher.record(member, result, draft.message, draft, msg.chat.id, msg.message_id) + paths = [c.path for c in draft.changes] + status = pipeline.initial_status(paths) if pipeline else initial_status(False, False, paths) + push = await publisher.record(member, result, draft.message, draft, msg.chat.id, msg.message_id, status) + await msg.edit_text(render(texts, push, commit_url_template=commit_url_template)) await drafts.clear(member.tg_id) await state.update_data(commit_virtual=[]) + if pipeline: + pipeline.wake() case PushStatus.CONFLICT: builder = InlineKeyboardBuilder() builder.button(text=t.overwrite, callback_data=CommitCb(a=Act.PUSH_FORCE)) builder.button(text=texts.menu.cancel, callback_data=CommitCb(a=Act.DRAFT)) - paths = "\n".join(f"/{p}" for p in result.conflicts) - await msg.edit_text(t.conflict.format(paths=paths), reply_markup=builder.as_markup()) + conflicts = "\n".join(f"/{p}" for p in result.conflicts) + await msg.edit_text(t.conflict.format(paths=conflicts), reply_markup=builder.as_markup()) case PushStatus.EMPTY: await drafts.clear(member.tg_id) await msg.edit_text(t.push_empty) diff --git a/app/config.py b/app/config.py index bfa4a8a..6b0829a 100644 --- a/app/config.py +++ b/app/config.py @@ -66,6 +66,23 @@ class SourceCraftSettings(BaseSettings): max_attempts: int = Field(default=5, ge=1) +class CISettings(BaseSettings): + model_config = SettingsConfigDict(**_ENV, env_prefix="CI_") + + # CI of the uploads repo the bot waits for before deploying (#11). none — deploy right after push, no CI status. + provider: Literal["none", "gitea"] = "none" + api_url: str = "https://gitea.24studbank.ru" + # Read access to the uploads repo's Actions is enough + token: SecretStr = SecretStr("") + repo: str = "HSE/tgit-uploads" + workflow: str = "check.yml" + poll_interval: float = Field(default=10.0, gt=0) + max_poll_interval: float = Field(default=60.0, gt=0) # backoff while the API is failing + run_timeout_minutes: float = Field(default=30.0, gt=0) + timeout: float = Field(default=30.0, gt=0) + max_attempts: int = Field(default=5, ge=1) + + class GitSettings(BaseSettings): model_config = SettingsConfigDict(**_ENV, env_prefix="GIT_") @@ -131,6 +148,7 @@ class Settings(BaseSettings): llm: LLMSettings = Field(default_factory=LLMSettings) sourcecraft: SourceCraftSettings = Field(default_factory=SourceCraftSettings) git: GitSettings = Field(default_factory=GitSettings) + ci: CISettings = Field(default_factory=CISettings) docker: DockerSettings = Field(default_factory=DockerSettings) runner: RunnerSettings = Field(default_factory=RunnerSettings) runtime: RuntimeSettings = Field(default_factory=RuntimeSettings) diff --git a/app/db/migrations/versions/20261019_0002_untracked_pushes.py b/app/db/migrations/versions/20261019_0002_untracked_pushes.py new file mode 100644 index 0000000..3788577 --- /dev/null +++ b/app/db/migrations/versions/20261019_0002_untracked_pushes.py @@ -0,0 +1,25 @@ +"""pushes made before the PipelineWatcher (#11): nobody waits for their CI/CD — mark them skipped + +Revision ID: 0002 +Revises: 0001 +Create Date: 2026-10-19 12:00:00 +""" + +from collections.abc import Sequence + +from alembic import op + +revision: str = "0002" +down_revision: str | None = "0001" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + # Otherwise the first watcher tick would time them out and message their authors about long-gone pushes + op.execute("UPDATE pushes SET ci_status = 'skipped' WHERE ci_status IN ('pending', 'running')") + op.execute("UPDATE pushes SET cd_status = 'skipped' WHERE cd_status IN ('pending', 'running')") + + +def downgrade() -> None: + pass # data only: nothing to undo diff --git a/app/main.py b/app/main.py index f9246c3..2e642b2 100644 --- a/app/main.py +++ b/app/main.py @@ -3,6 +3,7 @@ import asyncio import contextlib import logging +from datetime import timedelta from aiogram import Bot, Dispatcher from aiogram.client.session.aiohttp import AiohttpSession @@ -26,6 +27,8 @@ from app.health import heartbeat from app.llm.factory import make_provider from app.logs.journal import LogJournal from app.logs_setup import setup_logging +from app.pipeline.factory import make_ci +from app.pipeline.watcher import PipelineWatcher from app.repo.drafts import DraftService from app.repo.git import UploadsRepo from app.repo.publisher import Publisher @@ -50,6 +53,7 @@ def build_dispatcher( agent: Agent | None = None, alerts: AlertService | None = None, runner: RunnerClient | None = None, + pipeline: PipelineWatcher | None = None, ) -> Dispatcher: dp = Dispatcher(storage=make_storage(sessions)) gate = AccessMiddleware(access, texts) @@ -57,10 +61,9 @@ def build_dispatcher( dp.message.outer_middleware(gate) dp.callback_query.outer_middleware(gate) dp.include_router(members.build_router(texts, access)) - deployer = Deployer(repo, runner, alerts) if runner else None dp.include_router( commit.build_router( - texts, repo, DraftService(sessions), Publisher(repo, sessions), commit_url_template, deployer + texts, repo, DraftService(sessions), Publisher(repo, sessions), commit_url_template, pipeline ) ) dp.include_router(status.build_router(texts, docker, checker, runner)) @@ -95,35 +98,59 @@ async def run(settings: Settings) -> None: alerts_cfg, log_source=(lambda cid, n: docker.logs(cid, tail=n)) if docker else None, ) + repo = UploadsRepo(settings.git) + runner = RunnerClient(settings.runner) if settings.runner.host else None + ci = make_ci(settings.ci) + deployer = Deployer(repo, runner, alerts) if runner else None + # Without CI and runner there is nothing to watch: the push message just says what was pushed + pipeline = ( + PipelineWatcher( + sessions, + bot, + texts, + ci, + deployer, + commit_url_template=settings.git.commit_url_template, + poll_interval=settings.ci.poll_interval, + max_poll_interval=settings.ci.max_poll_interval, + run_timeout=timedelta(minutes=settings.ci.run_timeout_minutes), + ) + if ci or deployer + else None + ) dp = build_dispatcher( texts, sessions, access, - UploadsRepo(settings.git), + repo, settings.git.commit_url_template, docker, checker, agent, alerts, - RunnerClient(settings.runner) if settings.runner.host else None, + runner, + pipeline, ) rt = settings.runtime background = [ asyncio.create_task(heartbeat(rt.heartbeat_file, rt.heartbeat_interval)), asyncio.create_task(cleanup_loop(sessions)), ] + if pipeline: + background.append(asyncio.create_task(pipeline.run())) if docker: monitor = AlertMonitor(alerts, store, alerts_cfg) background.append( asyncio.create_task(snapshot_loop(docker, store, settings.docker.snapshot_interval, monitor.on_round)) ) log.info( - "tgit %s starting, %d bootstrap admin(s), telegram %s, docker %s, agent %s, runner %s", + "tgit %s starting, %d bootstrap admin(s), telegram %s, docker %s, agent %s, ci %s, runner %s", __version__, len(settings.access.admin_ids), "via proxy" if proxy else "direct", "on" if docker else "off", settings.llm.provider if agent else "off", + settings.ci.provider if ci else "off", settings.runner.host or "off", ) try: @@ -134,6 +161,10 @@ async def run(settings: Settings) -> None: for task in background: with contextlib.suppress(asyncio.CancelledError): await task + if pipeline: + await pipeline.aclose() + if ci: + await ci.aclose() await bot.session.close() if docker: await docker.aclose() diff --git a/app/pipeline/__init__.py b/app/pipeline/__init__.py new file mode 100644 index 0000000..9f5b8c2 --- /dev/null +++ b/app/pipeline/__init__.py @@ -0,0 +1 @@ +"""CI/CD of the uploads repo after a push: wait for the CI run, then deploy, telling the author (#11).""" diff --git a/app/pipeline/factory.py b/app/pipeline/factory.py new file mode 100644 index 0000000..721a697 --- /dev/null +++ b/app/pipeline/factory.py @@ -0,0 +1,10 @@ +from app.config import CISettings +from app.pipeline.gitea import GiteaCI +from app.pipeline.provider import CIProvider + + +def make_ci(settings: CISettings) -> CIProvider | None: + """None — no CI to wait for: services deploy right after push.""" + if settings.provider == "gitea": + return GiteaCI(settings) + return None diff --git a/app/pipeline/gitea.py b/app/pipeline/gitea.py new file mode 100644 index 0000000..ed1433e --- /dev/null +++ b/app/pipeline/gitea.py @@ -0,0 +1,118 @@ +"""Gitea Actions as the uploads CI (ADR-016). Unlike SourceCraft, a Gitea run carries `head_sha` — no meta cube.""" + +import asyncio +import logging +import re +from collections.abc import Awaitable, Callable +from typing import Any + +import httpx + +from app.config import CISettings +from app.db.models import PipelineStatus +from app.pipeline.provider import CIRun + +log = logging.getLogger(__name__) + +_TIMESTAMP = re.compile(r"^\d{4}-\d\d-\d\dT[\d:.]+Z ?", re.MULTILINE) +_FAILED = {"failure", "cancelled"} + +Sleep = Callable[[float], Awaitable[None]] + + +class GiteaError(Exception): + pass + + +def _status(status: str, conclusion: str | None) -> PipelineStatus: + if status != "completed": + return PipelineStatus.RUNNING if status == "in_progress" else PipelineStatus.PENDING # queued · waiting + if conclusion == "success": + return PipelineStatus.SUCCESS + if conclusion == "skipped": + return PipelineStatus.SKIPPED + return PipelineStatus.FAILED # failure · cancelled + + +class GiteaCI: + def __init__( + self, + settings: CISettings, + transport: httpx.AsyncBaseTransport | None = None, + sleep: Sleep = asyncio.sleep, + ) -> None: + self._s = settings + self._sleep = sleep + self._base = f"/api/v1/repos/{settings.repo}/actions" + self._http = httpx.AsyncClient( + base_url=settings.api_url, + headers={"Authorization": f"token {settings.token.get_secret_value()}"}, + timeout=settings.timeout, + transport=transport, + ) + + async def aclose(self) -> None: + await self._http.aclose() + + async def find(self, sha: str) -> CIRun | None: + data = await self._get(f"{self._base}/runs", params={"head_sha": sha, "limit": 20}) + # path is "@"; other workflows of the repo (if any) don't gate the deploy + runs = [r for r in data.get("workflow_runs") or [] if str(r.get("path", "")).split("@")[0] == self._s.workflow] + if not runs: + return None + run = max(runs, key=lambda r: int(r["id"])) # a re-run is a new id + status = _status(str(run.get("status")), run.get("conclusion")) + failed_step = await self._failed_step(run["id"]) if status is PipelineStatus.FAILED else None + return CIRun(str(run["id"]), status, str(run.get("html_url", "")), failed_step) + + async def failure_log(self, run: CIRun) -> str: + job = await self._failed_job(run.id) + if job is None: + return "" + text = await self._get_text(f"{self._base}/jobs/{job['id']}/logs") + return _TIMESTAMP.sub("", text) + + async def _failed_job(self, run_id: object) -> dict[str, Any] | None: + data = await self._get(f"{self._base}/runs/{run_id}/jobs") + jobs = data.get("jobs") or [] + return next((j for j in jobs if j.get("conclusion") in _FAILED), None) + + async def _failed_step(self, run_id: object) -> str | None: + job = await self._failed_job(run_id) + if job is None: + return None + step = next((s for s in job.get("steps") or [] if s.get("conclusion") in _FAILED), None) + if job.get("conclusion") == "cancelled" and step is None: + return f"{job.get('name')} / cancelled" + return f"{job.get('name')} / {step['name']}" if step else str(job.get("name")) + + async def _request(self, path: str, params: dict[str, Any] | None = None) -> httpx.Response: + attempts = self._s.max_attempts + for attempt in range(1, attempts + 1): + try: + response = await self._http.get(path, params=params) + except httpx.TransportError as e: + if attempt == attempts: + raise GiteaError(f"{path}: {type(e).__name__}") from e + log.warning("gitea %s: %s, retry %d/%d", path, type(e).__name__, attempt, attempts - 1) + else: + if response.status_code < 500 or attempt == attempts: + if response.status_code >= 400: + raise GiteaError(f"{path}: HTTP {response.status_code}") + return response + log.warning("gitea %s: HTTP %d, retry %d/%d", path, response.status_code, attempt, attempts - 1) + await self._sleep(min(0.5 * 2 ** (attempt - 1), 8.0)) + raise AssertionError("unreachable") + + async def _get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]: + response = await self._request(path, params) + try: + body = response.json() + except ValueError as e: + raise GiteaError(f"{path}: not JSON") from e + if not isinstance(body, dict): + raise GiteaError(f"{path}: expected a JSON object") + return body + + async def _get_text(self, path: str) -> str: + return (await self._request(path)).text diff --git a/app/pipeline/provider.py b/app/pipeline/provider.py new file mode 100644 index 0000000..66ff9c4 --- /dev/null +++ b/app/pipeline/provider.py @@ -0,0 +1,26 @@ +"""What the watcher needs from a CI: the run of a commit and the log of what failed. Vendors live next to it.""" + +from dataclasses import dataclass +from typing import Protocol + +from app.db.models import PipelineStatus + + +@dataclass(frozen=True) +class CIRun: + id: str + status: PipelineStatus # pending · running · success · failed · skipped + url: str + failed_step: str | None = None # "job / step" of the first failure + + +class CIProvider(Protocol): + async def find(self, sha: str) -> CIRun | None: + """The newest run of the uploads workflow for this commit; None — not started (yet).""" + ... + + async def failure_log(self, run: CIRun) -> str: + """Log of the failed job, plain text without runner timestamps.""" + ... + + async def aclose(self) -> None: ... diff --git a/app/pipeline/watcher.py b/app/pipeline/watcher.py new file mode 100644 index 0000000..3b26609 --- /dev/null +++ b/app/pipeline/watcher.py @@ -0,0 +1,361 @@ +"""PipelineWatcher: after a push — wait for the uploads CI, then deploy through tgit-runner (SPEC C-24, C-30…C-34). + +The truth is in the DB (`pushes.ci_status/cd_status`, `pipeline_runs`), so a restarted bot picks up where it stopped. +The CI is polled only while some push waits for it (ADR-003); with nothing to wait for the loop sleeps until `wake()`. + +The push message is edited in place (CI ⏳ → ✅/❌ → CD ⏳ → 🚀/⛔); the author also gets separate messages for the +outcomes — an edit doesn't notify. A failed deploy is an alert ⛔ for the whole team (raised by the Deployer). +""" + +import asyncio +import contextlib +import html +import logging +from collections.abc import Awaitable, Callable +from datetime import datetime, timedelta + +from aiogram import Bot +from aiogram.enums import ParseMode +from aiogram.exceptions import TelegramAPIError, TelegramBadRequest +from aiogram.types import BufferedInputFile, ReplyParameters +from sqlalchemy import or_, select, update + +from app.db.models import PipelineRun, PipelineStatus, Push, Stage, as_utc, utcnow +from app.db.session import SessionMaker +from app.pipeline.provider import CIProvider, CIRun +from app.runner.deployer import Deployer, Outcome, changed_services +from app.texts import Texts + +log = logging.getLogger(__name__) + +WAITING = (PipelineStatus.PENDING, PipelineStatus.RUNNING) +PASSED = (PipelineStatus.SUCCESS, PipelineStatus.SKIPPED) +RUNNER = "tgit-runner" # run_slug of the CD stage: the deploy is the bot's own, not a CI run +LOG_CHARS = 3000 # inline tail; the whole log goes as a file when it is longer +MAX_LOG_BYTES = 5 * 1024 * 1024 # SPEC L-5 + + +def initial_status(ci: bool, deploys: bool, paths: list[str]) -> tuple[PipelineStatus, PipelineStatus]: + """(ci, cd) of a fresh push: SKIPPED where there is nothing to wait for.""" + cd = PipelineStatus.PENDING if deploys and changed_services(paths) else PipelineStatus.SKIPPED + return (PipelineStatus.PENDING if ci else PipelineStatus.SKIPPED), cd + + +def render( + texts: Texts, + push: Push, + ci_run: PipelineRun | None = None, + cd_run: PipelineRun | None = None, + commit_url_template: str = "", + timeout_minutes: float = 30, +) -> str: + """The push message: what was pushed, then a line per stage that applies to it.""" + t = texts.pipeline + first_line = push.message.splitlines()[0] if push.message else "" + lines = [texts.commit.pushed.format(sha=push.sha[:7], message=first_line)] + if commit_url_template: + lines.append(commit_url_template.format(sha=push.sha)) + match push.ci_status: + case PipelineStatus.PENDING: + lines.append(t.ci_pending) + case PipelineStatus.RUNNING: + lines.append(t.ci_running) + case PipelineStatus.SUCCESS: + lines.append(t.ci_success) + case PipelineStatus.FAILED: + lines.append(t.ci_failed.format(step=(ci_run and ci_run.failed_cube) or "?")) + case PipelineStatus.TIMEOUT: + lines.append(t.ci_timeout.format(minutes=round(timeout_minutes))) + if ci_run is not None and ci_run.url: + lines.append(ci_run.url) + + services = ", ".join(changed_services(push.paths)) + failed = (cd_run.failed_cube or "") if cd_run else "" + match push.cd_status: + case PipelineStatus.PENDING if push.ci_status in WAITING: + lines.append(t.cd_waiting.format(services=services)) + case PipelineStatus.PENDING | PipelineStatus.RUNNING: + lines.append(t.cd_running.format(services=services)) + case PipelineStatus.SUCCESS: + lines.append(t.cd_success.format(services=services)) + case PipelineStatus.FAILED: + lines.append(t.cd_failed.format(services=failed.replace(",", ", "))) + case PipelineStatus.SKIPPED if services and push.ci_status not in PASSED: + lines.append(t.cd_not_run) + case PipelineStatus.SKIPPED if cd_run is not None: + lines.append(t.cd_superseded) + return "\n".join(lines) + + +class PipelineWatcher: + def __init__( + self, + sessions: SessionMaker, + bot: Bot, + texts: Texts, + ci: CIProvider | None, + deployer: Deployer | None, + *, + commit_url_template: str = "", + poll_interval: float = 10.0, + max_poll_interval: float = 60.0, + run_timeout: timedelta = timedelta(minutes=30), + clock: Callable[[], datetime] = utcnow, + ) -> None: + self._sessions = sessions + self._bot = bot + self._texts = texts + self._ci = ci + self._deployer = deployer + self._url = commit_url_template + self._interval = poll_interval + self._max_interval = max_poll_interval + self._timeout = run_timeout + self._clock = clock + self._wake = asyncio.Event() + self._deploying: dict[int, asyncio.Task[None]] = {} + + def initial_status(self, paths: list[str]) -> tuple[PipelineStatus, PipelineStatus]: + return initial_status(self._ci is not None, self._deployer is not None, paths) + + def wake(self) -> None: + """A new push: look at it now rather than after the current pause.""" + self._wake.set() + + async def run(self) -> None: + delay = self._interval + while True: + self._wake.clear() # before reading the DB: a push recorded during the tick wakes the next wait + try: + polling = await self.tick() + delay = self._interval + except Exception: + log.exception("pipeline watcher tick") + polling = True + delay = min(delay * 2, self._max_interval) + if polling: + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(self._wake.wait(), timeout=delay) + else: + await self._wake.wait() + + async def tick(self) -> bool: + """One pass over unfinished pushes. True — some push still waits for CI, poll again soon.""" + async with self._sessions() as s: + pushes = list( + await s.scalars( + select(Push) + .where( + or_( + Push.ci_status.in_(WAITING), + Push.ci_status.in_(PASSED) & Push.cd_status.in_(WAITING), + ) + ) + .order_by(Push.id) + ) + ) + polling = False + for push in pushes: + if push.ci_status in WAITING: + await self._poll_ci(push) + polling |= push.ci_status in WAITING + elif push.id not in self._deploying: + await self._start_cd(push) + return polling + + async def drain(self) -> None: + """Wait for the deploys in flight.""" + while self._deploying: + await asyncio.gather(*self._deploying.values(), return_exceptions=True) + + async def aclose(self) -> None: + """Stop deploys in flight: their push stays `cd=running` in the DB, and the next start deploys it again.""" + tasks = list(self._deploying.values()) + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + # ---- CI ------------------------------------------------------------------------------------------------------ + + async def _poll_ci(self, push: Push) -> None: + if self._ci is None: # CI switched off since the push: don't keep it waiting forever + await self._set(push, ci=PipelineStatus.SKIPPED) + await self._refresh(push) + return + run = await self._ci.find(push.sha) + if run is None or run.status in WAITING: + if self._clock() - as_utc(push.created_at) > self._timeout: + await self._finish_ci(push, PipelineStatus.TIMEOUT, run) + elif run is not None and run.status is not push.ci_status: + await self._save_run(push, Stage.CI, run.id, run.status, None, run.url) + await self._set(push, ci=run.status) + await self._refresh(push) + return + await self._finish_ci(push, run.status, run) + + async def _finish_ci(self, push: Push, status: PipelineStatus, run: CIRun | None) -> None: + if run is not None: + await self._save_run(push, Stage.CI, run.id, status, run.failed_step, run.url) + passed = status in PASSED + cd = push.cd_status if passed or push.cd_status is not PipelineStatus.PENDING else PipelineStatus.SKIPPED + await self._set(push, ci=status, cd=cd) + await self._refresh(push) + t = self._texts.pipeline + sha = push.sha[:7] + url = html.escape(run.url) if run else "" + if passed: + await self._notify(push, t.notify_ci_ok.format(sha=sha, url=url)) + if push.cd_status is PipelineStatus.PENDING: + await self._start_cd(push) + elif status is PipelineStatus.TIMEOUT: + minutes = round(self._timeout.total_seconds() / 60) + await self._notify(push, t.notify_ci_timeout.format(sha=sha, minutes=minutes)) + else: + assert run is not None + await self._notify_failure(push, run) + + async def _notify_failure(self, push: Push, run: CIRun) -> None: + t = self._texts.pipeline + step = html.escape(run.failed_step or "?") + text = t.notify_ci_failed.format(step=step, sha=push.sha[:7], url=html.escape(run.url)) + assert self._ci is not None + try: + full = await self._ci.failure_log(run) + except Exception as e: # the verdict matters more than its log + log.warning("ci log for %s: %s", push.sha[:7], e) + full = "" + tail = full.strip()[-LOG_CHARS:] + if tail: + text += f"\n
{html.escape(tail)}
" + await self._notify(push, text) + if len(full.strip()) > LOG_CHARS: + data = full.encode()[-MAX_LOG_BYTES:] + await self._send( + lambda: self._bot.send_document( + push.chat_id, + BufferedInputFile(data, f"ci-{push.sha[:7]}.log"), + caption=t.log_file, + reply_parameters=self._reply(push), + ) + ) + + # ---- CD ------------------------------------------------------------------------------------------------------ + + async def _start_cd(self, push: Push) -> None: + if self._deployer is None: # the runner was switched off since the push + await self._set(push, cd=PipelineStatus.SKIPPED) + await self._refresh(push) + return + services = await self._not_superseded(push) + if not services: + await self._save_run(push, Stage.CD, RUNNER, PipelineStatus.SKIPPED, None, None) + await self._set(push, cd=PipelineStatus.SKIPPED) + await self._refresh(push) + return + await self._set(push, cd=PipelineStatus.RUNNING) + await self._refresh(push) + task = asyncio.create_task(self._deploy(push, services)) + self._deploying[push.id] = task + task.add_done_callback(lambda _: self._deploying.pop(push.id, None)) + + async def _not_superseded(self, push: Push) -> list[str]: + """Services of this push that no newer passed push will deploy anyway (CI runs may finish out of order).""" + services = changed_services(push.paths) + async with self._sessions() as s: + newer = await s.scalars(select(Push.paths).where(Push.id > push.id, Push.ci_status.in_(PASSED))) + taken = {svc for paths in newer for svc in changed_services(paths)} + return [svc for svc in services if svc not in taken] + + async def _deploy(self, push: Push, services: list[str]) -> None: + assert self._deployer is not None + outcomes: list[Outcome] = [] + + async def report(o: Outcome) -> None: + outcomes.append(o) + t = self._texts.pipeline + name = html.escape(o.service) + if o.result.ok and o.removed: + await self._notify(push, t.removed.format(service=name)) + elif not o.result.ok: + text = t.deploy_failed.format(service=name, reason=html.escape(o.result.summary)) + if o.result.tail: + text += f"\n
{html.escape(o.result.tail[-LOG_CHARS:])}
" + await self._notify(push, text) + + try: + await self._deployer.deploy(push.sha, services, report) + except Exception: + log.exception("deploy of push %s", push.sha[:7]) + failed = [o.service for o in outcomes if not o.result.ok] + services[len(outcomes) :] + deployed = [o.service for o in outcomes if o.result.ok and not o.removed] + status = PipelineStatus.FAILED if failed else PipelineStatus.SUCCESS + await self._save_run(push, Stage.CD, RUNNER, status, ",".join(failed)[:100] or None, None) + await self._set(push, cd=status) + await self._refresh(push) + if deployed: + names = ", ".join(deployed) + await self._notify(push, self._texts.pipeline.notify_cd_ok.format(services=names, sha=push.sha[:7])) + + # ---- state and messages -------------------------------------------------------------------------------------- + + async def _set(self, push: Push, ci: PipelineStatus | None = None, cd: PipelineStatus | None = None) -> None: + values = {} + if ci is not None: + values["ci_status"] = push.ci_status = ci + if cd is not None: + values["cd_status"] = push.cd_status = cd + async with self._sessions.begin() as s: + await s.execute(update(Push).where(Push.id == push.id).values(**values)) + + async def _save_run( + self, push: Push, stage: Stage, slug: str, status: PipelineStatus, failed: str | None, url: str | None + ) -> None: + async with self._sessions.begin() as s: + row = await s.scalar(select(PipelineRun).where(PipelineRun.push_id == push.id, PipelineRun.stage == stage)) + if row is None: + row = PipelineRun(push_id=push.id, stage=stage) + s.add(row) + row.run_slug, row.status, row.failed_cube, row.url = slug, status, failed, url + + async def _runs(self, push: Push) -> dict[Stage, PipelineRun]: + async with self._sessions() as s: + rows = await s.scalars(select(PipelineRun).where(PipelineRun.push_id == push.id)) + return {r.stage: r for r in rows} + + async def _refresh(self, push: Push) -> None: + """Re-render the push message from the DB.""" + if push.message_id is None: + return + runs = await self._runs(push) + text = render( + self._texts, + push, + runs.get(Stage.CI), + runs.get(Stage.CD), + self._url, + self._timeout.total_seconds() / 60, + ) + message_id = push.message_id + await self._send(lambda: self._bot.edit_message_text(text, chat_id=push.chat_id, message_id=message_id)) + + def _reply(self, push: Push) -> ReplyParameters | None: + if push.message_id is None: + return None + return ReplyParameters(message_id=push.message_id, allow_sending_without_reply=True) + + async def _notify(self, push: Push, text: str) -> None: + await self._send( + lambda: self._bot.send_message( + push.chat_id, text, parse_mode=ParseMode.HTML, reply_parameters=self._reply(push) + ) + ) + + @staticmethod + async def _send(call: Callable[[], Awaitable[object]]) -> None: + """The status in the DB is what matters: a message that didn't get through is logged, not retried.""" + try: + await call() + except TelegramBadRequest as e: # not modified · deleted by the user · too old to edit + log.info("pipeline message: %s", e) + except TelegramAPIError as e: # blocked by the user · flood control + log.warning("pipeline message: %s", e) diff --git a/app/repo/publisher.py b/app/repo/publisher.py index 08dcfc5..2207e14 100644 --- a/app/repo/publisher.py +++ b/app/repo/publisher.py @@ -13,7 +13,7 @@ from pathlib import Path from sqlalchemy import func, select -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 DraftView from app.repo.git import GitError, UploadsRepo @@ -67,21 +67,31 @@ class Publisher: return PushResult(PushStatus.FAILED) async def record( - self, member: User, result: PushResult, message: str, draft: DraftView, chat_id: int, message_id: int - ) -> None: + self, + member: User, + result: PushResult, + message: str, + draft: DraftView, + chat_id: int, + message_id: int, + status: tuple[PipelineStatus, PipelineStatus], + ) -> Push: + """status — (ci, cd) to start from; the PipelineWatcher takes it from there (#11).""" assert result.sha is not None + push = Push( + user_id=member.tg_id, + sha=result.sha, + message=message, + paths=[c.path for c in draft.changes], + chat_id=chat_id, + message_id=message_id, + ci_status=status[0], + cd_status=status[1], + created_at=self._clock(), + ) async with self._sessions.begin() as s: - s.add( - Push( - user_id=member.tg_id, - sha=result.sha, - message=message, - paths=[c.path for c in draft.changes], - chat_id=chat_id, - message_id=message_id, - created_at=self._clock(), - ) - ) + s.add(push) + return push async def _push(self, member: User, draft: DraftView, message: str, overwrite: bool) -> PushResult: repo = self._repo diff --git a/app/repo/validate.py b/app/repo/validate.py index 9c28f28..3f845b0 100644 --- a/app/repo/validate.py +++ b/app/repo/validate.py @@ -8,7 +8,7 @@ from dataclasses import dataclass MAX_FILE_BYTES = 100 * 1024 NAME = re.compile(r"^[A-Za-z0-9._-]{1,100}$") # Bot users can't touch the CI of the uploads repo or its helper scripts (SPEC C-16, ADR-014) -PROTECTED_PREFIXES = (".sourcecraft/", ".tgit/", ".git/") +PROTECTED_PREFIXES = (".sourcecraft/", ".gitea/", ".tgit/", ".git/") PROTECTED_FILES = {"ruff.toml"} _FENCE = re.compile(r"^\s*```[\w+-]*[ \t]*\n(.*?)\n?```\s*$", re.DOTALL) diff --git a/app/runner/deployer.py b/app/runner/deployer.py index ef84c4e..105b52d 100644 --- a/app/runner/deployer.py +++ b/app/runner/deployer.py @@ -1,4 +1,4 @@ -"""After a push: the services it touched are (re)deployed through tgit-runner, the pusher hears how it went. +"""After a push passes CI: the services it touched are (re)deployed through tgit-runner (PipelineWatcher, #11). A service = `services//`. Gone from the commit — removed on the host. A refused deploy keeps the old version running (the runner switches only to a healthy one) and raises ⛔ for the whole team. @@ -43,16 +43,6 @@ class Deployer: self._runner = runner self._alerts = alerts self._lock = asyncio.Lock() # one deploy at a time: builds are heavy for the shared host - self._tasks: set[asyncio.Task[None]] = set() - - def schedule(self, sha: str, paths: Iterable[str], report: Report) -> list[str]: - """Start in the background; returns the services that will be deployed.""" - services = changed_services(paths) - if services: - task = asyncio.create_task(self.deploy(sha, services, report)) - self._tasks.add(task) - task.add_done_callback(self._tasks.discard) - return services async def deploy(self, sha: str, services: list[str], report: Report) -> None: async with self._lock: diff --git a/app/texts.py b/app/texts.py index 0dfc6cd..0c91dcb 100644 --- a/app/texts.py +++ b/app/texts.py @@ -113,16 +113,32 @@ class CommitTexts(_Strict): push_go: str pushing: str pushed: str - ci_pending: str conflict: str overwrite: str push_empty: str push_limit: str push_failed: str - deploying: str - deployed: str + + +class PipelineTexts(_Strict): + ci_pending: str + ci_running: str + ci_success: str + ci_failed: str + ci_timeout: str + cd_waiting: str + cd_running: str + cd_success: str + cd_failed: str + cd_not_run: str + cd_superseded: str + notify_ci_ok: str + notify_ci_failed: str + notify_ci_timeout: str + notify_cd_ok: str deploy_failed: str removed: str + log_file: str class StatusTexts(_Strict): @@ -234,6 +250,7 @@ class Texts(_Strict): section: SectionTexts members: MembersTexts commit: CommitTexts + pipeline: PipelineTexts status: StatusTexts logs: LogsTexts alerts: AlertsTexts diff --git a/config/texts.yaml b/config/texts.yaml index df2ebe0..4c0a668 100644 --- a/config/texts.yaml +++ b/config/texts.yaml @@ -119,7 +119,6 @@ commit: push_go: "✅ Запушить" pushing: "⏳ Пушу…" pushed: "✅ Запушено: {sha} — {message}" - ci_pending: "⏳ CI…" conflict: |- ⚠️ Пока вы готовили черновик, в main изменились те же файлы: {paths} @@ -129,12 +128,32 @@ commit: push_empty: "В main уже всё так же — коммитить нечего. Черновик очищен." push_limit: "Лимит — 20 пушей в час. Попробуйте позже." push_failed: "Не удалось запушить: репозиторий недоступен. Черновик сохранён — попробуйте позже." - deploying: "🚀 Разворачиваю: {services}…" - deployed: "🚀 {service} работает · {sha}" + +pipeline: + ci_pending: "CI: ⏳ ждёт раннер" + ci_running: "CI: ⏳ идёт проверка" + ci_success: "CI: ✅ пройден" + ci_failed: "CI: ❌ упал: {step}" + ci_timeout: "CI: ⌛ не завершился за {minutes} мин" + cd_waiting: "CD: ⏸ {services} — после CI" + cd_running: "CD: 🚀 разворачиваю {services}…" + cd_success: "CD: 🚀 {services} развёрнуто" + cd_failed: "CD: ⛔ не развёрнуто: {services}" + cd_not_run: "CD: ⏭ не запускался — CI не прошёл" + cd_superseded: "CD: ⏭ эти сервисы развернёт более новый пуш" + notify_ci_ok: |- + ✅ CI пройден · {sha} + {url} + notify_ci_failed: |- + ❌ CI упал на шаге {step} · {sha} + Сервисы не развёрнуты. {url} + notify_ci_timeout: "⌛ CI не завершился за {minutes} мин · {sha} — сервисы не развёрнуты." + notify_cd_ok: "🚀 CD: сервисы {services} развёрнуты, ревизия {sha}" deploy_failed: |- ⛔ {service} не развёрнут: {reason} Работает прежняя версия, если она была. removed: "🗑 {service} удалён из репозитория — остановлен на хосте" + log_file: "Полный лог CI" status: title: "🩺 Статус · {count} конт. · {time} UTC" diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 889cb75..9801a13 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -11,8 +11,8 @@ │ tgit-bot (один процесс, asyncio) │ │ ├─ bot/ меню, FSM, роутеры разделов, middleware доступа │ │ ├─ access/ приглашения (QR), участники, антиперебор │ - │ ├─ repo/ дерево репозитория, черновики, Publisher (git + SSH) ────────┼──► SourceCraft - │ ├─ pipeline/ PipelineWatcher: CI и CD по sha, уведомления автору ◄────────┼─── REST API + │ ├─ repo/ дерево репозитория, черновики, Publisher (git + SSH) ────────┼──► tgit-uploads (Gitea) + │ ├─ pipeline/ PipelineWatcher: CI по sha → CD, уведомления автору ◄────────┼─── Gitea Actions API │ ├─ runtime/ состояние контейнеров, проверки, логи │ │ ├─ logs/ grep-движок (Python, без shell) │ │ ├─ agent/ фильтр → LLM-страж → LLM → LogQuery ─────────────────────────┼──► LLM API @@ -28,7 +28,7 @@ │ │ tgit-svc- контейнеры (песочница) │ └─────┼──────────────────────────────────────────────────────────────────────── ┘ │ - SourceCraft CI (tgit-uploads): check → deploy ──┘ + бот (pipeline/), после зелёного CI tgit-uploads ──┘ ``` **Репозитории:** @@ -36,7 +36,7 @@ | Репозиторий | Что внутри | CI/CD | |---|---|---| | `tgit` (Gitea → зеркало в SourceCraft) | код бота | push в любую ветку: lint + types + tests; push `main`: + деплой бота на хост | -| `tgit-uploads` (SourceCraft) | общий репозиторий команды: `services//…` и любые файлы | push `main`: `check` → при успехе `deploy` изменённых сервисов через `tgit-runner` | +| `tgit-uploads` (Gitea, ADR-015/016) | общий репозиторий команды: `services//…` и любые файлы | push `main`: `check` в Gitea Actions → при успехе бот разворачивает изменённые сервисы через `tgit-runner` | ## 2. Компоненты @@ -73,10 +73,26 @@ aiogram 3. Главное меню — reply-клавиатура (📝 Комм pydantic-модели ответов. Методы: `list_runs`, `get_run`, `get_cube_logs` (постранично), `run_commit_sha` — sha из лога куба `check/meta` (у прогона в API нет коммита, ADR-014). Подробности — [`dev/SOURCECRAFT_API.md`](dev/SOURCECRAFT_API.md). ### 2.5 `app/pipeline` — наблюдатель CI/CD -Фоновая задача. Пока есть незавершённые пуши — опрашивает `/cicd/runs` (10 с, бэкофф до 60 с), сопоставляет прогон с -пушем по sha (из куба `meta`, ADR-014). Отслеживает задачи `check` (CI) и `deploy` (CD) workflow `check-and-deploy` отдельно; на смене статуса — редактирует сообщение -о пуше и шлёт автору уведомление (успех / падение + лог упавшего куба). Падение `deploy` → событие для `alerts`. -Таймаут — 30 мин. Webhooks у SourceCraft нет (ADR-003). +CI репозитория загрузок — Gitea Actions (`.gitea/workflows/check.yml`), CD — сам бот через `tgit-runner` (ADR-016). + +``` +push ─► pushes(ci=pending, cd=pending|skipped) ─► wake() +PipelineWatcher: CIProvider.find(sha) каждые 10 с (бэкофф до 60 с при ошибках API) + ├ pending/running ─► правка сообщения о push + ├ success/skipped ─► «✅ CI пройден» ─► Deployer (tgit-runner) ─► 🚀 / ⛔ + алерт ⛔ всем + ├ failed ─► «❌ CI упал на шаге …» + хвост лога (длинный — ещё и файлом), CD не запускается + └ нет вердикта 30 мин ─► ⌛ timeout, CD не запускается +``` +- **Провайдер за протоколом** `CIProvider` (`find(sha)`, `failure_log(run)`): `gitea.py` — у прогона Gitea есть + `head_sha`, сопоставление прямое; SourceCraft (ADR-014, куб `meta`) подключится второй реализацией. `CI_PROVIDER=none` — + CI не ждём, сервисы разворачиваются сразу после push. +- **Состояние — в БД**: `pushes.ci_status/cd_status` и `pipeline_runs` (`ci` — id прогона, ссылка, упавший шаг; + `cd` — `tgit-runner`, не развёрнутые сервисы). Сообщение о push каждый раз рендерится из БД. После рестарта бот + продолжает: ждёт CI дальше, а прерванный деплой (`cd=running`) повторяет. +- **Опрос только по делу**: нет незавершённых push — цикл спит до `wake()` от нового push. +- **Порядок версий**: прогоны CI могут закончиться не по порядку; старый push не разворачивает сервис, который уже + развернёт более новый прошедший CI push — иначе откатил бы его. +- Правка сообщения не уведомляет, поэтому исходы — отдельными сообщениями автору ответом на сообщение о push (C-30…C-32). ### 2.6 `app/runtime` — контейнеры и проверки Через `docker-socket-proxy` (только `GET`): @@ -134,7 +150,7 @@ pydantic-модели ответов. Методы: `list_runs`, `get_run`, `get закрывает. Пропущенное событие не теряет алерт, а бесконечный рестарт остаётся одним алертом. - **Дедупликация** (`service.py`): ключ `kind + service`. Условие, вернувшееся в течение 10 мин после закрытия, снова открывает тот же алерт с `count += 1`: исходные сообщения редактируются, новые не шлются. -- **Разовые события** (🛡 атака через агента; 📦 CD и ⛔ отказ после деплоя — после #11/#13) открываются сразу +- **Разовые события** (🛡 атака через агента; ⛔ отказ при деплое — от `Deployer`) открываются сразу закрытыми; повтор в окне 10 мин — тоже счётчик. - **Рассылка** (`broadcast.py`) — всем активным участникам, 🛡 — только администраторам; ~25 сообщений/с, `RetryAfter` пережидается, заблокировавший бота участник пропускается. `alert_deliveries` хранит @@ -186,9 +202,9 @@ SQLAlchemy 2 (async) + Alembic. PostgreSQL в Compose; в тестах — SQLit │ Publisher (очередь) ─ fetch, конфликт?, commit, push ──► tgit-uploads@main │ SourceCraft CI - «a1b2c3d ⏳ CI» ◄──────────────────────────────────────────────────────────────────────────┘ check: py_compile · ruff · pytest · bandit - PipelineWatcher ─ poll runs(sha) ─► «✅ CI» ─► «⏳ CD» ─► «🚀 CD: weather» deploy: ssh → tgit-runner deploy - └► ❌ + лог шага автору; ❌ CD → алерт всем + «a1b2c3d ⏳ CI» ◄──────────────────────────────────────────────────────────────────────────┘ Gitea Actions check: compile · ruff · bandit · pytest + PipelineWatcher ─ poll runs(sha) ─► «✅ CI» ─► «🚀 CD: weather…» ─► «🚀 CD: weather развёрнут» CD: бот → ssh tgit-runner deploy + └► ❌ + лог шага автору; ⛔ CD → алерт всем ``` ### 3.3 Логи через агента @@ -204,7 +220,7 @@ SQLAlchemy 2 (async) + Alembic. PostgreSQL в Compose; в тестах — SQLit `probe ` (HTTP-проба изнутри сети сервиса). Аргументы валидируются: имя `^[a-z][a-z0-9-]{1,30}$`, sha — hex. `deploy`: получает архив `services/` в stdin — его делает бот (`git archive :services/`) -сразу после push, так что у хоста нет доступа к репозиторию → проверяет архив (только файлы и папки, без `..`, +после зелёного CI (§2.5), так что у хоста нет доступа к репозиторию → проверяет архив (только файлы и папки, без `..`, ≤ 20 МБ, есть `main.py`) → собирает образ `tgit-svc-:` из шаблона (`python:3.12-slim` + `requirements.txt`, `HEALTHCHECK` из `tgit.yaml` сервиса, если задан) → запускает: @@ -235,13 +251,15 @@ docker run -d --name tgit-svc- --label tgit.managed=true --label tgit.s | push в любую ветку (через зеркало Gitea; PR живут в Gitea — ADR-010) | `check` | `uv sync --frozen` → `ruff check` + `ruff format --check` → `mypy app` → `pytest --cov=app --cov-fail-under=70` | | push в `main` | `check` + `deploy` | → `infra/deploy/deploy.sh` по SSH (секреты `DEPLOY_SSH_KEY`, `DEPLOY_HOST`) → smoke: healthcheck контейнера | -### 5.2 Репозиторий загрузок (`tgit-uploads/.sourcecraft/ci.yaml`) -Шаблон хранится в `tgit/infra/uploads-template/` и ставится в `tgit-uploads` один раз. +### 5.2 Репозиторий загрузок (`HSE/tgit-uploads`, `.gitea/workflows/check.yml`) +Шаблон хранится в `tgit/infra/uploads-template/` и ставится в `tgit-uploads` один раз (ADR-016). -| Workflow | Шаги | +| Что | Шаги | |---|---| -| `check` | изменённые файлы (`git diff --name-only HEAD~1`) → `py_compile` · `ruff check` (мягкий набор правил) · `pytest` для затронутых сервисов · `bandit` | -| `deploy` (после `check`) | изменённые `services//` → `ssh tgit-deploy@host deploy ` для каждого | +| `check` (Gitea Actions, push в `main`) | `compileall` → `ruff check` (мягкий набор правил) → `bandit -ll` → `pytest` для сервисов с тестами; первый упавший шаг бот показывает автору | +| CD (бот, после зелёного `check`) | изменённые `services//` → `git archive` → `ssh tgit-deploy@host deploy ` для каждого | + +`.sourcecraft/ci.yaml` в шаблоне остаётся для возврата к SourceCraft. ## 6. Деплой diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index 86136de..a8c4bb4 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -137,3 +137,21 @@ **Решение:** CI (`.gitea/workflows/ci.yml`: ruff, mypy, миграции на Postgres, pytest) и CD (`.gitea/workflows/deploy.yml` → `infra/deploy/deploy.sh`) — на Gitea Actions, как у Vobot и meditron, на общих раннерах Gitea. Деплой — на хост Vobot, `dev`/`prod` по веткам, через пользователя `tgit-deploy`. Репозиторий загрузок на сервере — временно `HSE/tgit-uploads` в Gitea. **Альтернативы:** ждать SourceCraft — блокирует проверку и демо. **Последствия:** `.sourcecraft/ci.yaml` бота остаётся для будущего возврата; при возврате к SourceCraft меняются `GIT_UPLOADS_URL` и ключ, код не трогаем. На RF-сервере Telegram — только через прокси (`TELEGRAM_PROXY_URL`). + +## ADR-016: CI репозитория загрузок — Gitea Actions; CD — бот после зелёного CI +**Дата:** 2026-10-19 · **Статус:** принято (меняет ADR-003/014 в части репозитория загрузок; SourceCraft — при возврате) + +**Контекст:** после ADR-015 репозиторий загрузок живёт в Gitea (`HSE/tgit-uploads`), а CI у него не было: бот +разворачивал сервис сразу после push, проверив только синтаксис, и «⏳ CI» в сообщении не обновлялся (#11). SourceCraft +по-прежнему не принимает ключ бота (#7). +**Решение:** CI загрузок — Gitea Actions (`.gitea/workflows/check.yml`: compile · ruff · bandit · pytest). Бот +опрашивает Gitea API (`/actions/runs?head_sha=…`) — у прогона есть sha, куб `meta` не нужен. CD остаётся у бота: после +зелёного `check` он отправляет изменённые сервисы в `tgit-runner`, как в #13. Наблюдатель зависит от протокола +`CIProvider`, вендорский код — в `app/pipeline/gitea.py`. +**Альтернативы:** SourceCraft — блокирован ключом бота; деплой из самого workflow (ssh на хост из раннера Gitea) — ключ +к `tgit-runner` пришлось бы отдать общим раннерам Gitea, а бот всё равно должен знать исход деплоя для автора и алертов; +не ждать CI — ruff/bandit/тесты из CI ничего бы не решали. +**Последствия:** боту нужен токен с чтением Actions репозитория загрузок (`CI_TOKEN`); без `CI_PROVIDER` поведение +прежнее (деплой сразу). Деплой ждёт CI: на общих раннерах это от ~30 с до минут при очереди, таймаут — 30 мин. +Возврат к SourceCraft — вторая реализация `CIProvider` поверх готового `SourceCraftClient` (ADR-014), код наблюдателя +не меняется. diff --git a/docs/dev/DEPLOY.md b/docs/dev/DEPLOY.md index cec1b90..db6f578 100644 --- a/docs/dev/DEPLOY.md +++ b/docs/dev/DEPLOY.md @@ -12,6 +12,7 @@ CD — `.gitea/workflows/deploy.yml` (Gitea Actions, ADR-015): push в `dev` → | Контейнеры | Compose-проект `tgit-`: `tgit--bot-1`, `tgit--postgres-1`; volumes `pgdata`, `uploads` | | Telegram | с сервера напрямую недоступен → `TELEGRAM_PROXY_URL` (тот же прокси, что у meditron) | | Репозиторий загрузок | временно `ssh://git@gitea.24studbank.ru:2222/HSE/tgit-uploads.git`, deploy key бота с записью | +| CI загрузок | Gitea Actions `HSE/tgit-uploads` (`check.yml`); боту в `.env` — `CI_PROVIDER=gitea`, `CI_TOKEN` (чтение репозитория загрузок). Без них деплой сервисов идёт сразу после push, без CI | Секреты репозитория в Gitea (Settings → Actions → Secrets): `DEPLOY_SSH_KEY` (приватный ключ `tgit-cd@gitea`), `DEPLOY_HOST` (`tgit-deploy@72.56.247.166`). @@ -42,4 +43,4 @@ cd /srv/tgit/dev && rm -rf app && mv app.old app && cd app && docker compose -p Нужен **отдельный токен бота** (один токен — один процесс polling): создать бота у @BotFather → `/srv/tgit/prod/.env` по образцу dev (`TELEGRAM_BOT_TOKEN`, `ADMIN_IDS`, `POSTGRES_PASSWORD`, `GIT_UPLOADS_URL`, -`GIT_SSH_KEY_B64`, `TELEGRAM_PROXY_URL`, `LLM_PROVIDER`). Пока файла нет, деплой `main` пропускается. +`GIT_SSH_KEY_B64`, `TELEGRAM_PROXY_URL`, `LLM_PROVIDER`, `CI_PROVIDER`, `CI_TOKEN`, `RUNNER_*`). Пока файла нет, деплой `main` пропускается. diff --git a/infra/uploads-template/.gitea/workflows/check.yml b/infra/uploads-template/.gitea/workflows/check.yml new file mode 100644 index 0000000..7b2728e --- /dev/null +++ b/infra/uploads-template/.gitea/workflows/check.yml @@ -0,0 +1,39 @@ +# CI репозитория загрузок tgit на Gitea Actions (ADR-016). Шаблон живёт в HSE/tgit: infra/uploads-template/ — правки делать там. +# Бот находит этот прогон по sha своего push и разворачивает изменённые сервисы только после зелёного check (#11). +# Имя файла — CI_WORKFLOW бота. Из Telegram не редактируется (SPEC C-16). +name: check + +on: + push: + branches: [main] + +# Без cancel-in-progress: каждый push ждёт свой вердикт — иначе бот не дождётся отменённого прогона. + +jobs: + check: + runs-on: ubuntu-latest + timeout-minutes: 25 + steps: + - uses: actions/checkout@v4 + + - name: Setup Python 3.12 + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install uv + # pip, а не astral-sh/setup-uv: на self-hosted Gitea экшен ходит в GitHub API и падает (опыт Vobot) + run: pip install uv + + # Шаги последовательно: имя первого упавшего бот покажет автору push вместе с логом. + - name: compile + run: python -m compileall -q -x '(^|/)\.' . + + - name: lint + run: uvx ruff@0.13 check . + + - name: security + run: uvx bandit@1.8 -q -r services -ll + + - name: tests + run: python .tgit/run_tests.py diff --git a/infra/uploads-template/.sourcecraft/ci.yaml b/infra/uploads-template/.sourcecraft/ci.yaml index 1fd4d5c..11b1a05 100644 --- a/infra/uploads-template/.sourcecraft/ci.yaml +++ b/infra/uploads-template/.sourcecraft/ci.yaml @@ -47,6 +47,6 @@ workflows: script: - python .tgit/changed_services.py > changed.txt - cat changed.txt - # Разворачивает бот сразу после push (tgit-runner, #13); здесь — только список для лога. - # Когда CI переедет сюда целиком (#11), бот будет ждать зелёного check перед деплоем. + # Разворачивает бот после зелёного check (tgit-runner, #13); здесь — только список для лога. + # Сейчас CI загрузок — Gitea Actions (.gitea/workflows/check.yml, ADR-016); этот файл — для возврата. - 'echo "$(wc -l < changed.txt) service(s) changed — deployed by the bot via tgit-runner"' diff --git a/infra/uploads-template/README.md b/infra/uploads-template/README.md index f8d8c87..a85266e 100644 --- a/infra/uploads-template/README.md +++ b/infra/uploads-template/README.md @@ -11,12 +11,15 @@ services// сервис = папка; имя — ^[a-z][a-z0-9-]{1, requirements.txt зависимости (необязательно) test_*.py тесты pytest (необязательно, запускаются в CI) tgit.yaml настройки запуска: health_url, network (необязательно) -.sourcecraft/ci.yaml CI/CD — из Telegram не редактируется +.gitea/workflows/ CI (Gitea Actions) — из Telegram не редактируется +.sourcecraft/ci.yaml CI для SourceCraft — на будущее возвращение, сейчас не используется .tgit/ служебные скрипты CI — из Telegram не редактируются ruff.toml мягкий набор правил линтера ``` -## CI/CD (`check-and-deploy`, push в `main`) +## CI/CD (push в `main`) + +CI — workflow `check` в Gitea Actions; CD делает бот, когда `check` зелёный. | Шаг | Что делает | Падает, если | |---|---|---| @@ -24,7 +27,7 @@ ruff.toml мягкий набор правил линтера | `lint` | `ruff check` (мягкие правила) | неопределённые имена, неиспользуемые импорты, опасные значения по умолчанию | | `security` | `bandit -ll` по `services/` | найдена уязвимость уровня medium+ | | `tests` | `pytest` для каждого сервиса с тестами, со своими `requirements.txt` | упал тест | -| `deploy` | список изменённых сервисов → `tgit-runner` на хосте | сервис не поднялся (healthy) за 60 с | +| CD (бот) | изменённые сервисы → `tgit-runner` на хосте | сервис не поднялся (healthy) за 60 с — работает прежняя версия | ## Развернуть заново diff --git a/tests/bot/test_commit_push.py b/tests/bot/test_commit_push.py index e41c347..b70c3c5 100644 --- a/tests/bot/test_commit_push.py +++ b/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)) diff --git a/tests/pipeline/__init__.py b/tests/pipeline/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/pipeline/test_gitea.py b/tests/pipeline/test_gitea.py new file mode 100644 index 0000000..0ef06c8 --- /dev/null +++ b/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) diff --git a/tests/pipeline/test_watcher.py b/tests/pipeline/test_watcher.py new file mode 100644 index 0000000..09dbfa4 --- /dev/null +++ b/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 упал на шаге check / lint · aaaaaaa") + assert "
ruff check .\nF821 undefined name 'x'
" 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("⛔ weather не развёрнут: not healthy") and "
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)
diff --git a/tests/repo/test_publisher.py b/tests/repo/test_publisher.py
index bdf7e08..d6e34ab 100644
--- a/tests/repo/test_publisher.py
+++ b/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))
diff --git a/tests/repo/test_validate.py b/tests/repo/test_validate.py
index 5c48fdb..1f00583 100644
--- a/tests/repo/test_validate.py
+++ b/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),
diff --git a/tests/runner/test_deploy.py b/tests/runner/test_deploy.py
index 0c8f38f..5d4df80 100644
--- a/tests/runner/test_deploy.py
+++ b/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("🚀 hello работает · ")
+    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]