release: dev → main #54

Merged
Eleboba merged 62 commits from dev into main 2026-10-22 17:50:43 +03:00
40 changed files with 2504 additions and 20 deletions
Showing only changes of commit 4e5930b4c0 - Show all commits
+2 -1
View File
@@ -13,7 +13,8 @@ POSTGRES_PASSWORD=tgit # (secret на сервере) пароль Postg
# --- SourceCraft (#7, #10) --- # --- SourceCraft (#7, #10) ---
SOURCECRAFT_TOKEN= # (secret) персональный токен API SOURCECRAFT_TOKEN= # (secret) персональный токен API
UPLOADS_REPO=rogachevegorushka/tgit-uploads SOURCECRAFT_UPLOADS_REPO=rogachevegorushka/tgit-uploads
# SOURCECRAFT_RUN_URL_TEMPLATE=https://sourcecraft.dev/{repo}/cicd/runs/{run}
# --- Git: коммиты в репозиторий загрузок (#7, #9) --- # --- Git: коммиты в репозиторий загрузок (#7, #9) ---
GIT_SSH_KEY_PATH= # путь к приватному deploy-ключу бота (монтируется read-only); GIT_SSH_KEY_PATH= # путь к приватному deploy-ключу бота (монтируется read-only);
+3
View File
@@ -23,3 +23,6 @@ build/
.vscode/ .vscode/
.claude/worktrees/ .claude/worktrees/
.DS_Store .DS_Store
# local SQLite DB (bot without Docker)
*.db
+5
View File
@@ -0,0 +1,5 @@
# Developer CLI: `uv run alembic revision --autogenerate -m "..."`. The DB URL comes from DATABASE_URL (.env),
# the bot itself applies migrations at startup (app/db/migrate.py).
[alembic]
script_location = app.db:migrations
file_template = %%(year)d%%(month).2d%%(day).2d_%%(rev)s_%%(slug)s
+6 -4
View File
@@ -1,8 +1,10 @@
"""FSM storage factory. In-memory for now; Postgres-backed storage that survives restarts arrives with the DB (#5).""" """FSM storage factory: dialog state lives in the DB and survives restarts (SPEC M-4)."""
from aiogram.fsm.storage.base import BaseStorage from aiogram.fsm.storage.base import BaseStorage
from aiogram.fsm.storage.memory import MemoryStorage
from app.bot.fsm_storage import SqlStorage
from app.db.session import SessionMaker
def make_storage() -> BaseStorage: def make_storage(sessions: SessionMaker) -> BaseStorage:
return MemoryStorage() return SqlStorage(sessions)
+54
View File
@@ -0,0 +1,54 @@
"""aiogram FSM storage on top of the `fsm_state` table — dialog state survives bot restarts (SPEC M-4)."""
from collections.abc import Mapping
from typing import Any
from aiogram.fsm.state import State
from aiogram.fsm.storage.base import BaseStorage, StateType, StorageKey
from app.db.models import FsmState
from app.db.session import SessionMaker
def key_to_str(key: StorageKey) -> str:
parts = (key.bot_id, key.chat_id, key.user_id, key.thread_id, key.business_connection_id, key.destiny)
return ":".join("" if p is None else str(p) for p in parts)
class SqlStorage(BaseStorage):
def __init__(self, sessions: SessionMaker) -> None:
self._sessions = sessions
async def set_state(self, key: StorageKey, state: StateType = None) -> None:
value = state.state if isinstance(state, State) else state
await self._write(key, state=value)
async def get_state(self, key: StorageKey) -> str | None:
async with self._sessions() as session:
row = await session.get(FsmState, key_to_str(key))
return row.state if row else None
async def set_data(self, key: StorageKey, data: Mapping[str, Any]) -> None:
await self._write(key, data=dict(data))
async def get_data(self, key: StorageKey) -> dict[str, Any]:
async with self._sessions() as session:
row = await session.get(FsmState, key_to_str(key))
return dict(row.data) if row else {}
async def close(self) -> None:
pass # the engine is owned and disposed by the app
async def _write(self, key: StorageKey, **fields: Any) -> None:
async with self._sessions.begin() as session:
row = await session.get(FsmState, key_to_str(key))
state = fields.get("state", row.state if row else None)
data = fields.get("data", row.data if row else {})
if state is None and not data:
if row is not None:
await session.delete(row) # empty state: keep the table small
return
if row is None:
session.add(FsmState(key=key_to_str(key), state=state, data=data))
else:
row.state, row.data = state, data
+21
View File
@@ -42,6 +42,25 @@ class LLMSettings(BaseSettings):
temperature: float = Field(default=0.0, ge=0, le=2) temperature: float = Field(default=0.0, ge=0, le=2)
class DatabaseSettings(BaseSettings):
model_config = SettingsConfigDict(**_ENV, env_prefix="DATABASE_")
# Compose sets a PostgreSQL URL; the default lets the bot run locally without Docker.
url: str = "sqlite+aiosqlite:///./tgit.db"
class SourceCraftSettings(BaseSettings):
model_config = SettingsConfigDict(**_ENV, env_prefix="SOURCECRAFT_")
token: SecretStr = SecretStr("")
api_url: str = "https://api.sourcecraft.tech"
uploads_repo: str = "rogachevegorushka/tgit-uploads"
# Web link to a CI run. The SPA answers 200 on any path, so the format is unverified — adjust if it 404s in UI.
run_url_template: str = "https://sourcecraft.dev/{repo}/cicd/runs/{run}"
timeout: float = Field(default=30.0, gt=0)
max_attempts: int = Field(default=5, ge=1)
class RuntimeSettings(BaseSettings): class RuntimeSettings(BaseSettings):
model_config = _ENV model_config = _ENV
@@ -59,5 +78,7 @@ class Settings(BaseSettings):
telegram: TelegramSettings = Field(default_factory=TelegramSettings) telegram: TelegramSettings = Field(default_factory=TelegramSettings)
access: AccessSettings = Field(default_factory=AccessSettings) access: AccessSettings = Field(default_factory=AccessSettings)
database: DatabaseSettings = Field(default_factory=DatabaseSettings)
llm: LLMSettings = Field(default_factory=LLMSettings) llm: LLMSettings = Field(default_factory=LLMSettings)
sourcecraft: SourceCraftSettings = Field(default_factory=SourceCraftSettings)
runtime: RuntimeSettings = Field(default_factory=RuntimeSettings) runtime: RuntimeSettings = Field(default_factory=RuntimeSettings)
View File
+46
View File
@@ -0,0 +1,46 @@
"""Periodic cleanup of TTL data: expired drafts, old container snapshots and access attempts."""
import asyncio
import logging
from datetime import datetime, timedelta
from typing import Any, cast
from sqlalchemy import CursorResult, Delete, delete
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import AccessAttempt, ContainerSnapshot, Draft, utcnow
from app.db.session import SessionMaker
log = logging.getLogger(__name__)
SNAPSHOT_TTL = timedelta(days=1)
ATTEMPT_TTL = timedelta(days=1)
async def _delete(session: AsyncSession, stmt: Delete) -> int:
# DML executes into a CursorResult; the generic annotation of execute() just doesn't say so
return cast(CursorResult[Any], await session.execute(stmt)).rowcount
async def cleanup_expired(sessions: SessionMaker, now: datetime) -> dict[str, int]:
async with sessions.begin() as session:
return {
"drafts": await _delete(session, delete(Draft).where(Draft.expires_at < now)),
"snapshots": await _delete(
session, delete(ContainerSnapshot).where(ContainerSnapshot.at < now - SNAPSHOT_TTL)
),
"access_attempts": await _delete(
session, delete(AccessAttempt).where(AccessAttempt.at < now - ATTEMPT_TTL)
),
}
async def cleanup_loop(sessions: SessionMaker, interval: float = 3600.0) -> None:
while True:
try:
removed = await cleanup_expired(sessions, utcnow())
if any(removed.values()):
log.info("cleanup: %s", removed)
except Exception: # a failed cleanup must not kill the bot; retry next round
log.exception("cleanup failed")
await asyncio.sleep(interval)
+52
View File
@@ -0,0 +1,52 @@
"""Apply Alembic migrations programmatically at startup — no separate step in the container.
`python -m app.db.migrate [revision]` upgrades the DB from DATABASE_URL (default: head).
"""
import asyncio
import sys
from alembic import command
from alembic.config import Config
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import AsyncEngine
from app.config import DatabaseSettings
from app.db.session import make_engine
SCRIPT_LOCATION = "app.db:migrations"
def alembic_config(connection: Connection | None = None) -> Config:
cfg = Config()
cfg.set_main_option("script_location", SCRIPT_LOCATION)
# env.py reuses this connection instead of opening its own (async engine, one transaction)
cfg.attributes["connection"] = connection
return cfg
async def upgrade(engine: AsyncEngine, revision: str = "head") -> None:
async with engine.begin() as conn:
await conn.run_sync(lambda sync_conn: command.upgrade(alembic_config(sync_conn), revision))
async def downgrade(engine: AsyncEngine, revision: str) -> None:
async with engine.begin() as conn:
await conn.run_sync(lambda sync_conn: command.downgrade(alembic_config(sync_conn), revision))
def main() -> None:
revision = sys.argv[1] if len(sys.argv) > 1 else "head"
async def _run() -> None:
engine = make_engine(DatabaseSettings().url)
try:
await upgrade(engine, revision)
finally:
await engine.dispose()
asyncio.run(_run())
if __name__ == "__main__":
main()
View File
+39
View File
@@ -0,0 +1,39 @@
"""Alembic environment. Runs on a connection handed over by app.db.migrate (or the CLI via alembic.ini)."""
import asyncio
from alembic import context
from sqlalchemy.engine import Connection
from app.config import DatabaseSettings
from app.db.models import Base
from app.db.session import make_engine
target_metadata = Base.metadata
def _run(connection: Connection) -> None:
context.configure(
connection=connection,
target_metadata=target_metadata,
render_as_batch=connection.dialect.name == "sqlite", # SQLite can't ALTER most things in place
compare_type=True,
)
with context.begin_transaction():
context.run_migrations()
async def _run_cli() -> None:
engine = make_engine(DatabaseSettings().url)
try:
async with engine.begin() as conn:
await conn.run_sync(_run)
finally:
await engine.dispose()
connection = context.config.attributes.get("connection")
if connection is not None:
_run(connection)
else:
asyncio.run(_run_cli())
+25
View File
@@ -0,0 +1,25 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
${imports if imports else ""}
revision: str = ${repr(up_revision)}
down_revision: str | None = ${repr(down_revision)}
branch_labels: str | Sequence[str] | None = ${repr(branch_labels)}
depends_on: str | Sequence[str] | None = ${repr(depends_on)}
def upgrade() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}
@@ -0,0 +1,311 @@
"""initial schema
Revision ID: 0001
Revises:
Create Date: 2026-10-14 17:41:05.672912
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
revision: str = "0001"
down_revision: str | None = None
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
"access_attempts",
sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=False),
sa.Column("tg_id", sa.BigInteger(), nullable=False),
sa.Column("at", sa.DateTime(timezone=True), nullable=False),
sa.Column("ok", sa.Boolean(), nullable=False),
sa.PrimaryKeyConstraint("id", name=op.f("pk_access_attempts")),
)
with op.batch_alter_table("access_attempts", schema=None) as batch_op:
batch_op.create_index("ix_access_attempts_tg_id_at", ["tg_id", "at"], unique=False)
op.create_table(
"alerts",
sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=False),
sa.Column(
"kind",
sa.Enum(
"crash",
"oom",
"restart_loop",
"unhealthy",
"overload",
"deploy_refused",
"cd_failed",
"disk_low",
"dependency_down",
"agent_attack",
name="alert_kind",
native_enum=False,
create_constraint=True,
length=20,
),
nullable=False,
),
sa.Column("service", sa.String(length=100), nullable=True),
sa.Column(
"payload", sa.JSON().with_variant(postgresql.JSONB(astext_type=sa.Text()), "postgresql"), nullable=False
),
sa.Column("count", sa.Integer(), nullable=False),
sa.Column("opened_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("last_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("resolved_at", sa.DateTime(timezone=True), nullable=True),
sa.PrimaryKeyConstraint("id", name=op.f("pk_alerts")),
)
with op.batch_alter_table("alerts", schema=None) as batch_op:
batch_op.create_index("ix_alerts_kind_service_resolved_at", ["kind", "service", "resolved_at"], unique=False)
op.create_table(
"container_snapshots",
sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=False),
sa.Column("container", sa.String(length=100), nullable=False),
sa.Column("at", sa.DateTime(timezone=True), nullable=False),
sa.Column("restart_count", sa.Integer(), nullable=False),
sa.Column("cpu_pct", sa.Float(), nullable=False),
sa.Column("mem_pct", sa.Float(), nullable=False),
sa.Column("health", sa.String(length=20), nullable=True),
sa.PrimaryKeyConstraint("id", name=op.f("pk_container_snapshots")),
)
with op.batch_alter_table("container_snapshots", schema=None) as batch_op:
batch_op.create_index("ix_container_snapshots_container_at", ["container", "at"], unique=False)
op.create_table(
"fsm_state",
sa.Column("key", sa.String(length=255), nullable=False),
sa.Column("state", sa.String(length=255), nullable=True),
sa.Column(
"data", sa.JSON().with_variant(postgresql.JSONB(astext_type=sa.Text()), "postgresql"), nullable=False
),
sa.PrimaryKeyConstraint("key", name=op.f("pk_fsm_state")),
)
op.create_table(
"users",
sa.Column("tg_id", sa.BigInteger(), autoincrement=False, nullable=False),
sa.Column("username", sa.String(length=64), nullable=True),
sa.Column("full_name", sa.String(length=256), nullable=False),
sa.Column(
"role",
sa.Enum("member", "admin", name="role", native_enum=False, create_constraint=True, length=20),
nullable=False,
),
sa.Column("is_active", sa.Boolean(), nullable=False),
sa.Column("invited_by", sa.BigInteger(), nullable=True),
sa.Column("joined_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(["invited_by"], ["users.tg_id"], name=op.f("fk_users_invited_by_users")),
sa.PrimaryKeyConstraint("tg_id", name=op.f("pk_users")),
)
op.create_table(
"agent_requests",
sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=False),
sa.Column("user_id", sa.BigInteger(), nullable=False),
sa.Column("prompt", sa.Text(), nullable=False),
sa.Column(
"verdict",
sa.Enum("safe", "attack", "error", name="verdict", native_enum=False, create_constraint=True, length=20),
nullable=False,
),
sa.Column("reason", sa.Text(), nullable=True),
sa.Column(
"log_query", sa.JSON().with_variant(postgresql.JSONB(astext_type=sa.Text()), "postgresql"), nullable=True
),
sa.Column("tokens", sa.Integer(), nullable=False),
sa.Column("at", sa.DateTime(timezone=True), nullable=False),
sa.ForeignKeyConstraint(["user_id"], ["users.tg_id"], name=op.f("fk_agent_requests_user_id_users")),
sa.PrimaryKeyConstraint("id", name=op.f("pk_agent_requests")),
)
op.create_table(
"alert_deliveries",
sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=False),
sa.Column("alert_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=False),
sa.Column("chat_id", sa.BigInteger(), nullable=False),
sa.Column("message_id", sa.BigInteger(), nullable=False),
sa.ForeignKeyConstraint(
["alert_id"], ["alerts.id"], name=op.f("fk_alert_deliveries_alert_id_alerts"), ondelete="CASCADE"
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_alert_deliveries")),
sa.UniqueConstraint("alert_id", "chat_id", name=op.f("uq_alert_deliveries_alert_id")),
)
op.create_table(
"drafts",
sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=False),
sa.Column("user_id", sa.BigInteger(), nullable=False),
sa.Column("base_sha", sa.String(length=40), nullable=False),
sa.Column("message", sa.Text(), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
sa.ForeignKeyConstraint(["user_id"], ["users.tg_id"], name=op.f("fk_drafts_user_id_users"), ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id", name=op.f("pk_drafts")),
sa.UniqueConstraint("user_id", name=op.f("uq_drafts_user_id")),
)
op.create_table(
"invites",
sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=False),
sa.Column("token_hash", sa.String(length=64), nullable=False),
sa.Column("created_by", sa.BigInteger(), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("used_by", sa.BigInteger(), nullable=True),
sa.Column("used_at", sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(["created_by"], ["users.tg_id"], name=op.f("fk_invites_created_by_users")),
sa.ForeignKeyConstraint(["used_by"], ["users.tg_id"], name=op.f("fk_invites_used_by_users")),
sa.PrimaryKeyConstraint("id", name=op.f("pk_invites")),
sa.UniqueConstraint("token_hash", name=op.f("uq_invites_token_hash")),
)
op.create_table(
"log_queries",
sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=False),
sa.Column("user_id", sa.BigInteger(), nullable=False),
sa.Column("container", sa.String(length=100), nullable=False),
sa.Column(
"mode",
sa.Enum("tail", "grep", "agent", name="log_mode", native_enum=False, create_constraint=True, length=20),
nullable=False,
),
sa.Column(
"query", sa.JSON().with_variant(postgresql.JSONB(astext_type=sa.Text()), "postgresql"), nullable=False
),
sa.Column("lines", sa.Integer(), nullable=False),
sa.Column("at", sa.DateTime(timezone=True), nullable=False),
sa.ForeignKeyConstraint(["user_id"], ["users.tg_id"], name=op.f("fk_log_queries_user_id_users")),
sa.PrimaryKeyConstraint("id", name=op.f("pk_log_queries")),
)
op.create_table(
"pushes",
sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=False),
sa.Column("user_id", sa.BigInteger(), nullable=False),
sa.Column("sha", sa.String(length=40), nullable=False),
sa.Column("message", sa.Text(), nullable=False),
sa.Column(
"paths", sa.JSON().with_variant(postgresql.JSONB(astext_type=sa.Text()), "postgresql"), nullable=False
),
sa.Column("chat_id", sa.BigInteger(), nullable=False),
sa.Column("message_id", sa.BigInteger(), nullable=True),
sa.Column(
"ci_status",
sa.Enum(
"pending",
"running",
"success",
"failed",
"timeout",
"skipped",
name="ci_status",
native_enum=False,
create_constraint=True,
length=20,
),
nullable=False,
),
sa.Column(
"cd_status",
sa.Enum(
"pending",
"running",
"success",
"failed",
"timeout",
"skipped",
name="cd_status",
native_enum=False,
create_constraint=True,
length=20,
),
nullable=False,
),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.ForeignKeyConstraint(["user_id"], ["users.tg_id"], name=op.f("fk_pushes_user_id_users")),
sa.PrimaryKeyConstraint("id", name=op.f("pk_pushes")),
sa.UniqueConstraint("sha", name=op.f("uq_pushes_sha")),
)
op.create_table(
"draft_changes",
sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=False),
sa.Column("draft_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=False),
sa.Column("path", sa.String(length=500), nullable=False),
sa.Column(
"op",
sa.Enum("add", "modify", "delete", name="change_op", native_enum=False, create_constraint=True, length=20),
nullable=False,
),
sa.Column("content", sa.LargeBinary(), nullable=True),
sa.ForeignKeyConstraint(
["draft_id"], ["drafts.id"], name=op.f("fk_draft_changes_draft_id_drafts"), ondelete="CASCADE"
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_draft_changes")),
sa.UniqueConstraint("draft_id", "path", name=op.f("uq_draft_changes_draft_id")),
)
op.create_table(
"pipeline_runs",
sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=False),
sa.Column("push_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=False),
sa.Column(
"stage",
sa.Enum("ci", "cd", name="stage", native_enum=False, create_constraint=True, length=20),
nullable=False,
),
sa.Column("run_slug", sa.String(length=50), nullable=False),
sa.Column(
"status",
sa.Enum(
"pending",
"running",
"success",
"failed",
"timeout",
"skipped",
name="run_status",
native_enum=False,
create_constraint=True,
length=20,
),
nullable=False,
),
sa.Column("failed_cube", sa.String(length=100), nullable=True),
sa.Column("url", sa.String(length=500), nullable=True),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.ForeignKeyConstraint(
["push_id"], ["pushes.id"], name=op.f("fk_pipeline_runs_push_id_pushes"), ondelete="CASCADE"
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_pipeline_runs")),
sa.UniqueConstraint("push_id", "stage", name=op.f("uq_pipeline_runs_push_id")),
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table("pipeline_runs")
op.drop_table("draft_changes")
op.drop_table("pushes")
op.drop_table("log_queries")
op.drop_table("invites")
op.drop_table("drafts")
op.drop_table("alert_deliveries")
op.drop_table("agent_requests")
op.drop_table("users")
op.drop_table("fsm_state")
with op.batch_alter_table("container_snapshots", schema=None) as batch_op:
batch_op.drop_index("ix_container_snapshots_container_at")
op.drop_table("container_snapshots")
with op.batch_alter_table("alerts", schema=None) as batch_op:
batch_op.drop_index("ix_alerts_kind_service_resolved_at")
op.drop_table("alerts")
with op.batch_alter_table("access_attempts", schema=None) as batch_op:
batch_op.drop_index("ix_access_attempts_tg_id_at")
op.drop_table("access_attempts")
# ### end Alembic commands ###
+283
View File
@@ -0,0 +1,283 @@
"""Data model (ARCHITECTURE §2.10). One schema for PostgreSQL (prod) and SQLite (tests).
Enums are stored as VARCHAR + CHECK rather than native PG enums: portable and painless to migrate.
"""
from datetime import UTC, datetime
from enum import StrEnum
from typing import Any, ClassVar
from sqlalchemy import (
JSON,
BigInteger,
Boolean,
DateTime,
Enum,
Float,
ForeignKey,
Index,
Integer,
LargeBinary,
MetaData,
String,
Text,
UniqueConstraint,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
# Stable constraint names so Alembic migrations stay deterministic across databases.
NAMING = {
"ix": "ix_%(column_0_label)s",
"uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s",
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
"pk": "pk_%(table_name)s",
}
# Surrogate keys: BIGINT on Postgres, INTEGER on SQLite (only INTEGER PRIMARY KEY autoincrements there).
Id = BigInteger().with_variant(Integer(), "sqlite")
Json = JSON().with_variant(JSONB(), "postgresql")
TgId = BigInteger
def utcnow() -> datetime:
return datetime.now(UTC)
def enum(cls: type[StrEnum], name: str) -> Enum:
return Enum(
cls,
name=name,
native_enum=False,
create_constraint=True, # SQLAlchemy 2 skips the CHECK for non-native enums unless asked
length=20,
values_callable=lambda e: [m.value for m in e],
)
class Role(StrEnum):
MEMBER = "member"
ADMIN = "admin"
class ChangeOp(StrEnum):
ADD = "add"
MODIFY = "modify"
DELETE = "delete"
class Stage(StrEnum):
CI = "ci"
CD = "cd"
class PipelineStatus(StrEnum):
PENDING = "pending"
RUNNING = "running"
SUCCESS = "success"
FAILED = "failed"
TIMEOUT = "timeout"
SKIPPED = "skipped"
class AlertKind(StrEnum):
CRASH = "crash"
OOM = "oom"
RESTART_LOOP = "restart_loop"
UNHEALTHY = "unhealthy"
OVERLOAD = "overload"
DEPLOY_REFUSED = "deploy_refused"
CD_FAILED = "cd_failed"
DISK_LOW = "disk_low"
DEPENDENCY_DOWN = "dependency_down"
AGENT_ATTACK = "agent_attack"
class LogMode(StrEnum):
TAIL = "tail"
GREP = "grep"
AGENT = "agent"
class Verdict(StrEnum):
SAFE = "safe"
ATTACK = "attack"
ERROR = "error"
class Base(DeclarativeBase):
metadata = MetaData(naming_convention=NAMING)
type_annotation_map: ClassVar[dict[Any, Any]] = {
datetime: DateTime(timezone=True),
dict[str, Any]: Json,
list[str]: Json,
}
class User(Base):
__tablename__ = "users"
tg_id: Mapped[int] = mapped_column(TgId, primary_key=True, autoincrement=False)
username: Mapped[str | None] = mapped_column(String(64))
full_name: Mapped[str] = mapped_column(String(256))
role: Mapped[Role] = mapped_column(enum(Role, "role"), default=Role.MEMBER)
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
invited_by: Mapped[int | None] = mapped_column(TgId, ForeignKey("users.tg_id"))
joined_at: Mapped[datetime] = mapped_column(default=utcnow)
last_seen_at: Mapped[datetime | None]
class Invite(Base):
__tablename__ = "invites"
id: Mapped[int] = mapped_column(Id, primary_key=True)
# sha256 hex of the token; the token itself is never stored (SPEC A-3)
token_hash: Mapped[str] = mapped_column(String(64), unique=True)
created_by: Mapped[int] = mapped_column(TgId, ForeignKey("users.tg_id"))
created_at: Mapped[datetime] = mapped_column(default=utcnow)
expires_at: Mapped[datetime]
used_by: Mapped[int | None] = mapped_column(TgId, ForeignKey("users.tg_id"))
used_at: Mapped[datetime | None]
class AccessAttempt(Base):
__tablename__ = "access_attempts"
__table_args__ = (Index("ix_access_attempts_tg_id_at", "tg_id", "at"),)
id: Mapped[int] = mapped_column(Id, primary_key=True)
tg_id: Mapped[int] = mapped_column(TgId)
at: Mapped[datetime] = mapped_column(default=utcnow)
ok: Mapped[bool] = mapped_column(Boolean)
class FsmState(Base):
__tablename__ = "fsm_state"
key: Mapped[str] = mapped_column(String(255), primary_key=True)
state: Mapped[str | None] = mapped_column(String(255))
data: Mapped[dict[str, Any]] = mapped_column(default=dict)
class Draft(Base):
__tablename__ = "drafts"
id: Mapped[int] = mapped_column(Id, primary_key=True)
user_id: Mapped[int] = mapped_column(TgId, ForeignKey("users.tg_id", ondelete="CASCADE"), unique=True)
base_sha: Mapped[str] = mapped_column(String(40))
message: Mapped[str | None] = mapped_column(Text)
created_at: Mapped[datetime] = mapped_column(default=utcnow)
expires_at: Mapped[datetime]
changes: Mapped[list["DraftChange"]] = relationship(
back_populates="draft", cascade="all, delete-orphan", order_by="DraftChange.id"
)
class DraftChange(Base):
__tablename__ = "draft_changes"
__table_args__ = (UniqueConstraint("draft_id", "path"),)
id: Mapped[int] = mapped_column(Id, primary_key=True)
draft_id: Mapped[int] = mapped_column(Id, ForeignKey("drafts.id", ondelete="CASCADE"))
path: Mapped[str] = mapped_column(String(500))
op: Mapped[ChangeOp] = mapped_column(enum(ChangeOp, "change_op"))
content: Mapped[bytes | None] = mapped_column(LargeBinary)
draft: Mapped[Draft] = relationship(back_populates="changes")
class Push(Base):
__tablename__ = "pushes"
id: Mapped[int] = mapped_column(Id, primary_key=True)
user_id: Mapped[int] = mapped_column(TgId, ForeignKey("users.tg_id"))
sha: Mapped[str] = mapped_column(String(40), unique=True)
message: Mapped[str] = mapped_column(Text)
paths: Mapped[list[str]] = mapped_column(default=list)
chat_id: Mapped[int] = mapped_column(TgId)
message_id: Mapped[int | None] = mapped_column(BigInteger)
ci_status: Mapped[PipelineStatus] = mapped_column(enum(PipelineStatus, "ci_status"), default=PipelineStatus.PENDING)
cd_status: Mapped[PipelineStatus] = mapped_column(enum(PipelineStatus, "cd_status"), default=PipelineStatus.PENDING)
created_at: Mapped[datetime] = mapped_column(default=utcnow)
class PipelineRun(Base):
__tablename__ = "pipeline_runs"
__table_args__ = (UniqueConstraint("push_id", "stage"),)
id: Mapped[int] = mapped_column(Id, primary_key=True)
push_id: Mapped[int] = mapped_column(Id, ForeignKey("pushes.id", ondelete="CASCADE"))
# CI and CD are tasks of one SourceCraft workflow (check-and-deploy), hence "stage", not "workflow"
stage: Mapped[Stage] = mapped_column(enum(Stage, "stage"))
run_slug: Mapped[str] = mapped_column(String(50))
status: Mapped[PipelineStatus] = mapped_column(enum(PipelineStatus, "run_status"))
failed_cube: Mapped[str | None] = mapped_column(String(100))
url: Mapped[str | None] = mapped_column(String(500))
updated_at: Mapped[datetime] = mapped_column(default=utcnow, onupdate=utcnow)
class ContainerSnapshot(Base):
__tablename__ = "container_snapshots"
__table_args__ = (Index("ix_container_snapshots_container_at", "container", "at"),)
id: Mapped[int] = mapped_column(Id, primary_key=True)
container: Mapped[str] = mapped_column(String(100))
at: Mapped[datetime] = mapped_column(default=utcnow)
restart_count: Mapped[int] = mapped_column(Integer)
cpu_pct: Mapped[float] = mapped_column(Float)
mem_pct: Mapped[float] = mapped_column(Float)
health: Mapped[str | None] = mapped_column(String(20))
class Alert(Base):
__tablename__ = "alerts"
__table_args__ = (Index("ix_alerts_kind_service_resolved_at", "kind", "service", "resolved_at"),)
id: Mapped[int] = mapped_column(Id, primary_key=True)
kind: Mapped[AlertKind] = mapped_column(enum(AlertKind, "alert_kind"))
service: Mapped[str | None] = mapped_column(String(100))
payload: Mapped[dict[str, Any]] = mapped_column(default=dict)
count: Mapped[int] = mapped_column(Integer, default=1)
opened_at: Mapped[datetime] = mapped_column(default=utcnow)
last_at: Mapped[datetime] = mapped_column(default=utcnow)
resolved_at: Mapped[datetime | None]
deliveries: Mapped[list["AlertDelivery"]] = relationship(back_populates="alert", cascade="all, delete-orphan")
class AlertDelivery(Base):
__tablename__ = "alert_deliveries"
__table_args__ = (UniqueConstraint("alert_id", "chat_id"),)
id: Mapped[int] = mapped_column(Id, primary_key=True)
alert_id: Mapped[int] = mapped_column(Id, ForeignKey("alerts.id", ondelete="CASCADE"))
chat_id: Mapped[int] = mapped_column(TgId)
message_id: Mapped[int] = mapped_column(BigInteger)
alert: Mapped[Alert] = relationship(back_populates="deliveries")
class LogQuery(Base):
__tablename__ = "log_queries"
id: Mapped[int] = mapped_column(Id, primary_key=True)
user_id: Mapped[int] = mapped_column(TgId, ForeignKey("users.tg_id"))
container: Mapped[str] = mapped_column(String(100))
mode: Mapped[LogMode] = mapped_column(enum(LogMode, "log_mode"))
query: Mapped[dict[str, Any]] = mapped_column(default=dict)
lines: Mapped[int] = mapped_column(Integer, default=0)
at: Mapped[datetime] = mapped_column(default=utcnow)
class AgentRequest(Base):
__tablename__ = "agent_requests"
id: Mapped[int] = mapped_column(Id, primary_key=True)
user_id: Mapped[int] = mapped_column(TgId, ForeignKey("users.tg_id"))
prompt: Mapped[str] = mapped_column(Text)
verdict: Mapped[Verdict] = mapped_column(enum(Verdict, "verdict"))
reason: Mapped[str | None] = mapped_column(Text)
log_query: Mapped[dict[str, Any] | None]
tokens: Mapped[int] = mapped_column(Integer, default=0)
at: Mapped[datetime] = mapped_column(default=utcnow)
+35
View File
@@ -0,0 +1,35 @@
"""Async engine and session factory."""
from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool
SessionMaker = async_sessionmaker[AsyncSession]
def make_engine(url: str) -> AsyncEngine:
if url.startswith("sqlite"):
# In-memory SQLite lives per connection: share one connection so every session sees the same DB.
kwargs = (
{"poolclass": StaticPool, "connect_args": {"check_same_thread": False}}
if ":memory:" in url or url.endswith("://")
else {}
)
engine = create_async_engine(url, **kwargs)
event.listen(engine.sync_engine, "connect", _sqlite_foreign_keys)
return engine
return create_async_engine(url, pool_pre_ping=True)
def _sqlite_foreign_keys(dbapi_conn: object, _record: object) -> None:
# SQLite ignores FOREIGN KEY / ON DELETE CASCADE unless asked per connection
cursor = dbapi_conn.cursor() # type: ignore[attr-defined]
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
def make_sessionmaker(engine: AsyncEngine) -> SessionMaker:
return async_sessionmaker(engine, expire_on_commit=False)
__all__ = ["SessionMaker", "make_engine", "make_sessionmaker"]
+18 -6
View File
@@ -11,6 +11,9 @@ from app.bot.fsm import make_storage
from app.bot.middlewares.access import AccessMiddleware from app.bot.middlewares.access import AccessMiddleware
from app.bot.routers.common import build_router from app.bot.routers.common import build_router
from app.config import Settings from app.config import Settings
from app.db.maintenance import cleanup_loop
from app.db.migrate import upgrade
from app.db.session import SessionMaker, make_engine, make_sessionmaker
from app.health import heartbeat from app.health import heartbeat
from app.logs_setup import setup_logging from app.logs_setup import setup_logging
from app.texts import Texts, load_texts from app.texts import Texts, load_texts
@@ -18,13 +21,13 @@ from app.texts import Texts, load_texts
log = logging.getLogger("tgit") log = logging.getLogger("tgit")
def build_dispatcher(settings: Settings, texts: Texts) -> Dispatcher: def build_dispatcher(settings: Settings, texts: Texts, sessions: SessionMaker) -> Dispatcher:
admin_ids = settings.access.admin_ids admin_ids = settings.access.admin_ids
async def is_member(tg_id: int) -> bool: async def is_member(tg_id: int) -> bool:
return tg_id in admin_ids return tg_id in admin_ids
dp = Dispatcher(storage=make_storage()) dp = Dispatcher(storage=make_storage(sessions))
access = AccessMiddleware(is_member, texts) access = AccessMiddleware(is_member, texts)
# Outer middleware: runs before filters, so nothing — not even a stale button — reaches a handler unchecked. # Outer middleware: runs before filters, so nothing — not even a stale button — reaches a handler unchecked.
dp.message.outer_middleware(access) dp.message.outer_middleware(access)
@@ -35,18 +38,27 @@ def build_dispatcher(settings: Settings, texts: Texts) -> Dispatcher:
async def run(settings: Settings) -> None: async def run(settings: Settings) -> None:
texts = load_texts(settings.runtime.config_dir) texts = load_texts(settings.runtime.config_dir)
dp = build_dispatcher(settings, texts) engine = make_engine(settings.database.url)
await upgrade(engine)
sessions = make_sessionmaker(engine)
dp = build_dispatcher(settings, texts, sessions)
rt = settings.runtime rt = settings.runtime
beat = asyncio.create_task(heartbeat(rt.heartbeat_file, rt.heartbeat_interval)) background = [
asyncio.create_task(heartbeat(rt.heartbeat_file, rt.heartbeat_interval)),
asyncio.create_task(cleanup_loop(sessions)),
]
bot = Bot(settings.telegram.bot_token.get_secret_value()) bot = Bot(settings.telegram.bot_token.get_secret_value())
log.info("tgit %s starting, %d bootstrap admin(s)", __version__, len(settings.access.admin_ids)) log.info("tgit %s starting, %d bootstrap admin(s)", __version__, len(settings.access.admin_ids))
try: try:
await dp.start_polling(bot, handle_signals=True) await dp.start_polling(bot, handle_signals=True)
finally: finally:
beat.cancel() for task in background:
task.cancel()
for task in background:
with contextlib.suppress(asyncio.CancelledError): with contextlib.suppress(asyncio.CancelledError):
await beat await task
await bot.session.close() await bot.session.close()
await engine.dispose()
log.info("tgit stopped") log.info("tgit stopped")
View File
+28
View File
@@ -0,0 +1,28 @@
"""Debug CLI: `python -m app.sourcecraft.cli runs [repo]` — recent CI runs with their commit and failed cubes."""
import asyncio
import sys
from app.config import SourceCraftSettings
from app.sourcecraft.client import SourceCraftClient
async def runs(repo: str | None) -> None:
settings = SourceCraftSettings()
repo = repo or settings.uploads_repo
async with SourceCraftClient(settings) as client:
for run in await client.list_runs(repo, page_size=10):
full = await client.get_run(repo, run.slug)
sha = await client.run_commit_sha(repo, full)
failed = ", ".join(f"{t}/{c.slug}" for _, t, c in full.failed_cubes()) or "-"
print(f"{run.slug} {run.event_type:<8} {run.status:<10} sha={sha or '?':<40.12} failed={failed}")
def main() -> None:
if len(sys.argv) < 2 or sys.argv[1] != "runs":
sys.exit("usage: python -m app.sourcecraft.cli runs [org/repo]")
asyncio.run(runs(sys.argv[2] if len(sys.argv) > 2 else None))
if __name__ == "__main__":
main()
+140
View File
@@ -0,0 +1,140 @@
"""SourceCraft REST API client (ARCHITECTURE §2.4). The API drops TLS connections now and then: every call retries.
Runs don't carry a commit sha, so the uploads CI prints it from its first cube (`check/meta`, ADR-014);
`run_commit_sha` reads it back.
"""
import asyncio
import logging
import re
from collections.abc import Awaitable, Callable
from types import TracebackType
from typing import Any, Self
import httpx
from app.config import SourceCraftSettings
from app.sourcecraft.models import CubeLog, Run
log = logging.getLogger(__name__)
META_TASK = "check"
META_CUBE = "meta"
_SHA_LINE = re.compile(r"^TGIT_COMMIT_SHA=([0-9a-f]{40})\s*$", re.MULTILINE)
Sleep = Callable[[float], Awaitable[None]]
class SourceCraftError(Exception):
def __init__(self, status: int, code: str, message: str) -> None:
super().__init__(f"{status} {code}: {message}")
self.status = status
self.code = code
self.message = message
class SourceCraftClient:
def __init__(
self,
settings: SourceCraftSettings,
transport: httpx.AsyncBaseTransport | None = None,
sleep: Sleep = asyncio.sleep,
) -> None:
self._settings = settings
self._sleep = sleep
self._http = httpx.AsyncClient(
base_url=settings.api_url,
headers={"Authorization": f"Bearer {settings.token.get_secret_value()}"},
timeout=settings.timeout,
transport=transport,
)
async def __aenter__(self) -> Self:
return self
async def __aexit__(
self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None
) -> None:
await self.aclose()
async def aclose(self) -> None:
await self._http.aclose()
def run_url(self, repo: str, run: str) -> str:
return self._settings.run_url_template.format(repo=repo, run=run)
async def list_runs(self, repo: str, page_size: int = 20) -> list[Run]:
"""Newest first. One page is enough for the watcher: it only looks for runs of recent pushes."""
data = await self._get(f"/repos/{repo}/cicd/runs", params={"page_size": page_size})
return [Run.model_validate(r) for r in data.get("runs", [])]
async def get_run(self, repo: str, run: str) -> Run:
return Run.model_validate(await self._get(f"/repos/{repo}/cicd/runs/{run}"))
async def get_cube_logs(self, repo: str, run: str, workflow: str, task: str, cube: str) -> CubeLog:
"""Concatenate log pages in order; a page that isn't complete yet is the last readable one."""
parts: list[str] = []
done = False
page = 1
while True:
try:
data = await self._get(f"/repos/{repo}/cicd/logs/{run}/{workflow}/{task}/{cube}", params={"page": page})
except SourceCraftError as e:
if e.status == 404 and page > 1:
break
raise
parts.append(data.get("logs", ""))
done = bool(data.get("done", False))
if not data.get("page_complete", False):
break
page += 1
return CubeLog(text="".join(parts), done=done)
async def run_commit_sha(self, repo: str, run: Run) -> str | None:
"""The commit a run was started for, read from the meta cube's log (None if not printed yet)."""
for workflow in run.workflows:
if run.task(workflow.slug, META_TASK) is None:
continue
try:
logs = await self.get_cube_logs(repo, run.slug, workflow.slug, META_TASK, META_CUBE)
except SourceCraftError as e:
if e.status == 404:
return None
raise
match = _SHA_LINE.search(logs.text)
return match.group(1) if match else None
return None
async def _get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
attempts = self._settings.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
log.warning("sourcecraft %s: %s, retry %d/%d", path, type(e).__name__, attempt, attempts - 1)
else:
if response.status_code < 500:
return self._parse(response)
if attempt == attempts:
return self._parse(response)
log.warning("sourcecraft %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")
@staticmethod
def _parse(response: httpx.Response) -> dict[str, Any]:
try:
body = response.json()
except ValueError:
body = {}
if response.status_code >= 400:
raise SourceCraftError(
response.status_code,
str(body.get("error_code", "")),
str(body.get("message", response.text[:200])),
)
if not isinstance(body, dict):
raise SourceCraftError(response.status_code, "BadResponse", "expected a JSON object")
return body
+113
View File
@@ -0,0 +1,113 @@
"""Typed SourceCraft CI/CD API responses (only the fields the bot uses)."""
from collections.abc import Iterator
from datetime import datetime
from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field, field_validator
class RunStatus(StrEnum):
CREATED = "created"
PREPARED = "prepared"
PROCESSING = "processing"
SUCCESS = "success"
FAILED = "failed"
CANCELED = "canceled"
TIMEOUT = "timeout"
SKIPPED = "skipped"
AWAITING_APPROVAL = "awaiting_approval"
REJECTED = "rejected"
@property
def finished(self) -> bool:
return self in _FINISHED
_FINISHED = {
RunStatus.SUCCESS,
RunStatus.FAILED,
RunStatus.CANCELED,
RunStatus.TIMEOUT,
RunStatus.SKIPPED,
RunStatus.REJECTED,
}
class EventType(StrEnum):
PUSH = "push"
PR_UPDATE = "pr_update"
MANUAL = "manual"
RESTART = "restart"
SCHEDULE = "schedule"
REPOSITORY_EVENT = "repository_event"
class _Model(BaseModel):
model_config = ConfigDict(extra="ignore", frozen=True)
class Dates(_Model):
created_at: datetime | None = None
started_at: datetime | None = None
finished_at: datetime | None = None
@field_validator("created_at", "started_at", "finished_at", mode="after")
@classmethod
def _epoch_is_unset(cls, value: datetime | None) -> datetime | None:
# The API reports "not yet" as 1970-01-01T00:00:00Z
return None if value is not None and value.year == 1970 else value
class Cube(_Model):
slug: str
status: RunStatus
dates: Dates = Field(default_factory=Dates)
@property
def is_service(self) -> bool:
"""Platform steps (#git-clone, #prepare-image-for-…): not in our config, logs not readable."""
return self.slug.startswith("#")
class Task(_Model):
slug: str
status: RunStatus
cubes: list[Cube] = []
class Workflow(_Model):
slug: str
status: RunStatus
tasks: list[Task] = []
class Run(_Model):
slug: str
status: RunStatus
event_type: EventType
dates: Dates = Field(default_factory=Dates)
workflows: list[Workflow] = []
error_messages: list[str] = []
def task(self, workflow: str, task: str) -> Task | None:
for wf in self.workflows:
if wf.slug == workflow:
return next((t for t in wf.tasks if t.slug == task), None)
return None
def cubes(self) -> Iterator[tuple[str, str, Cube]]:
"""(workflow, task, cube) for every user-defined cube."""
for wf in self.workflows:
for t in wf.tasks:
for c in t.cubes:
if not c.is_service:
yield wf.slug, t.slug, c
def failed_cubes(self) -> list[tuple[str, str, Cube]]:
return [entry for entry in self.cubes() if entry[2].status is RunStatus.FAILED]
class CubeLog(_Model):
text: str
done: bool
+3 -3
View File
@@ -70,11 +70,11 @@ aiogram 3. Главное меню — reply-клавиатура (📝 Комм
### 2.4 `app/sourcecraft` — клиент API ### 2.4 `app/sourcecraft` — клиент API
`httpx.AsyncClient`, Bearer-токен, ретраи на сетевые ошибки и 5xx (экспоненциально, до 5 попыток), `httpx.AsyncClient`, Bearer-токен, ретраи на сетевые ошибки и 5xx (экспоненциально, до 5 попыток),
pydantic-модели ответов. Методы: `list_runs`, `get_run`, `get_cube_logs`. Подробности — [`dev/SOURCECRAFT_API.md`](dev/SOURCECRAFT_API.md). 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 ### 2.5 `app/pipeline` — наблюдатель CI/CD
Фоновая задача. Пока есть незавершённые пуши — опрашивает `/cicd/runs` (10 с, бэкофф до 60 с), сопоставляет прогон с Фоновая задача. Пока есть незавершённые пуши — опрашивает `/cicd/runs` (10 с, бэкофф до 60 с), сопоставляет прогон с
пушем по sha. Отслеживает workflow `check` (CI) и `deploy` (CD) отдельно; на смене статуса — редактирует сообщение пушем по sha (из куба `meta`, ADR-014). Отслеживает задачи `check` (CI) и `deploy` (CD) workflow `check-and-deploy` отдельно; на смене статуса — редактирует сообщение
о пуше и шлёт автору уведомление (успех / падение + лог упавшего куба). Падение `deploy` → событие для `alerts`. о пуше и шлёт автору уведомление (успех / падение + лог упавшего куба). Падение `deploy` → событие для `alerts`.
Таймаут — 30 мин. Webhooks у SourceCraft нет (ADR-003). Таймаут — 30 мин. Webhooks у SourceCraft нет (ADR-003).
@@ -138,7 +138,7 @@ SQLAlchemy 2 (async) + Alembic. PostgreSQL в Compose; в тестах — SQLit
| `drafts` | `id`, `user_id` UNIQUE, `base_sha`, `message`, `created_at`, `expires_at` | | `drafts` | `id`, `user_id` UNIQUE, `base_sha`, `message`, `created_at`, `expires_at` |
| `draft_changes` | `id`, `draft_id`, `path`, `op` (`add`/`modify`/`delete`), `content` BYTEA | | `draft_changes` | `id`, `draft_id`, `path`, `op` (`add`/`modify`/`delete`), `content` BYTEA |
| `pushes` | `id`, `user_id`, `sha`, `message`, `paths` JSON, `chat_id`, `message_id`, `ci_status`, `cd_status`, `created_at` | | `pushes` | `id`, `user_id`, `sha`, `message`, `paths` JSON, `chat_id`, `message_id`, `ci_status`, `cd_status`, `created_at` |
| `pipeline_runs` | `id`, `push_id`, `workflow` (`check`/`deploy`), `run_slug`, `status`, `failed_cube`, `url`, `updated_at` | | `pipeline_runs` | `id`, `push_id`, `stage` (`ci`/`cd` — задачи одного workflow `check-and-deploy`), `run_slug`, `status`, `failed_cube`, `url`, `updated_at` |
| `container_snapshots` | `container`, `at`, `restart_count`, `cpu_pct`, `mem_pct`, `health` — для проверок «за 5 мин» (TTL 1 сутки) | | `container_snapshots` | `container`, `at`, `restart_count`, `cpu_pct`, `mem_pct`, `health` — для проверок «за 5 мин» (TTL 1 сутки) |
| `alerts` | `id`, `kind`, `service`, `payload` JSON, `count`, `opened_at`, `last_at`, `resolved_at` | | `alerts` | `id`, `kind`, `service`, `payload` JSON, `count`, `opened_at`, `last_at`, `resolved_at` |
| `alert_deliveries` | `alert_id`, `chat_id`, `message_id` | | `alert_deliveries` | `alert_id`, `chat_id`, `message_id` |
+8
View File
@@ -121,3 +121,11 @@
**Решение:** конвейер из пяти шагов (SPEC §4.3.1): детерминированный фильтр → отдельный LLM-вызов «страж» с JSON-вердиктом `safe|attack` (невалидный ответ = `attack`) → отдельный LLM-вызов «планировщик», который получает только промт и список контейнеров и возвращает `LogQuery` → валидация → выполнение в Python. Пользователю показывается итоговый grep и найденные строки. Атака → отказ + алерт админам. **Решение:** конвейер из пяти шагов (SPEC §4.3.1): детерминированный фильтр → отдельный LLM-вызов «страж» с JSON-вердиктом `safe|attack` (невалидный ответ = `attack`) → отдельный LLM-вызов «планировщик», который получает только промт и список контейнеров и возвращает `LogQuery` → валидация → выполнение в Python. Пользователю показывается итоговый grep и найденные строки. Атака → отказ + алерт админам.
**Альтернативы:** один вызов LLM с инструкцией «откажись, если атака» — страж и исполнитель в одном контексте, injection обходит оба; передавать логи в LLM для поиска — injection из логов и утечка секретов. **Альтернативы:** один вызов LLM с инструкцией «откажись, если атака» — страж и исполнитель в одном контексте, injection обходит оба; передавать логи в LLM для поиска — injection из логов и утечка секретов.
**Последствия:** два LLM-вызова на запрос (латентность, токены → квота 30/сутки); даже при обходе стража вред ограничен схемой `LogQuery`. **Последствия:** два LLM-вызова на запрос (латентность, токены → квота 30/сутки); даже при обходе стража вред ограничен схемой `LogQuery`.
## ADR-014: Связь «прогон ↔ коммит» — через куб `meta`, печатающий sha
**Дата:** 2026-10-14 · **Статус:** принято (уточняет ADR-003)
**Контекст:** бот должен понять, какой прогон CI относится к его пушу. В API SourceCraft у прогона нет sha коммита (ни в списке, ни в деталях), фильтра по коммиту нет; логи служебных кубов (`#git-clone`) API не отдаёт.
**Решение:** первый куб задачи `check` в `tgit-uploads` — `meta`: `echo "TGIT_COMMIT_SHA=$SOURCECRAFT_COMMIT_SHA"`. Бот читает его лог (`SourceCraftClient.run_commit_sha`) и сопоставляет прогон с пушем.
**Альтернативы:** запускать CI из бота вручную (`POST /cicd/runs` с `head.commit`) — номер прогона известен сразу, но push-триггер пришлось бы убрать, и ручные пуши в репозиторий остались бы без CI; сопоставление по времени — ломается при параллельных пушах.
**Последствия:** +1 короткий куб на прогон; куб `meta` нельзя удалять или переименовывать (он в `.sourcecraft/`, который бот защищает от правок); прогоны старше этого решения sha не имеют.
+6
View File
@@ -79,6 +79,12 @@ curl -s -X PUT -H "$H" -H "$J" "$R/secrets/DEPLOY_HOST" -d "{\"value\":\"$(print
- Прогон: `slug` — числовой (`1000000`…), `status`: `created` → `processing` → `success`/`failed`; `event_type`: - Прогон: `slug` — числовой (`1000000`…), `status`: `created` → `processing` → `success`/`failed`; `event_type`:
`push`/`manual`/…. У кубов служебные шаги со `slug`, начинающимся на `#` (`#git-clone`, `#prepare-image-for-<cube>`). `push`/`manual`/…. У кубов служебные шаги со `slug`, начинающимся на `#` (`#git-clone`, `#prepare-image-for-<cube>`).
- Если куб упал, остальные кубы задачи остаются в `created` и не запускаются. - Если куб упал, остальные кубы задачи остаются в `created` и не запускаются.
- **В прогоне нет sha коммита** (ни в списке, ни в деталях), фильтра по коммиту нет → в `tgit-uploads` куб `check/meta` печатает `TGIT_COMMIT_SHA=…`, бот читает его лог (ADR-014).
- Логи служебных кубов (`#git-clone` и т.п.) API не отдаёт: `400 CITaskName.Pattern`.
- Логи постранично: `?page=N` с 1; читать по порядку, пока `page_complete: true`; следующая страница незаконченной — 404.
- Статусы прогона: `created`, `prepared`, `processing`, `success`, `failed`, `canceled`, `timeout`, `skipped`, `awaiting_approval`, `rejected`. Типы события: `push`, `pr_update`, `manual`, `restart`, `schedule`, `repository_event`.
- Очередь бесплатных раннеров бывает минутами: прогон может стоять в `prepared`/`created` до ~6 мин до старта.
- Веб-интерфейс — SPA, на любой путь отвечает 200: формат ссылки на прогон снаружи не проверить (`SOURCECRAFT_RUN_URL_TEMPLATE`).
- Код в кубе склонирован в `/sourcecraft/workspace` (`$SOURCECRAFT_WORKSPACE`) — это рабочий каталог. Переменные: - Код в кубе склонирован в `/sourcecraft/workspace` (`$SOURCECRAFT_WORKSPACE`) — это рабочий каталог. Переменные:
`SOURCECRAFT_COMMIT_SHA`, `SOURCECRAFT_COMMIT_REF_NAME` и др. `SOURCECRAFT_COMMIT_SHA`, `SOURCECRAFT_COMMIT_REF_NAME` и др.
- Исходящих webhooks нет — статус CI получаем опросом (ADR-003). - Исходящих webhooks нет — статус CI получаем опросом (ADR-003).
@@ -13,7 +13,13 @@ workflows:
tasks: tasks:
- name: check - name: check
cubes: cubes:
# The API doesn't tell which commit a run is for — the bot reads it from this log (ADR-014). Keep it first.
- name: meta
image: ghcr.io/astral-sh/uv:0.8-python3.12-bookworm-slim
script:
- echo "TGIT_COMMIT_SHA=$SOURCECRAFT_COMMIT_SHA"
- name: compile - name: compile
needs: ["-"]
image: ghcr.io/astral-sh/uv:0.8-python3.12-bookworm-slim image: ghcr.io/astral-sh/uv:0.8-python3.12-bookworm-slim
script: script:
- python -m compileall -q -x '(^|/)\.' . - python -m compileall -q -x '(^|/)\.' .
+1 -1
View File
@@ -63,4 +63,4 @@ omit = ["app/__main__.py"]
show_missing = true show_missing = true
skip_covered = true skip_covered = true
# Ratchet: raise as code lands, never lower. Target by M4 — 70 (SPEC §5.2). # Ratchet: raise as code lands, never lower. Target by M4 — 70 (SPEC §5.2).
fail_under = 45 fail_under = 80
+3 -2
View File
@@ -4,6 +4,7 @@ from aiogram.types import InlineKeyboardMarkup, ReplyKeyboardMarkup
from app.bot.menu import Nav, NavAction from app.bot.menu import Nav, NavAction
from app.config import Settings from app.config import Settings
from app.db.session import SessionMaker
from app.main import build_dispatcher from app.main import build_dispatcher
from app.texts import Texts from app.texts import Texts
from tests.conftest import ADMIN, OUTSIDER from tests.conftest import ADMIN, OUTSIDER
@@ -11,8 +12,8 @@ from tests.fakes import TgClient
@pytest.fixture @pytest.fixture
def tg(settings: Settings, texts: Texts) -> TgClient: def tg(settings: Settings, texts: Texts, sessions: SessionMaker) -> TgClient:
return TgClient(build_dispatcher(settings, texts)) return TgClient(build_dispatcher(settings, texts, sessions))
def buttons(markup: ReplyKeyboardMarkup) -> list[str]: def buttons(markup: ReplyKeyboardMarkup) -> list[str]:
+48
View File
@@ -0,0 +1,48 @@
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.storage.base import StorageKey
from sqlalchemy import func, select
from app.bot.fsm_storage import SqlStorage, key_to_str
from app.db.models import FsmState
from app.db.session import SessionMaker
KEY = StorageKey(bot_id=42, chat_id=7, user_id=7)
class Flow(StatesGroup):
name = State()
async def test_state_and_data_survive_a_new_storage(sessions: SessionMaker) -> None:
first = SqlStorage(sessions)
await first.set_state(KEY, Flow.name)
await first.set_data(KEY, {"path": "services/a", "page": 2})
# A fresh storage on the same DB = the bot after a restart
second = SqlStorage(sessions)
assert await second.get_state(KEY) == Flow.name.state
assert await second.get_data(KEY) == {"path": "services/a", "page": 2}
async def test_unknown_key_is_empty(sessions: SessionMaker) -> None:
storage = SqlStorage(sessions)
assert await storage.get_state(KEY) is None
assert await storage.get_data(KEY) == {}
async def test_cleared_state_removes_row(sessions: SessionMaker) -> None:
storage = SqlStorage(sessions)
await storage.set_state(KEY, "x")
await storage.set_data(KEY, {"a": 1})
await storage.set_state(KEY, None)
await storage.set_data(KEY, {})
await storage.close()
async with sessions() as s:
assert await s.scalar(select(func.count()).select_from(FsmState)) == 0
def test_key_includes_every_part() -> None:
key = StorageKey(bot_id=1, chat_id=2, user_id=3, thread_id=4, business_connection_id="b", destiny="d")
assert key_to_str(key) == "1:2:3:4:b:d"
assert key_to_str(KEY) == "42:7:7:::default"
+18
View File
@@ -1,8 +1,12 @@
from collections.abc import AsyncIterator
from pathlib import Path from pathlib import Path
import pytest import pytest
from sqlalchemy.ext.asyncio import AsyncEngine
from app.config import AccessSettings, RuntimeSettings, Settings, TelegramSettings from app.config import AccessSettings, RuntimeSettings, Settings, TelegramSettings
from app.db.migrate import upgrade
from app.db.session import SessionMaker, make_engine, make_sessionmaker
from app.texts import Texts, load_texts from app.texts import Texts, load_texts
ROOT = Path(__file__).resolve().parent.parent ROOT = Path(__file__).resolve().parent.parent
@@ -22,3 +26,17 @@ def settings(tmp_path: Path) -> Settings:
access=AccessSettings(admin_ids=frozenset({ADMIN})), access=AccessSettings(admin_ids=frozenset({ADMIN})),
runtime=RuntimeSettings(config_dir=ROOT / "config", heartbeat_file=tmp_path / "hb"), runtime=RuntimeSettings(config_dir=ROOT / "config", heartbeat_file=tmp_path / "hb"),
) )
@pytest.fixture
async def engine() -> AsyncIterator[AsyncEngine]:
"""Fresh in-memory SQLite with all migrations applied — the same path the bot takes at startup."""
eng = make_engine("sqlite+aiosqlite://")
await upgrade(eng)
yield eng
await eng.dispose()
@pytest.fixture
def sessions(engine: AsyncEngine) -> SessionMaker:
return make_sessionmaker(engine)
View File
+30
View File
@@ -0,0 +1,30 @@
from datetime import timedelta
from sqlalchemy import func, select
from app.db.maintenance import cleanup_expired
from app.db.models import AccessAttempt, ContainerSnapshot, Draft, User, utcnow
from app.db.session import SessionMaker
async def test_cleanup_removes_only_expired(sessions: SessionMaker) -> None:
now = utcnow()
async with sessions.begin() as s:
s.add_all([User(tg_id=1, full_name="a"), User(tg_id=2, full_name="b")])
await s.flush() # no relationship() to order inserts: users first for the FK
s.add_all(
[
Draft(user_id=1, base_sha="a" * 40, expires_at=now - timedelta(minutes=1)),
Draft(user_id=2, base_sha="b" * 40, expires_at=now + timedelta(hours=1)),
ContainerSnapshot(container="c", at=now - timedelta(days=2), restart_count=0, cpu_pct=0, mem_pct=0),
ContainerSnapshot(container="c", at=now, restart_count=0, cpu_pct=0, mem_pct=0),
AccessAttempt(tg_id=9, at=now - timedelta(days=2), ok=False),
]
)
removed = await cleanup_expired(sessions, now)
assert removed == {"drafts": 1, "snapshots": 1, "access_attempts": 1}
async with sessions() as s:
assert await s.scalar(select(func.count()).select_from(Draft)) == 1
assert await s.scalar(select(func.count()).select_from(ContainerSnapshot)) == 1
+82
View File
@@ -0,0 +1,82 @@
from datetime import timedelta
import pytest
from alembic.autogenerate import compare_metadata
from alembic.migration import MigrationContext
from sqlalchemy import inspect, text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncEngine
from app.db.migrate import downgrade, upgrade
from app.db.models import Base, ChangeOp, Draft, DraftChange, User, utcnow
from app.db.session import SessionMaker, make_engine
TABLES = {
"users",
"invites",
"access_attempts",
"fsm_state",
"drafts",
"draft_changes",
"pushes",
"pipeline_runs",
"container_snapshots",
"alerts",
"alert_deliveries",
"log_queries",
"agent_requests",
}
async def table_names(engine: AsyncEngine) -> set[str]:
async with engine.connect() as conn:
return set(await conn.run_sync(lambda c: inspect(c).get_table_names())) - {"alembic_version"}
async def test_upgrade_creates_all_tables(engine: AsyncEngine) -> None:
assert await table_names(engine) == TABLES
async def test_models_match_migrations(engine: AsyncEngine) -> None:
# A model change without a migration shows up here as a non-empty diff.
async with engine.connect() as conn:
diff = await conn.run_sync(lambda c: compare_metadata(MigrationContext.configure(c), Base.metadata))
assert diff == []
async def test_downgrade_to_base_and_back() -> None:
eng = make_engine("sqlite+aiosqlite://")
try:
await upgrade(eng)
await downgrade(eng, "base")
assert await table_names(eng) == set()
await upgrade(eng)
assert await table_names(eng) == TABLES
finally:
await eng.dispose()
INSERT_USER = text(
"INSERT INTO users (tg_id, full_name, role, is_active, joined_at) VALUES (:id, 'u', :role, 1, '2026-01-01')"
)
async def test_enum_check_rejects_garbage(engine: AsyncEngine) -> None:
async with engine.begin() as conn:
await conn.execute(INSERT_USER, {"id": 1, "role": "admin"})
with pytest.raises(IntegrityError):
async with engine.begin() as conn:
await conn.execute(INSERT_USER, {"id": 2, "role": "root"})
async def test_deleting_draft_cascades_to_changes(sessions: SessionMaker) -> None:
async with sessions.begin() as s:
s.add(User(tg_id=1, full_name="a"))
await s.flush()
draft = Draft(user_id=1, base_sha="0" * 40, expires_at=utcnow() + timedelta(hours=1))
draft.changes.append(DraftChange(path="services/a/main.py", op=ChangeOp.ADD, content=b"print(1)"))
s.add(draft)
async with sessions.begin() as s:
await s.execute(text("DELETE FROM drafts"))
async with sessions() as s:
assert (await s.execute(text("SELECT count(*) FROM draft_changes"))).scalar_one() == 0
+5
View File
@@ -0,0 +1,5 @@
{
"logs": "Downloading ruff (12.8MiB)\n Downloading ruff\nInstalled 1 package in 1ms\nF401 [*] `os` imported but unused\n --> services/broken/main.py:1:8\n |\n1 | import os\n | ^^\n2 |\n3 | def run():\n |\nhelp: Remove unused import: `os`\n\nF821 Undefined name `undefined_name`\n --> services/broken/main.py:4:12\n |\n3 | def run():\n4 | return undefined_name\n | ^^^^^^^^^^^^^^\n |\n\nFound 2 errors.\n[*] 1 fixable with the `--fix` option.\n",
"page_complete": false,
"done": true
}
+5
View File
@@ -0,0 +1,5 @@
{
"logs": "TGIT_COMMIT_SHA=932f4b1a144cfed4fed8eafd94433e0995e5b12d\n",
"page_complete": false,
"done": true
}
+396
View File
@@ -0,0 +1,396 @@
{
"id": "",
"slug": "1000001",
"dates": {
"created_at": "2026-10-13T20:58:52.273658Z",
"started_at": "2026-10-13T20:58:53.940173Z",
"finished_at": "2026-10-13T20:59:17.833911Z",
"updated_at": "2026-10-13T20:59:17.867686Z"
},
"status": "failed",
"workflows": [
{
"id": "",
"slug": "check-and-deploy",
"description": "",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "2026-10-13T20:58:53.940173Z",
"finished_at": "2026-10-13T20:59:17.833911Z",
"updated_at": "2026-10-13T20:59:17.862572Z"
},
"status": "failed",
"tasks": [
{
"id": "",
"slug": "check",
"description": "",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "2026-10-13T20:58:53.940173Z",
"finished_at": "2026-10-13T20:59:11.150164Z",
"updated_at": "2026-10-13T20:59:12.846332Z"
},
"status": "failed",
"cubes": [
{
"id": "",
"slug": "#create-environment",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "2026-10-13T20:58:54.031424Z",
"finished_at": "2026-10-13T20:58:54.090132Z",
"updated_at": "2026-10-13T20:58:54.155008Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": []
}
},
{
"id": "",
"slug": "#git-clone",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "2026-10-13T20:58:54.090481Z",
"finished_at": "2026-10-13T20:58:55.142644Z",
"updated_at": "2026-10-13T20:58:55.211046Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#create-environment"
}
]
}
},
{
"id": "",
"slug": "#prepare-image-for-compile",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "2026-10-13T20:58:55.143033Z",
"finished_at": "2026-10-13T20:58:59.605194Z",
"updated_at": "2026-10-13T20:58:59.706155Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#git-clone"
}
]
}
},
{
"id": "",
"slug": "#prepare-image-for-lint",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "2026-10-13T20:58:59.605538Z",
"finished_at": "2026-10-13T20:59:00.536788Z",
"updated_at": "2026-10-13T20:59:00.604498Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#git-clone"
}
]
}
},
{
"id": "",
"slug": "#prepare-image-for-security",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "2026-10-13T20:59:00.537171Z",
"finished_at": "2026-10-13T20:59:01.447150Z",
"updated_at": "2026-10-13T20:59:01.511994Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#git-clone"
}
]
}
},
{
"id": "",
"slug": "#prepare-image-for-tests",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "2026-10-13T20:59:01.447481Z",
"finished_at": "2026-10-13T20:59:02.624921Z",
"updated_at": "2026-10-13T20:59:02.700313Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#git-clone"
}
]
}
},
{
"id": "",
"slug": "compile",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "2026-10-13T20:59:02.625267Z",
"finished_at": "2026-10-13T20:59:09.366621Z",
"updated_at": "2026-10-13T20:59:09.454706Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#prepare-image-for-compile"
}
]
}
},
{
"id": "",
"slug": "lint",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "2026-10-13T20:59:09.366953Z",
"finished_at": "2026-10-13T20:59:11.088201Z",
"updated_at": "2026-10-13T20:59:11.173518Z"
},
"status": "failed",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#prepare-image-for-lint"
}
]
}
},
{
"id": "",
"slug": "security",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "2026-10-13T20:59:11.088571Z",
"updated_at": "2026-10-13T20:59:11.206851Z"
},
"status": "created",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#prepare-image-for-security"
}
]
}
},
{
"id": "",
"slug": "tests",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "2026-10-13T20:59:11.088804Z",
"updated_at": "2026-10-13T20:59:11.242177Z"
},
"status": "created",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#prepare-image-for-tests"
}
]
}
},
{
"id": "",
"slug": "#tear-down-environment",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "2026-10-13T20:59:11.089254Z",
"finished_at": "2026-10-13T20:59:11.150164Z",
"updated_at": "2026-10-13T20:59:11.346169Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "compile"
},
{
"name": "lint"
},
{
"name": "security"
},
{
"name": "tests"
}
]
}
}
],
"relations": {
"needs": []
}
},
{
"id": "",
"slug": "deploy",
"description": "",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "2026-10-13T20:59:17.833911Z",
"updated_at": "2026-10-13T20:59:17.840273Z"
},
"status": "created",
"cubes": [
{
"id": "",
"slug": "#create-environment",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-13T20:58:52.727666Z"
},
"status": "created",
"artifacts": [],
"relations": {
"needs": []
}
},
{
"id": "",
"slug": "#git-clone",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-13T20:58:52.727666Z"
},
"status": "created",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#create-environment"
}
]
}
},
{
"id": "",
"slug": "#prepare-artifacts",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-13T20:58:52.727666Z"
},
"status": "created",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#git-clone"
}
]
}
},
{
"id": "",
"slug": "#prepare-image-for-deploy",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-13T20:58:52.727666Z"
},
"status": "created",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#prepare-artifacts"
}
]
}
},
{
"id": "",
"slug": "deploy",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-13T20:58:52.727666Z"
},
"status": "created",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#prepare-image-for-deploy"
}
]
}
},
{
"id": "",
"slug": "#tear-down-environment",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-13T20:58:52.727666Z"
},
"status": "created",
"artifacts": [],
"relations": {
"needs": [
{
"name": "deploy"
}
]
}
}
],
"relations": {
"needs": [
{
"name": "check"
}
]
}
}
],
"environment_id": ""
}
],
"event_type": "manual",
"error_messages": [],
"user": {
"id": "01a06158-1562-7db1-b6e9-018ddd4d4c29",
"slug": "rogachevegorushka"
}
}
+437
View File
@@ -0,0 +1,437 @@
{
"id": "",
"slug": "1000002",
"dates": {
"created_at": "2026-10-14T14:51:56.652199Z",
"started_at": "2026-10-14T14:57:43.223468Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-14T14:57:43.242955Z"
},
"status": "processing",
"workflows": [
{
"id": "",
"slug": "check-and-deploy",
"description": "",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:57:43.223468Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-14T14:57:43.234813Z"
},
"status": "processing",
"tasks": [
{
"id": "",
"slug": "check",
"description": "",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:57:43.223468Z",
"finished_at": "2026-10-14T14:58:19.375134Z",
"updated_at": "2026-10-14T14:58:19.571691Z"
},
"status": "success",
"cubes": [
{
"id": "",
"slug": "#create-environment",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:57:43.514526Z",
"finished_at": "2026-10-14T14:57:43.584723Z",
"updated_at": "2026-10-14T14:57:43.657871Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": []
}
},
{
"id": "",
"slug": "#git-clone",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:57:43.585127Z",
"finished_at": "2026-10-14T14:57:45.528681Z",
"updated_at": "2026-10-14T14:57:45.606302Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#create-environment"
}
]
}
},
{
"id": "",
"slug": "#prepare-image-for-compile",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:57:45.529064Z",
"finished_at": "2026-10-14T14:57:55.435126Z",
"updated_at": "2026-10-14T14:57:55.537030Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#git-clone"
}
]
}
},
{
"id": "",
"slug": "#prepare-image-for-lint",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:57:55.435513Z",
"finished_at": "2026-10-14T14:57:58.411834Z",
"updated_at": "2026-10-14T14:57:58.479777Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#git-clone"
}
]
}
},
{
"id": "",
"slug": "#prepare-image-for-meta",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:57:58.412252Z",
"finished_at": "2026-10-14T14:58:03.323469Z",
"updated_at": "2026-10-14T14:58:03.420479Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#git-clone"
}
]
}
},
{
"id": "",
"slug": "#prepare-image-for-security",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:58:03.323798Z",
"finished_at": "2026-10-14T14:58:04.898405Z",
"updated_at": "2026-10-14T14:58:04.986588Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#git-clone"
}
]
}
},
{
"id": "",
"slug": "#prepare-image-for-tests",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:58:04.898847Z",
"finished_at": "2026-10-14T14:58:09.732911Z",
"updated_at": "2026-10-14T14:58:09.807301Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#git-clone"
}
]
}
},
{
"id": "",
"slug": "compile",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:58:09.733227Z",
"finished_at": "2026-10-14T14:58:12.400619Z",
"updated_at": "2026-10-14T14:58:12.533056Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#prepare-image-for-compile"
}
]
}
},
{
"id": "",
"slug": "lint",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:58:12.401008Z",
"finished_at": "2026-10-14T14:58:14.235114Z",
"updated_at": "2026-10-14T14:58:14.304352Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#prepare-image-for-lint"
}
]
}
},
{
"id": "",
"slug": "meta",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:58:14.235485Z",
"finished_at": "2026-10-14T14:58:14.707956Z",
"updated_at": "2026-10-14T14:58:14.778120Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#prepare-image-for-meta"
}
]
}
},
{
"id": "",
"slug": "security",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:58:14.708362Z",
"finished_at": "2026-10-14T14:58:17.148819Z",
"updated_at": "2026-10-14T14:58:17.221280Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#prepare-image-for-security"
}
]
}
},
{
"id": "",
"slug": "tests",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:58:17.149213Z",
"finished_at": "2026-10-14T14:58:19.298832Z",
"updated_at": "2026-10-14T14:58:19.405984Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#prepare-image-for-tests"
}
]
}
},
{
"id": "",
"slug": "#tear-down-environment",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:58:19.299366Z",
"finished_at": "2026-10-14T14:58:19.375117Z",
"updated_at": "2026-10-14T14:58:19.528162Z"
},
"status": "success",
"artifacts": [],
"relations": {
"needs": [
{
"name": "compile"
},
{
"name": "lint"
},
{
"name": "meta"
},
{
"name": "security"
},
{
"name": "tests"
}
]
}
}
],
"relations": {
"needs": []
}
},
{
"id": "",
"slug": "deploy",
"description": "",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-14T14:58:21.737007Z"
},
"status": "created",
"cubes": [
{
"id": "",
"slug": "#create-environment",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-14T14:51:57.522605Z"
},
"status": "created",
"artifacts": [],
"relations": {
"needs": []
}
},
{
"id": "",
"slug": "#git-clone",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-14T14:51:57.522605Z"
},
"status": "created",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#create-environment"
}
]
}
},
{
"id": "",
"slug": "#prepare-artifacts",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-14T14:51:57.522605Z"
},
"status": "created",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#git-clone"
}
]
}
},
{
"id": "",
"slug": "#prepare-image-for-deploy",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-14T14:51:57.522605Z"
},
"status": "created",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#prepare-artifacts"
}
]
}
},
{
"id": "",
"slug": "deploy",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-14T14:51:57.522605Z"
},
"status": "created",
"artifacts": [],
"relations": {
"needs": [
{
"name": "#prepare-image-for-deploy"
}
]
}
},
{
"id": "",
"slug": "#tear-down-environment",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "1970-01-01T00:00:00Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-14T14:51:57.522605Z"
},
"status": "created",
"artifacts": [],
"relations": {
"needs": [
{
"name": "deploy"
}
]
}
}
],
"relations": {
"needs": [
{
"name": "check"
}
]
}
}
],
"environment_id": ""
}
],
"event_type": "push",
"error_messages": [],
"user": {
"id": "01a06158-1562-7db1-b6e9-018ddd4d4c29",
"slug": "rogachevegorushka"
}
}
+104
View File
@@ -0,0 +1,104 @@
{
"runs": [
{
"id": "",
"slug": "1000002",
"dates": {
"created_at": "2026-10-14T14:51:56.652199Z",
"started_at": "2026-10-14T14:57:43.223468Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-14T14:57:43.242955Z"
},
"status": "processing",
"workflows": [
{
"id": "",
"slug": "check-and-deploy",
"description": "",
"dates": {
"created_at": "2026-10-14T14:51:57.521264Z",
"started_at": "2026-10-14T14:57:43.223468Z",
"finished_at": "1970-01-01T00:00:00Z",
"updated_at": "2026-10-14T14:57:43.234813Z"
},
"status": "processing",
"tasks": [],
"environment_id": ""
}
],
"event_type": "push",
"error_messages": [],
"user": {
"id": "01a06158-1562-7db1-b6e9-018ddd4d4c29",
"slug": "rogachevegorushka"
}
},
{
"id": "",
"slug": "1000001",
"dates": {
"created_at": "2026-10-13T20:58:52.273658Z",
"started_at": "2026-10-13T20:58:53.940173Z",
"finished_at": "2026-10-13T20:59:17.833911Z",
"updated_at": "2026-10-13T20:59:17.867686Z"
},
"status": "failed",
"workflows": [
{
"id": "",
"slug": "check-and-deploy",
"description": "",
"dates": {
"created_at": "2026-10-13T20:58:52.726601Z",
"started_at": "2026-10-13T20:58:53.940173Z",
"finished_at": "2026-10-13T20:59:17.833911Z",
"updated_at": "2026-10-13T20:59:17.862572Z"
},
"status": "failed",
"tasks": [],
"environment_id": ""
}
],
"event_type": "manual",
"error_messages": [],
"user": {
"id": "01a06158-1562-7db1-b6e9-018ddd4d4c29",
"slug": "rogachevegorushka"
}
},
{
"id": "",
"slug": "1000000",
"dates": {
"created_at": "2026-10-13T20:57:06.596315Z",
"started_at": "2026-10-13T20:57:06.821003Z",
"finished_at": "2026-10-13T20:58:26.698635Z",
"updated_at": "2026-10-13T20:58:27.884131Z"
},
"status": "success",
"workflows": [
{
"id": "",
"slug": "check-and-deploy",
"description": "",
"dates": {
"created_at": "2026-10-13T20:57:06.725971Z",
"started_at": "2026-10-13T20:57:06.821003Z",
"finished_at": "2026-10-13T20:58:26.698635Z",
"updated_at": "2026-10-13T20:58:27.878032Z"
},
"status": "success",
"tasks": [],
"environment_id": ""
}
],
"event_type": "push",
"error_messages": [],
"user": {
"id": "01a06158-1562-7db1-b6e9-018ddd4d4c29",
"slug": "rogachevegorushka"
}
}
],
"next_page_token": "147400973399132381"
}
View File
+167
View File
@@ -0,0 +1,167 @@
import json
from pathlib import Path
from typing import Any
import httpx
import pytest
import respx
from app.config import SourceCraftSettings
from app.sourcecraft.client import SourceCraftClient, SourceCraftError
from app.sourcecraft.models import EventType, RunStatus
FIXTURES = Path(__file__).resolve().parent.parent / "fixtures" / "sourcecraft"
API = "https://api.example.test"
REPO = "org/repo"
SHA = "932f4b1a144cfed4fed8eafd94433e0995e5b12d"
def fixture(name: str) -> Any:
return json.loads((FIXTURES / name).read_text(encoding="utf-8"))
class Sleeps:
def __init__(self) -> None:
self.delays: list[float] = []
async def __call__(self, delay: float) -> None:
self.delays.append(delay)
@pytest.fixture
def sleeps() -> Sleeps:
return Sleeps()
@pytest.fixture
async def client(sleeps: Sleeps) -> Any:
settings = SourceCraftSettings(token="secret", api_url=API, max_attempts=3) # type: ignore[arg-type]
async with SourceCraftClient(settings, sleep=sleeps) as c:
yield c
@respx.mock
async def test_list_runs_parses_real_response(client: SourceCraftClient) -> None:
route = respx.get(f"{API}/repos/{REPO}/cicd/runs").respond(json=fixture("runs.json"))
runs = await client.list_runs(REPO, page_size=3)
assert route.calls.last.request.headers["Authorization"] == "Bearer secret"
assert route.calls.last.request.url.params["page_size"] == "3"
assert [r.slug for r in runs] == ["1000002", "1000001", "1000000"]
assert runs[1].event_type is EventType.MANUAL
@respx.mock
async def test_failed_run_points_at_failed_cube(client: SourceCraftClient) -> None:
respx.get(f"{API}/repos/{REPO}/cicd/runs/1000001").respond(json=fixture("run_failed.json"))
run = await client.get_run(REPO, "1000001")
assert run.status is RunStatus.FAILED and run.status.finished
assert [(t, c.slug) for _, t, c in run.failed_cubes()] == [("check", "lint")]
assert all(not c.is_service for _, _, c in run.cubes())
assert run.dates.finished_at is not None
@respx.mock
async def test_unfinished_run_has_no_finish_date(client: SourceCraftClient) -> None:
respx.get(f"{API}/repos/{REPO}/cicd/runs/1000002").respond(json=fixture("run_processing.json"))
run = await client.get_run(REPO, "1000002")
assert not run.status.finished
assert run.dates.finished_at is None # API sends 1970-01-01 for "not yet"
check = run.task("check-and-deploy", "check")
assert check is not None and check.status is RunStatus.SUCCESS
assert run.task("check-and-deploy", "nope") is None
@respx.mock
async def test_commit_sha_from_meta_cube(client: SourceCraftClient) -> None:
respx.get(f"{API}/repos/{REPO}/cicd/runs/1000002").respond(json=fixture("run_processing.json"))
respx.get(f"{API}/repos/{REPO}/cicd/logs/1000002/check-and-deploy/check/meta").respond(
json=fixture("log_meta.json")
)
run = await client.get_run(REPO, "1000002")
assert await client.run_commit_sha(REPO, run) == SHA
@respx.mock
async def test_commit_sha_none_when_meta_log_missing(client: SourceCraftClient) -> None:
respx.get(f"{API}/repos/{REPO}/cicd/runs/1000002").respond(json=fixture("run_processing.json"))
respx.get(f"{API}/repos/{REPO}/cicd/logs/1000002/check-and-deploy/check/meta").respond(
404, json={"error_code": "NotFound", "message": "no logs"}
)
run = await client.get_run(REPO, "1000002")
assert await client.run_commit_sha(REPO, run) is None
@respx.mock
async def test_logs_are_read_page_by_page(client: SourceCraftClient) -> None:
url = f"{API}/repos/{REPO}/cicd/logs/1/wf/task/cube"
respx.get(url, params={"page": "1"}).respond(json={"logs": "a\n", "page_complete": True, "done": False})
respx.get(url, params={"page": "2"}).respond(json={"logs": "b\n", "page_complete": True, "done": True})
respx.get(url, params={"page": "3"}).respond(404, json={"error_code": "NotFound", "message": ""})
logs = await client.get_cube_logs(REPO, "1", "wf", "task", "cube")
assert logs.text == "a\nb\n" and logs.done
@respx.mock
async def test_real_failed_lint_log(client: SourceCraftClient) -> None:
respx.get(f"{API}/repos/{REPO}/cicd/logs/1000001/check-and-deploy/check/lint").respond(
json=fixture("log_lint_failed.json")
)
logs = await client.get_cube_logs(REPO, "1000001", "check-and-deploy", "check", "lint")
assert "F821 Undefined name" in logs.text and logs.done
@respx.mock
async def test_retries_transport_errors_and_5xx(client: SourceCraftClient, sleeps: Sleeps) -> None:
respx.get(f"{API}/repos/{REPO}/cicd/runs").mock(
side_effect=[httpx.ConnectError("tls eof"), httpx.Response(503), httpx.Response(200, json={"runs": []})]
)
assert await client.list_runs(REPO) == []
assert sleeps.delays == [0.5, 1.0]
@respx.mock
async def test_gives_up_after_max_attempts(client: SourceCraftClient) -> None:
respx.get(f"{API}/repos/{REPO}/cicd/runs").mock(side_effect=httpx.ConnectError("down"))
with pytest.raises(httpx.ConnectError):
await client.list_runs(REPO)
@respx.mock
async def test_persistent_5xx_raises_api_error(client: SourceCraftClient) -> None:
respx.get(f"{API}/repos/{REPO}/cicd/runs").respond(502, text="bad gateway")
with pytest.raises(SourceCraftError) as exc:
await client.list_runs(REPO)
assert exc.value.status == 502
@respx.mock
async def test_4xx_is_not_retried(client: SourceCraftClient, sleeps: Sleeps) -> None:
route = respx.get(f"{API}/repos/{REPO}/cicd/runs/9").respond(
403, json={"error_code": "PermissionDenied", "message": "no access"}
)
with pytest.raises(SourceCraftError) as exc:
await client.get_run(REPO, "9")
assert (exc.value.status, exc.value.code) == (403, "PermissionDenied")
assert route.call_count == 1 and sleeps.delays == []
async def test_run_url(client: SourceCraftClient) -> None:
assert client.run_url(REPO, "7") == "https://sourcecraft.dev/org/repo/cicd/runs/7"
+3 -2
View File
@@ -23,8 +23,9 @@ def test_age_decides(tmp_path: Path) -> None:
async def test_heartbeat_touches_file(tmp_path: Path) -> None: async def test_heartbeat_touches_file(tmp_path: Path) -> None:
hb = tmp_path / "hb" hb = tmp_path / "hb"
task = asyncio.create_task(health.heartbeat(hb, interval=10)) task = asyncio.create_task(health.heartbeat(hb, interval=10))
await asyncio.sleep(0) async with asyncio.timeout(2):
assert hb.exists() while not hb.exists(): # noqa: ASYNC110 - polling the filesystem; touch runs in a worker thread
await asyncio.sleep(0.01)
task.cancel() task.cancel()
with contextlib.suppress(asyncio.CancelledError): with contextlib.suppress(asyncio.CancelledError):
await task await task