Contracts / Контракты

Типизированные схемы для REST API и Redis очередей.

Design Principles

Schema-first — все сообщения валидируются Pydantic схемами
1:1 Queues — одна очередь = один Writer → один Consumer (+ optional observers)
Logical Actors — указываем роль (PO ReactAgent, Developer-Worker, langgraph), не техническую прослойку
Traceable — correlation_id для сквозной трассировки

Canonical vocabularies (`shared/contracts/vocab.py`)

One StrEnum per cross-service concept. Producers and consumers import these instead of restating a Literal[...] or a local enum:

Enum	Values	Used by
`AgentType`	`claude`, `factory`, `codex`, `noop`	`WorkerConfig.agent_type`, `AgentConfigDTO.type`, worker-manager/worker-wrapper agent branching (re-exported from `queues.worker`)
`ActionType`	`create`, `feature`, `fix`	`EngineeringMessage.action`
`ResultStatus`	`success`, `failed`, `timeout`	`BaseResult.status` (and its subclasses)
`LifecycleEvent`	`started`, `progress`, `completed`, `failed`, `stopped` (canonical member set)	via the field-specific subsets below

LifecycleEvent is the canonical member set, but each wire accepts only the slice its producers emit — the subsets are Literal[...] over the enum members, kept explicit so the historical per-field vocabularies are not merged:

TaskProgressKind (started/progress/completed/failed, no stopped) — ProgressEvent.type, WorkerEvent.event_type.

Other vocabularies stay deliberately separate — they carry values the canonical enums do not, so merging them would broaden a field past what the wire supports:

DeployAction (create/feature/fix plus stop/undeploy) — deploy operations, a superset of ActionType. TaskType (plus refactor) — the planning-layer task kind.
WorkerCliKind (droid/claude_code/codex) is the CLI's self-reported wire identity on worker:events. It remains a separate concept from AgentType; only the codex spelling currently overlaps.

WorkerConfig.host_codex_home adds no new queue shape beyond an optional field. For agent_type=codex and auth_mode=host_session, worker-manager validates a dedicated file-backed ChatGPT profile before image resolution. It then mounts that host directory read-write at /home/worker/.codex and sets CODEX_HOME to the same path. The profile must contain access and refresh tokens so the non-interactive worker can refresh its session. Unknown agent values fail Pydantic validation or explicit LangGraph/image-routing checks; there is no fallback to Claude.

ResultStatus dropped the old error failure synonym: a failed result is failed, never error. The provisioner:results consumer treats a message that fails validation as terminal (logs it and ACKs) so a stale/invalid entry cannot poison-loop the reclaim.

Queue Registry

Complete list of all Redis Streams / Queues.

Source of Truth: shared/queues.py (QUEUE_TOPOLOGY)

Scaffolding

Queue	Group	DTO	Initiator	Consumer	Purpose
`scaffold:queue`	`scaffold-consumers`	ScaffoldMessage	Task Dispatcher (scheduler)	scaffolder	Prepare repo: copier + make setup + git push

Architect Pipeline

Queue	Group	DTO	Initiator	Consumer	Purpose
`architect:queue`	`architect-consumers`	ArchitectMessage	PO ReactAgent	scheduler	Story → tasks LLM decomposition

Engineering Flows

Queue	Group	DTO	Initiator	Consumer	Purpose
`engineering:queue`	`capability-workers`	EngineeringMessage	Task Dispatcher (scheduler)	langgraph	Start development task
`deploy:queue`	`capability-workers`	DeployMessage	Task Dispatcher (scheduler) / PO	langgraph	Start deploy task
`qa:queue`	`qa-consumers`	QAMessage	Task Dispatcher (scheduler) / Admin API	langgraph (qa-worker)	Post-deploy QA: HTTP checks for GET-only criteria, else Claude Code on prod server

Worker Management

Queue	Group	DTO	Initiator	Consumer	Purpose
`worker:commands`	`worker_manager`	WorkerCommand	langgraph	worker-manager	Create/Delete worker containers
`worker:responses:developer`	—	WorkerResponse	worker-manager	langgraph	Developer worker command responses

Worker I/O (Developer only)

Queue	Group	DTO	Initiator	Consumer	Purpose
`worker:{worker_id}:input`	—	DeveloperWorkerInput	langgraph (DeveloperNode)	worker-wrapper	Task input to Developer worker
`worker:{worker_id}:output`	—	DeveloperWorkerOutput	worker-wrapper	langgraph (DeveloperNode)	Developer worker results

Note: Worker I/O streams use worker:{worker_id}:input/output pattern. Used only for Developer workers. PO communicates via po:input / po:response:{request_id} (see PO ReactAgent I/O below).

Infrastructure

Queue	Group	DTO	Initiator	Consumer	Purpose
`provisioner:queue`	`infrastructure-workers`	ProvisionerMessage	scheduler	infra-service	Provision server
`provisioner:results`	`scheduler-consumers`	ProvisionerResult	infra-service	scheduler	Provisioning result
`provisioner:results`	`telegram-bot`	ProvisionerResult	infra-service	telegram-bot	Provisioning notifications

Events & Progress

Queue	Group	DTO	Initiator	Consumer	Purpose
`task_progress:{task_id}`	—	ProgressEvent	All services	telegram-bot	Task progress notifications
`workflow:status`	—	WorkflowStatusEvent	langgraph (poller)	telegram-bot	Deploy progress updates

Transport Layer Note

Important: The "Initiator" column shows the logical actor — who makes the decision to publish.

PO ReactAgent calls tools directly (Python functions → API/Redis). No CLI proxy needed.

For Developer-Worker messages, the actual transport is:
Developer Worker (AI Agent) → curl localhost:9090 → worker-wrapper → Redis
The HTTP server in worker-wrapper validates and proxies agent results to Redis.

Actor Roles

Actor	Type	Description
PO ReactAgent	LangGraph Agent	Product Owner, communicates via Redis streams `po:input`/`po:response`
Developer-Worker	Worker	Developer agent (inside engineering flow)
langgraph	Service	Workflow orchestrator
worker-manager	Service	Container lifecycle manager
worker-wrapper	Process	Agent bridge inside container
telegram-bot	Service	User interface
infra-service	Service	Server provisioning only (no app deploy)
scheduler	Service	Background tasks

MVP Notes

Important

Tester Node is an MVP stub. It does NOT spawn a Worker.
Implementation: A simple LangGraph node that always returns {"passed": True}.
Post-MVP: Will delegate to a Tester-Worker with code analysis capabilities.

Rate Limiting Policy

Important

External APIs have hard limits. Exceeding them blocks all operations.

Resource	Limit	Strategy
GitHub API	5000 req/hour	Token Bucket in `GitHubAppClient`, buffer 500
LLM API (Claude)	Per-plan limits	Token Bucket per service, cost alerts
External APIs	Varies	Per-client configuration

Design Principles:

Fail-fast: Raise RateLimitExceeded immediately, don't queue silently
Buffer zone: Use 90% of limit, leave 10% for emergencies
Monitoring: Expose metrics for all rate-limited resources
Per-service: Each client manages its own limits (MVP)
Post-MVP: Centralized Redis-based limiter for horizontal scaling

Implementation: shared/clients/github.py (GitHubAppClient).

Flow Diagrams

Engineering Flow

sequenceDiagram
    participant User
    participant TG as telegram-bot
    participant Redis
    participant PO as PO ReactAgent
    participant API
    participant SCH as scheduler
    participant SCF as scaffolder
    participant LG as langgraph
    participant WM as worker-manager

    User->>TG: "Сделай блог"
    TG->>Redis: XADD po:input {text, user_id, request_id}
    Redis-->>PO: Consumer reads (po-consumer group)

    Note over PO: ReactAgent tool calls
    PO->>API: create_project(name, modules)
    API-->>PO: project_id
    PO->>API: create_repository(project_id)
    PO->>API: create_story(project_id, description)
    PO->>Redis: XADD po:response:{request_id} {text: "Начал разработку!"}
    Redis-->>TG: XREAD po:response:{request_id}
    TG->>User: "Начал разработку!"

    Note over SCH: Task Dispatcher (30s poll) detects draft project + stories
    SCH->>Redis: XADD scaffold:queue {project_id, repo_id, modules}
    Redis-->>SCF: Consumer reads scaffold:queue
    SCF->>SCF: copier copy + make setup + git push
    SCF->>API: PATCH /projects/{id} {config.tree, config.specs_summary, status: active}

    PO->>Redis: XADD architect:queue {story_id, project_id}
    Redis-->>SCH: Architect Consumer reads architect:queue
    Note over SCH: Wait for project.status != draft (scaffold completion)
    SCH->>API: GET project (tree + specs_summary)
    Note over SCH: LLM: story → tasks (with project specs context)
    SCH->>API: create tasks with blocked_by chains

    Note over SCH: Task Dispatcher finds unblocked tasks
    SCH->>Redis: XADD engineering:queue {task_id}
    Redis-->>LG: Consumer reads engineering:queue
    LG->>Redis: XADD worker:commands {create}
    Redis-->>WM: Consumer reads worker:commands
    WM->>WM: Create worker container (mounts workspace volume)
    Note over LG: Developer node sends task to worker
    LG->>LG: Continue to Developer node

Deploy Flow (PO-triggered)

sequenceDiagram
    participant User
    participant TG as telegram-bot
    participant Redis
    participant PO as PO ReactAgent
    participant API
    participant LG as langgraph
    participant GH as GitHub API

    User->>TG: "Задеплой проект X"
    TG->>Redis: XADD po:input {text, user_id, request_id}
    Redis-->>PO: Consumer reads (po-consumer group)
    PO->>API: trigger_deploy(project_id)
    PO->>Redis: XADD deploy:queue
    PO->>Redis: XADD po:response:{request_id} {text: "Запускаю деплой!"}
    Redis-->>TG: XREAD po:response:{request_id}
    TG->>User: "Запускаю деплой!"
    Redis-->>LG: Consumer reads
    LG->>LG: DevOps Subgraph (EnvironmentContractLoader → SecretResolver)
    LG->>GH: POST /repos/{owner}/{repo}/actions/workflows/deploy.yml/dispatches
    Note over LG: Poll workflow status
    loop Every 15s
        LG->>GH: GET /repos/{owner}/{repo}/actions/runs?event=workflow_dispatch
        GH-->>LG: {status, conclusion}
    end
    LG->>API: PATCH /tasks/{id} {status: completed}
    LG->>Redis: XADD po:input {type: system_event, event: completed}

Deploy Flow (PR merge detection)

sequenceDiagram
    participant GH as GitHub
    participant SCH as scheduler (pr_poller)
    participant API as api service
    participant DB as PostgreSQL
    participant Redis
    participant DW as deploy-worker
    participant TG as telegram-bot

    Note over SCH: poll_merged_prs() — every 30s
    SCH->>API: GET stories (status=pr_review)
    SCH->>GH: GET pulls (state=closed, head=story/{id})
    GH-->>SCH: merged PR found
    SCH->>API: Transition story → deploying
    SCH->>DB: Create Run (type=deploy)
    SCH->>Redis: XADD deploy:queue {task_id, project_id, user_id, story_id}
    Redis-->>DW: Consumer reads deploy:queue
    DW->>DW: DevOps Subgraph (EnvironmentContractLoader → SecretResolver → Deployer)
    DW->>API: PATCH run.result = DeployOutcome (SUCCESS/CODE_FIX/RETRY/GIVE_UP)
    Note over SCH: supervise_deploying_stories() — every 30s
    SCH->>API: Read deploy run outcome
    alt SUCCESS
        SCH->>API: Story → testing, create QA run
        SCH->>Redis: XADD qa:queue {project_id, deployed_url, ...}
    else GIVE_UP
        SCH->>API: Story → failed
        SCH->>Redis: XADD po:proactive {text: "Deploy failed"}
    end

Note: GitHub webhooks were removed. All PR merge detection and CI failure handling is done by scheduler/src/tasks/pr_poller.py (polling, 30s interval). This eliminates webhook reliability issues for newly scaffolded repos.

Consumer Patterns

Implemented in: Redis Streams unification (#3+#5)

All Redis Stream consumers use unified RedisStreamClient.consume() API from shared.redis_client.

Unified consume() API

async for msg in client.consume(
    stream="engineering:queue",
    group="capability-workers",
    consumer="worker-1",
    auto_ack=False,          # False = caller must call ack() after processing
    claim_pending=True,      # Recover PEL (crashed messages) on startup
    pending_timeout_ms=60_000,  # Min idle time before re-claiming pending message
):
    await process(msg.data)
    await client.ack(stream, group, msg.message_id)

ACK Modes

Mode	`auto_ack`	Use Case	Services
Manual ACK	`False`	At-least-once delivery, ack after successful processing	engineering-worker, deploy-worker, infra-service, worker-manager, scheduler
Auto ACK	`True`	Fire-and-forget, ack on read	telegram-bot (ProactiveListener, ProvisionerNotifier)

PEL Recovery

On startup with claim_pending=True, the consumer calls XAUTOCLAIM to reclaim messages that were pending for longer than pending_timeout_ms. This handles the case where a consumer crashes mid-processing — on restart, the message is automatically re-delivered.

Special case: PO Consumer (services/langgraph/src/consumers/po.py) uses a custom while-loop for concurrent dispatch but still implements PEL recovery via direct XAUTOCLAIM calls on startup.

Consumer Inventory

#	Consumer	File	Queue	ACK	PEL Recovery	Validation
1	Engineering Consumer	`langgraph/src/consumers/engineering.py`	`engineering:queue`	manual	`claim_pending`	in `process_fn`
2	Deploy Consumer	`langgraph/src/consumers/deploy.py`	`deploy:queue`	manual	`claim_pending`	in `process_fn`
3	PO Consumer	`langgraph/src/consumers/po.py`	`po:input`	manual (finally)	`xautoclaim`	`TypeAdapter`
4	Worker Manager	`worker-manager/src/consumer.py`	`worker:commands`	manual	`claim_pending`	`validate_python`
5	Infra Service	`infra-service/src/main.py`	`provisioner:queue`	manual	`claim_pending`	raw dict
6	Scheduler	`scheduler/src/main.py`	`provisioner:results`	manual	`claim_pending`	`model_validate`
7	Provisioner Notifier	`telegram_bot/src/notifications.py`	`provisioner:results`	auto	—	`model_validate`
8	Proactive Listener	`telegram_bot/src/main.py`	`po:proactive`	auto	—	raw dict
9	Architect Consumer	`langgraph/src/consumers/architect.py`	`architect:queue`	manual	`claim_pending`	`model_validate`
10	Scaffolder	`scaffolder/src/consumer.py`	`scaffold:queue`	manual	`claim_pending`	`model_validate`
11	QA Consumer	`langgraph/src/consumers/qa.py`	`qa:queue`	manual	`claim_pending`	`model_validate`

Part 1: REST DTO

ProjectDTO

# shared/contracts/dto/project.py

from enum import StrEnum
from pydantic import BaseModel, ConfigDict

class ProjectStatus(StrEnum):
    """Project lifecycle status.

    Lifecycle only — observable state, not process.
    Activity is derived from child entities (Story/Run).
    Runtime state is tracked by Application.status.
    """

    DRAFT = "draft"
    ACTIVE = "active"
    PAUSED = "paused"
    ARCHIVED = "archived"


class ServiceModule(StrEnum):
    """Available project modules for scaffolding.

    Must match module names in service-template/copier.yml.
    """
    BACKEND = "backend"
    TG_BOT = "tg_bot"
    NOTIFICATIONS = "notifications"
    FRONTEND = "frontend"


class ProjectCreate(BaseModel):
    """Create project request."""
    id: uuid.UUID | None = None
    name: str
    description: str | None = None
    modules: list[ServiceModule] = [ServiceModule.BACKEND]  # Default: backend only
    status: ProjectStatus | None = None


class ProjectUpdate(BaseModel):
    """Update project request."""
    name: str | None = None
    description: str | None = None
    status: ProjectStatus | None = None
    modules: list[ServiceModule] | None = None
    project_spec: dict | None = None


class ProjectDTO(BaseModel):
    """Project response."""
    model_config = ConfigDict(from_attributes=True)

    id: uuid.UUID
    name: str
    description: str | None = None
    status: ProjectStatus
    modules: list[ServiceModule] = []
    config: dict = {}
    owner_id: int
    project_spec: dict | None = None

TaskDTO

# services/api/src/schemas/task.py & shared/contracts/dto/task.py

class TaskStatus(StrEnum):
    BACKLOG = "backlog"
    TODO = "todo"
    IN_DEV = "in_dev"
    IN_CI = "in_ci"
    TESTING = "testing"
    DONE = "done"
    BLOCKED = "blocked"
    WAITING_HUMAN_REVIEW = "waiting_human_review"
    FAILED = "failed"
    CANCELLED = "cancelled"

class TaskType(StrEnum):
    CREATE = "create"
    FEATURE = "feature"
    FIX = "fix"
    REFACTOR = "refactor"

class TaskRead(BaseModel):
    """Schema for reading a task."""
    id: str
    project_id: str
    type: str
    title: str
    description: str | None
    plan: str | None = None
    status: str
    priority: int
    acceptance_criteria: str | None
    need_e2e: bool = False
    current_iteration: int
    max_iterations: int
    failure_metadata: dict[str, Any] | None = None
    created_by: str
    created_at: datetime
    updated_at: datetime
    last_event: str | None = None
    elapsed_minutes: float | None = None

TaskEventDTO

# services/api/src/schemas/task.py & shared/contracts/dto/task.py

class TaskEventType(StrEnum):
    STATUS_CHANGE = "status_change"
    ITERATION_START = "iteration_start"
    ITERATION_END = "iteration_end"
    NOTE = "note"
    COMMENT = "comment"  # Jira-style discussion on a task

class TaskEventRead(BaseModel):
    """Schema for reading a task event."""
    id: int
    task_id: str
    event_type: str
    from_status: str | None
    to_status: str | None
    iteration: int | None
    details: dict[str, Any]
    actor: str
    created_at: datetime

RunDTO

# shared/contracts/dto/run.py

from enum import StrEnum
from datetime import datetime
from pydantic import BaseModel, ConfigDict

class RunStatus(StrEnum):
    QUEUED = "queued"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"


class RunType(StrEnum):
    ENGINEERING = "engineering"
    DEPLOY = "deploy"
    QA = "qa"


class RunCreate(BaseModel):
    """Create run request."""
    project_id: str
    type: RunType
    spec: str | None = None


class RunDTO(TimestampedDTO):
    """Run response."""

    id: str
    project_id: str
    type: RunType
    status: RunStatus
    story_id: str | None = None
    spec: str | None = None
    result: RunResult | None = None   # typed per `type`, see below

Typed `Run.result` (`shared/contracts/dto/run_result.py`)

Run.result is not a free-form dict. Each RunType has exactly one result shape, bound to type by RunDTO:

`RunType`	result model	required field	fields the scheduler routes on
`engineering`	`EngineeringRunResult`	`engineering_status`	(write-only; not routed)
`deploy`	`DeployRunResult`	`deploy_outcome`	`deploy_outcome`, `deployed_url`, `application_id`, `bot_username`, `deploy_fix_attempt`, `error_details`
`qa`	`QARunResult`	`qa_outcome`	`qa_outcome`, `summary`, `failed_checks` (`QAFailedCheck.name`/`.detail`)

Rules (all enforced by validation, tested in shared/tests/unit/test_run_result.py):

The models use extra="forbid", so an unknown field or a payload belonging to another run type (e.g. a QA payload on a deploy run) is rejected. Unknown enum values (an outcome string the code doesn't know) fail the same way.
result=None is allowed only while no outcome exists yet — QUEUED/RUNNING, or a CANCELLED (superseded) run such as a deploy that lost the project lock. A COMPLETED or FAILED run must carry a result; a terminal run without one is rejected, so it surfaces loudly instead of being silently skipped forever. Every producer failure path that reaches a terminal status writes a typed result (deploy outcomes, QAOutcome.ERROR on QA setup failures, EngineeringStatus.FAILED/GAVE_UP on engineering failures).
Producers (langgraph deploy/QA/engineering handlers) construct the typed model and send model_dump(mode="json"), so there is one wire form. Consumers (scheduler supervisor) read outcomes through typed attributes — no .get() guessing, no re-parsing outcome strings.
Storage is unchanged: the API keeps Run.result as a JSON column and its RunRead schema stays dict-typed (dumb passthrough). No DB migration is required — in-flight runs are written by the typed producers, so they parse by construction; historical runs are never re-validated on the API read path.
The scheduler validates only the latest run per story (SchedulerAPIClient.get_latest_run_by_story parses rows[0] alone), so an older legacy/corrupt run in the story's history can never fail a story whose current run is valid. If that latest run fails validation (wrong-type/corrupt result, or a terminal run with no result), the supervisor fails the story once with a loud log and admin notification (supervisor._fail_story_on_invalid_result) — no infinite retry, no silent skip.
deployed_url and application_id are optional on DeployRunResult (a standalone deploy, or a success where the app record couldn't be resolved, legitimately lacks application_id). But a story's QA handoff needs both, so the supervisor checks them in _handle_deploy_success_story before transitioning the story or creating a QA run: a SUCCESS result missing either is routed to a visible failure (fail story + notify admins), never a half-applied handoff that crashes the tick.

UserDTO

# shared/contracts/dto/user.py

from pydantic import BaseModel, ConfigDict

class UserDTO(BaseModel):
    """User response."""
    model_config = ConfigDict(from_attributes=True)
    
    id: int
    telegram_id: int

    is_admin: bool = False

Part 1.1: Additional DTOs

ServerDTO

# shared/contracts/dto/server.py

from enum import StrEnum
from pydantic import BaseModel, ConfigDict
from datetime import datetime

class ServerStatus(StrEnum):
    NEW = "new"
    PENDING_SETUP = "pending_setup"
    PROVISIONING = "provisioning"
    ACTIVE = "active"
    UNREACHABLE = "unreachable"
    MAINTENANCE = "maintenance"
    FORCE_REBUILD = "force_rebuild"
    DISCOVERED = "discovered"


class ServerCreate(BaseModel):
    """Create server request."""
    handle: str
    host: str
    public_ip: str
    ssh_user: str = "root"
    ssh_key: str | None = None
    is_managed: bool = True
    status: str = "discovered"
    labels: dict = {}


class ServerUpdate(BaseModel):
    """Update server request."""
    handle: str | None = None
    host: str | None = None
    public_ip: str | None = None
    ssh_user: str | None = None
    ssh_key: str | None = None
    status: ServerStatus | None = None
    labels: dict | None = None
    is_managed: bool | None = None
    provider_id: str | None = None
    capacity_cpu: int | None = None
    capacity_ram_mb: int | None = None
    capacity_disk_mb: int | None = None
    used_ram_mb: int | None = None
    used_disk_mb: int | None = None
    os_template: str | None = None
    provisioning_started_at: datetime | None = None


class ServerDTO(BaseModel):
    """Server response."""
    model_config = ConfigDict(from_attributes=True)

    handle: str
    host: str
    public_ip: str
    status: str
    provider_id: str | None = None
    is_managed: bool
    labels: dict = {}
    capacity_cpu: int = 0
    capacity_ram_mb: int = 0
    capacity_disk_mb: int = 0
    used_ram_mb: int = 0
    used_disk_mb: int = 0
    os_template: str | None = None
    last_health_check: datetime | None = None
    provisioning_started_at: datetime | None = None
    provisioning_attempts: int = 0

ApplicationDTO

# shared/contracts/dto/application.py

from enum import StrEnum

class ApplicationStatus(StrEnum):
    """Runtime state of an application on a server."""
    NOT_DEPLOYED = "not_deployed"
    DEPLOYING = "deploying"
    RUNNING = "running"
    STOPPING = "stopping"
    STOPPED = "stopped"
    UNDEPLOYING = "undeploying"
    DOWN = "down"
    DEGRADED = "degraded"


class ApplicationDTO(TimestampedDTO):
    """Application response from API."""
    id: int
    repo_id: str
    server_handle: str
    service_name: str
    status: str
    last_health_check: datetime | None = None
    response_time_ms: int | None = None
    ssl_expires_at: datetime | None = None
    uptime_pct_24h: float | None = None
    ports: list[dict[str, Any]] = []

DeploymentDTO

# shared/contracts/dto/deployment.py

from enum import StrEnum

class DeploymentResult(StrEnum):
    """Outcome of a deployment attempt. Immutable after completion."""
    PENDING = "pending"
    SUCCESS = "success"
    FAILED = "failed"
    CANCELED = "canceled"

# services/api/src/schemas/service_deployment.py

class DeploymentRead(TimestampedDTO):
    """Immutable record of a deployment attempt."""
    id: int
    application_id: int | None = None
    project_id: str
    service_name: str
    server_handle: str
    port: int
    result: str
    deployed_sha: str | None = None
    deployed_at: datetime
    deployment_info: dict = {}

Base DTOs

# shared/contracts/dto/base.py

class BaseDTO(BaseModel):
    """Base DTO for all entities."""
    model_config = ConfigDict(from_attributes=True)

class TimestampedDTO(BaseDTO):
    """Base DTO with timestamps."""
    created_at: datetime
    updated_at: datetime | None = None

AgentConfigDTO

# shared/contracts/dto/agent_config.py

from pydantic import BaseModel, ConfigDict

from shared.contracts.vocab import AgentType

class AgentConfigDTO(BaseModel):
    """Agent configuration response."""
    model_config = ConfigDict(from_attributes=True)
    
    id: int
    name: str
    type: AgentType
    model: str
    system_prompt: str
    is_active: bool = True

APIKeyDTO

# shared/contracts/dto/api_key.py

class APIKeyDTO(BaseModel):
    """API Key response."""
    model_config = ConfigDict(from_attributes=True)

    id: int
    service: str
    key_enc: str
    created_at: str | None = None

TaskExecutionDTO

# shared/contracts/dto/task_execution.py

from pydantic import BaseModel, ConfigDict
from datetime import datetime
from typing import Literal, Any

class TaskExecutionDTO(BaseModel):
    """Worker execution record."""
    model_config = ConfigDict(from_attributes=True)
    
    id: str                                    # request_id from worker
    task_id: str | None = None                 # Optional link to high-level task
    worker_id: str
    started_at: datetime
    finished_at: datetime
    duration_ms: int
    exit_code: int
    status: Literal["success", "failure", "in_progress", "error"]
    result_data: dict[str, Any] | None = None  # AgentVerdict or error details
    created_at: datetime

Part 2: Queue Messages

Base Types

# shared/contracts/base.py

from datetime import datetime
from typing import Literal
from pydantic import BaseModel, Field
import uuid


class QueueMeta(BaseModel):
    """Metadata for all queue messages."""
    version: Literal["1"] = "1"
    correlation_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: datetime = Field(default_factory=datetime.utcnow)


class BaseMessage(QueueMeta):
    """Base class for queue messages."""
    request_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    callback_stream: str | None = None


class BaseResult(BaseModel):
    """Base result for async operations."""
    request_id: str
    status: ResultStatus  # shared.contracts.vocab
    error: str | None = None
    duration_ms: int | None = None

ScaffoldMessage

Queue: scaffold:queue Initiator: Task Dispatcher (scheduler, 30s poll) Consumer: scaffolder service

# shared/contracts/queues/scaffold.py

class ScaffoldMessage(BaseMessage):
    """Trigger scaffolding for a project repository.

    Published by scheduler for both new (draft) and existing (active) projects.

    Modes:
        full: Full scaffold — copier + make setup + git push (new projects).
        ensure: Verify workspace exists; if missing, clone + setup (existing projects).
    """
    project_id: str
    repository_id: str
    user_id: str
    template_repo: str    # e.g. "gh:vladmesh/service-template"
    project_name: str     # sanitized name for copier
    modules: str          # comma-separated, e.g. "backend,tg_bot"
    task_description: str = ""
    mode: Literal["full", "ensure"] = "full"

Flow:

Full mode (new projects): Scheduler detects project.status == draft with stories → publishes ScaffoldMessage (mode=full) → Scaffolder runs copier + make setup + git push → saves tree to project.config.tree + parses YAML specs into project.config.specs_summary (models, events, domains) → sets project.status = active. Architect consumer waits for scaffold completion (polls project.status != draft) before decomposing stories.
Ensure mode (existing projects): Scaffold trigger detects ACTIVE projects with TODO tasks → publishes ScaffoldMessage (mode=ensure) → Scaffolder checks if workspace exists; if missing, clones repo + runs setup → sets repository.workspace_ready = True. Task dispatcher checks workspace_ready flag before dispatching. Worker-manager GC calls POST /repositories/{repo_id}/notify-workspace-deleted to clear workspace_ready on deletion.

ArchitectMessage

Queue: architect:queue Initiator: PO ReactAgent (create_story tool) Consumer: scheduler (Architect Consumer)

# shared/contracts/queues/architect.py

class ArchitectMessage(BaseMessage):
    """Trigger story decomposition into tasks."""
    story_id: str
    project_id: str
    user_id: str
    is_reopen: bool = False          # True when story is being reopened (not first decomposition)
    user_report: str | None = None   # User feedback on what's wrong (for reopened stories)

Flow: PO creates Story → publishes ArchitectMessage → Architect Consumer calls LLM to decompose story into N tasks with blocked_by_task_id dependency chains → Task Dispatcher picks up unblocked tasks and publishes EngineeringMessages.

Reopen flow: When user reports a problem with a completed story, PO calls reopen_story tool → story transitions back to in_progress → ArchitectMessage published with is_reopen=True and user_report containing user feedback. Architect reviews previous tasks before creating new fix tasks.

EngineeringMessage

Queue: engineering:queue Initiator: Task Dispatcher (scheduler) Consumer: langgraph

# shared/contracts/queues/engineering.py

class EngineeringMessage(BaseMessage):
    """Start engineering task."""
    task_id: str
    project_id: str
    user_id: str
    action: ActionType = ActionType.CREATE  # shared.contracts.vocab
    description: str | None = None
    skip_deploy: bool = False
    planning_task_id: str | None = None  # planning-layer Task ID for status updates
    story_id: str | None = None  # story ID for worker reuse across tasks
    deploy_fix_attempt: int = 0  # tracks deploy→engineering retry count
    branch: str | None = None  # story branch name (e.g. "story/{story_id}")


class EngineeringResult(BaseResult):
    """Engineering task result."""
    files_changed: list[str] | None = None
    commit_sha: str | None = None
    branch: str | None = None

Action types:

create (default) — new project: scaffold → develop → CI → deploy
feature — add feature to existing project: develop → CI → deploy (no scaffolding)
fix — fix issue in existing project: develop → CI → deploy (no scaffolding)

Flags:

skip_deploy=True — skip auto-deploy after CI passes (develop → CI only)
planning_task_id — when set, engineering worker updates task status (in_dev → done/failed) and writes iteration_end events. Dispatcher-created runs always set this + skip_deploy=True (deploy handled at story level).

DeployMessage

Queue: deploy:queue Initiator: Task Dispatcher (scheduler) / PO Consumer: langgraph

# shared/contracts/queues/deploy.py

class DeployTrigger(StrEnum):
    """Origin of a deploy request."""
    ENGINEERING = "engineering"
    WEBHOOK = "webhook"
    PO = "po"
    ADMIN = "admin"


class DeployAction(StrEnum):
    """Type of deploy operation."""
    CREATE = "create"
    FEATURE = "feature"
    FIX = "fix"
    STOP = "stop"
    UNDEPLOY = "undeploy"


class DeployOutcome(StrEnum):
    """Outcome stored in run.result for dispatcher consumption."""
    SUCCESS = "success"
    SMOKE_FAILURE = "smoke_failure"
    CODE_FIX = "code_fix"
    RETRY = "retry"
    GIVE_UP = "give_up"


class DeployMessage(BaseMessage):
    """Start deploy task."""
    task_id: str
    project_id: str
    user_id: str = ""
    story_id: str = ""
    triggered_by: DeployTrigger = DeployTrigger.ENGINEERING
    action: DeployAction = DeployAction.CREATE
    deploy_fix_attempt: int = 0


class DeployResult(BaseResult):
    """Deploy task result."""
    deployed_url: str | None = None
    server_ip: str | None = None
    port: int | None = None


---

## QAMessage

**Queue:** `qa:queue`
**Initiator:** Task Dispatcher (scheduler) / Admin API
**Consumer:** langgraph (qa-worker)

```python
# shared/contracts/queues/qa.py

class QAOutcome(StrEnum):
    """Outcome stored in run.result for dispatcher consumption."""
    PASSED = "passed"
    FAILED = "failed"
    EXHAUSTED = "exhausted"
    ERROR = "error"


class QAMessage(BaseMessage):
    """Trigger QA testing for a deployed project."""
    story_id: str = ""
    project_id: str
    user_id: str
    deployed_url: str
    application_id: int
    acceptance_criteria: str      # resolved by the producer, never by the consumer
    run_id: str = ""
    bot_username: str | None = None
    qa_attempt: int = 0

Acceptance criteria: Repository.acceptance_criteria is the single source of truth for what QA tests. POST /api/repositories/ seeds every repository with BASELINE_ACCEPTANCE_CRITERIA (shared/contracts/acceptance.py), so a story that never reached the architect still has criteria; the architect's update_acceptance_criteria tool extends the list as stories add functionality. Story and task criteria describe work to be done and are not what QA runs.

Producers (supervisor, admin run-e2e) resolve the criteria and put them on the message. Both refuse to create a QA run without them — the supervisor fails the story visibly before it reaches TESTING, and run-e2e answers 422 — so QA never starts a run it can only error out of.

Health-only criteria: criteria whose every line is a plain - GET <path> returns <status> are decided by the QA consumer over HTTP (parse_health_only_criteria → run_health_checks), with no SSH and no LLM. One prose line sends the whole block to Claude Code on the server instead.

returns <status> means the path itself answers that status, so the checks do not follow redirects: a criterion naming a redirect is checked against the redirect, and a criterion naming 200 is not satisfied by a path that redirects to a 200. Checks are retried while the service is still coming up.

The consumer parses the criteria before resolving anything else, and only the agent branch reads the server, its SSH key, and bot_username. A criteria block the deployed URL alone can answer must not fail over agent scaffolding it never uses.

Flow: Deploy succeeds → supervisor resolves criteria → transitions story to TESTING → creates QA run → publishes QAMessage → QA consumer runs the criteria (HTTP checks, or Claude Code on the prod server) → writes QAOutcome to run.result. Supervisor polls run outcome and routes: PASSED → complete story, FAILED → create fix task + redispatch to engineering, EXHAUSTED/ERROR → fail story.

Lifecycle operations: stop and undeploy actions are handled by the deploy_lifecycle module, which SSHes to the server and runs docker compose stop/down directly — skipping the full DevOps subgraph.

Workflow DTOs

# shared/contracts/queues/workflow.py

class WorkflowTriggerRequest(BaseModel):
    """Request to trigger GitHub Actions workflow."""
    project_id: str
    repo_full_name: str           # "org/repo"
    workflow_file: str = "main.yml"
    inputs: dict[str, str] = {}   # workflow_dispatch inputs

class WorkflowStatusResult(BaseResult):
    """
    Result of workflow execution.
    Derived from: shared.clients.github.WorkflowRun (GitHub API response).
    """
    run_id: int | None = None
    run_url: str | None = None
    deployed_url: str | None = None
    conclusion: Literal["success", "failure", "cancelled", "skipped"] | None = None

class WorkflowStatusEvent(BaseModel):
    """Progress event for workflow execution."""
    project_id: str
    run_id: int
    status: Literal["queued", "in_progress", "completed"]
    conclusion: Literal["success", "failure", "cancelled", "skipped"] | None = None
    current_step: str | None = None
    timestamp: datetime = Field(default_factory=datetime.utcnow)

3. Worker Communication (Single-Listener Pattern)

The Orchestrator (LangGraph) listens to one stream for all worker results:

Stream	Initiator	Consumer	Purpose
`worker:developer:output`	worker-wrapper, worker-manager	LangGraph	All results: success, logical errors, AND crash failures.

Why Single-Listener?
Simpler architecture: LangGraph has one entry point for all outcomes
worker-manager handles crashes (detected via Docker events) and publishes failure results to output:
# When worker-manager detects a container crash via Docker events:
DeveloperWorkerOutput(
    status="failed",
    error="Worker crashed: OOM killed",
    task_id=...,
    request_id=...
)
LangGraph treats all failures uniformly (retry logic in one place)

Queue	Initiator	Consumer	Purpose
`worker:commands`	LangGraph	worker-manager	Command to Create/Delete worker container.
`worker:responses:developer`	worker-manager	langgraph	Responses for Developer worker commands (e.g. "Developer container created").

WorkerCommand / WorkerResponse

Queue (commands): worker:commands Initiator: langgraph Consumer: worker-manager

Queue (responses): worker:responses:developer Initiator: worker-manager Consumer: langgraph

# shared/contracts/queues/worker.py

# AgentType is the canonical enum (shared/contracts/vocab.py), re-exported here.
from shared.contracts.vocab import AgentType  # claude / factory / codex / noop


class WorkerCapability(StrEnum):
    GIT = "git"
    GITHUB_CLI = "github_cli"
    CURL = "curl"


class WorkerChannels(StrEnum):
    """Redis stream channels and patterns."""
    COMMANDS = "worker:commands"
    INPUT_PATTERN = "worker:{worker_id}:input"
    OUTPUT_PATTERN = "worker:{worker_id}:output"


class WorkerConfig(BaseModel):
    """Worker container configuration."""
    name: str
    worker_type: Literal["developer"]         # Worker type for queue naming
    agent_type: AgentType                     # Which AI agent to use
    instructions: str                         # Content for instruction file (CLAUDE.md / AGENTS.md)
    task_content: str | None = None           # Content for TASK.md (optional, for task-driven workers)
    allowed_commands: list[str]               # ["project.*", "engineering.start"]
    capabilities: list[WorkerCapability]      # ["git", "copier"]
    env_vars: dict[str, str] = {}
    auth_mode: Literal["host_session", "api_key"] = "host_session"
    host_claude_dir: str | None = None
    api_key: str | None = None


class CreateWorkerCommand(QueueMeta):
    """Create new worker."""
    command: Literal["create"] = "create"
    request_id: str
    config: WorkerConfig
    context: dict[str, str] = {}   # Additional context (user_id, task_id, etc.)


class DeleteWorkerCommand(QueueMeta):
    """Delete worker."""
    command: Literal["delete"] = "delete"
    request_id: str
    worker_id: str
    reason: Literal["completed", "failed", "timeout"] | None = None


class StatusWorkerCommand(QueueMeta):
    """Get worker status."""
    command: Literal["status"] = "status"
    request_id: str
    worker_id: str


WorkerCommand = CreateWorkerCommand | DeleteWorkerCommand | StatusWorkerCommand


class CreateWorkerResponse(BaseModel):
    """Response to create command."""
    request_id: str
    success: bool
    worker_id: str | None = None
    error: str | None = None


class DeleteWorkerResponse(BaseModel):
    """Response to delete command."""
    request_id: str
    success: bool
    error: str | None = None


class StatusWorkerResponse(BaseModel):
    """Response to status command."""
    request_id: str
    success: bool
    status: Literal["starting", "running", "stopped", "failed"] | None = None
    error: str | None = None


WorkerResponse = CreateWorkerResponse | DeleteWorkerResponse | StatusWorkerResponse

Note: Message passing goes directly to worker queues (worker:{id}:input, etc.), NOT through worker-manager. The manager handles only container lifecycle.

WorkerStatus

# shared/contracts/dto/worker.py

class WorkerStatus(StrEnum):
    STARTING = "STARTING"
    RUNNING = "RUNNING"
    PAUSED = "PAUSED"
    DEAD = "DEAD"
    FAILED = "FAILED"
    STOPPED = "STOPPED"
    GONE = "GONE"       # Stale Redis entry, container no longer exists
    UNKNOWN = "UNKNOWN"

Used across worker-manager (manager, events, introspect router) and langgraph (worker_spawner). Replaces all hardcoded status strings.

EngineeringStatus

# shared/contracts/dto/engineering.py

class EngineeringStatus(StrEnum):
    """Status of the engineering subgraph execution.

    Lifecycle:
        IDLE → (subgraph runs) → DONE | GAVE_UP | FAILED
        FAILED → (supervisor retries) → IDLE  OR  (retries exhausted) → GAVE_UP

    FAILED is transient — supervisor either retries or escalates to GAVE_UP.
    """

    IDLE = "idle"
    DONE = "done"
    GAVE_UP = "gave_up"
    FAILED = "failed"

Used by Developer node and Engineering consumer. Replaces former bare strings ("done", "developer_blocked", "developer_rejected", etc.). The GAVE_UP status covers both former "blocked" (worker hit a blocker) and "rejected" (infra issue) paths — in both cases a human needs to intervene.

PO ReactAgent I/O

Queue	Group	DTO	Initiator	Consumer	Purpose
`po:input`	`po-consumer`	`POInputMessage` (discriminated union: `POUserMessage` / `POSystemEvent` / `POReminderMessage`)	telegram-bot, workers	langgraph (PO consumer)	User messages and system events to PO
`po:response:{request_id}`	—	`POResponse`	langgraph (PO consumer)	telegram-bot	PO response for specific request
`po:proactive`	`tg-bot-proactive`	`POProactiveMessage`	langgraph (PO `notify_user` tool, deploy-worker)	telegram-bot (ProactiveListener)	Proactive messages to users (PO notifications + webhook deploy results)

Transport note: PO streams use flat Redis fields (not JSON data wrapper). Use to_flat_fields() / from_flat_fields() helpers from shared.contracts.queues.po for serialization.

System events: Workers write to po:input (via callback_stream) with type: "system_event". PO decides whether to notify the user via notify_user tool → po:proactive. The old po:events:{task_id} pattern is replaced — events go directly to po:input.

Developer Worker I/O

Terminology: Developer — это конкретная нода внутри Engineering Subgraph. Engineering — абстракция "отдела" для PO. Developer — реализация (воркер, пишущий код). См. GLOSSARY.md.

Коммуникация между LangGraph (Engineering Subgraph, а именно DeveloperNode) и Developer Worker.

Design Decision: Developer Workers are ephemeral (stateless). Each task spawns a fresh worker. Context is the code in repo + error messages — no session persistence needed.

Queue (input): worker:{worker_id}:input Initiator: langgraph (DeveloperNode) Consumer: worker-wrapper (inside Developer container)

Queue (output): worker:{worker_id}:output Initiator: worker-wrapper (inside Developer container) Consumer: langgraph (DeveloperNode)

Note: Developer workers use the worker:{worker_id}:input/output pattern. Each worker gets unique streams identified by worker_id.

# shared/contracts/queues/developer_worker.py

from datetime import datetime
from pydantic import BaseModel, Field
import uuid


class DeveloperWorkerInput(BaseModel):
    """Task for Developer Worker from LangGraph."""

    request_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    task_id: str                    # Engineering task ID
    project_id: str                 # Project UUID
    prompt: str                     # Task specification
    timeout: int = 1800             # Max execution time (seconds)
    timestamp: datetime = Field(default_factory=datetime.utcnow)


class DeveloperWorkerOutput(BaseResult):
    """Result from Developer Worker to LangGraph."""

    # request_id, status, error, duration_ms inherited from BaseResult
    task_id: str                    # Engineering task ID
    commit_sha: str | None = None   # Commit SHA if code was written
    pr_url: str | None = None       # PR URL if created
    timestamp: datetime = Field(default_factory=datetime.utcnow)

Post-MVP: Add previous_attempts: list[AttemptLog] to Input and approach: str to Output for retry context when Tester/CI returns task for rework. For MVP, Developer sees current code and error — sufficient for simple iterations.

ProvisionerMessage

Queue: provisioner:queue
Initiator: scheduler
Consumer: infra-service

# shared/contracts/queues/provisioner.py

class ProvisionerMessage(BaseMessage):
    """Provision server."""
    server_handle: str       # Cloud provider ID (Droplet ID) or unique identifier
    force_reinstall: bool = False
    is_recovery: bool = False


class ProvisionerResult(BaseResult):
    """
    Provisioning result.
    Stream: provisioner:results
    Consumers: scheduler (update DB), telegram-bot (notify admin)
    """
    server_handle: str
    server_ip: str | None = None
    services_redeployed: int = 0
    errors: list[str] | None = None

ProgressEvent

Stream: task_progress:{task_id}
Initiator: All consumers
Consumer: telegram-bot

# shared/contracts/events.py

class ProgressEvent(BaseModel):
    """Task progress notification."""
    type: TaskProgressKind  # LifecycleEvent slice: started/progress/completed/failed
    request_id: str
    task_id: str | None = None
    timestamp: datetime = Field(default_factory=datetime.utcnow)
    message: str | None = None
    progress_pct: int | None = None
    current_step: str | None = None
    error: str | None = None

File Structure

shared/contracts/
├── __init__.py
├── base.py                  # QueueMeta, BaseMessage, BaseResult
├── events.py                # ProgressEvent
├── dto/
│   ├── __init__.py
│   ├── project.py           # ProjectDTO, ProjectCreate
│   ├── task.py              # TaskDTO, TaskCreate
│   ├── user.py              # UserDTO
│   ├── server.py
│   ├── story.py              # StoryDTO, StoryCreate
│   ├── repository.py         # RepositoryDTO, RepositoryCreate
│   ├── brainstorm.py         # BrainstormDTO, BrainstormCreate
│   ├── base.py              # BaseDTO, TimestampedDTO
│   ├── application.py       # ApplicationDTO, ApplicationStatus enum
│   ├── deployment.py        # DeploymentResult enum
│   ├── run.py               # RunDTO, RunType, RunStatus enums
│   ├── engineering.py       # EngineeringStatus enum
│   ├── service_deployment.py  # ServiceDeploymentDTO (legacy alias)
│   ├── agent_config.py
│   ├── allocation.py
│   ├── task_execution.py
│   └── worker.py            # WorkerStatus enum
└── queues/
    ├── __init__.py           # Re-exports PO contracts
    ├── engineering.py
    ├── deploy.py
    ├── scaffold.py           # ScaffoldMessage
    ├── architect.py          # ArchitectMessage
    ├── provisioner.py
    ├── workflow.py
    ├── worker.py
    ├── qa.py                 # QAMessage, QAOutcome, QAServerInfo
    ├── developer_worker.py
    └── po.py                 # POInputMessage, POResponse, POProactiveMessage, flat-field helpers

FilesExpand file tree

CONTRACTS.md

Latest commit

History