From 36564e3f2be530ab9d06e7b90dbf2791bd470493 Mon Sep 17 00:00:00 2001 From: mjalmoudy94 Date: Wed, 14 Oct 2026 19:02:58 +0330 Subject: [PATCH] =?UTF-8?q?feat(engine):=20Phase=204=20=E2=80=94=20schedul?= =?UTF-8?q?er=20tick,=20gate=20chain,=20platform=20git,=20budget?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The AI workforce engine's driver. One tick moves every ready task one step: a turn, a gate, or a platform merge. The AI never runs git. - scheduler.py: tick() over TURN_STAGES/MERGE_STAGES; selection caps (per-org 3, global 8); optimistic advance so two workers cannot both win; worktree created for code pipelines, released on completion/cancel/failure (kept during error_review); budget pause before any turn is spent; merge conflict back-off; transient-failure retries then the manager's error turn. - gates.py: diff guard (empty/.git/CI/symlink/oversize), no-progress hash, tests in the sandbox, green gate that only trusts a green tests row or an explicit waiver. Every verdict is a gate_results row + a task.gate event. Fixed forbidden_paths: lstrip("./") stripped all leading dots, hiding .git and .github from the guard. - container_engine.build_scheduler + worker drives the engine tick. - ManagerOutcome.error_kind; TaskSpendRepository.spend_for_task. - New settings engine_max_concurrent_turns, engine_max_turns_per_org, engine_tests_timeout_seconds, engine_max_merge_conflicts, engine_max_task_credits (+ .env.example). - tests: test_engine_gates.py (13), test_engine_flow.py (11). Checks: pytest 746 passed/1 skipped, ruff clean, mypy clean (160 files). --- .env.example | 12 + apps/api/app/application/container_engine.py | 199 ++++- apps/api/app/application/engine/gates.py | 287 +++++++ apps/api/app/application/engine/manager.py | 18 +- apps/api/app/application/engine/scheduler.py | 801 ++++++++++++++++++ apps/api/app/core/config.py | 10 + .../app/infrastructure/repositories/engine.py | 20 + apps/api/app/workers/__main__.py | 20 +- apps/api/app/workers/ai_runtime.py | 172 ++-- apps/api/tests/engine_fakes.py | 11 + apps/api/tests/test_engine_flow.py | 667 +++++++++++++++ apps/api/tests/test_engine_gates.py | 216 +++++ docs/plan/reports/agent-2-phase-4.md | 197 +++++ 13 files changed, 2563 insertions(+), 67 deletions(-) create mode 100644 apps/api/app/application/engine/gates.py create mode 100644 apps/api/app/application/engine/scheduler.py create mode 100644 apps/api/tests/test_engine_flow.py create mode 100644 apps/api/tests/test_engine_gates.py create mode 100644 docs/plan/reports/agent-2-phase-4.md diff --git a/.env.example b/.env.example index 3f5b7ee..dad4d3e 100644 --- a/.env.example +++ b/.env.example @@ -104,7 +104,19 @@ PLATFORM_INVITE_TTL_HOURS=72 # >>> agent-2 (AI workforce engine) ENGINE_MAX_TOOL_CALLS_PER_TURN=40 +#: Consecutive malformed replies tolerated before a turn gives up. +ENGINE_MAX_PROTOCOL_RETRIES=3 ENGINE_TURN_TIMEOUT_SECONDS=900 +#: Turns running at once, globally and per organization. +ENGINE_MAX_CONCURRENT_TURNS=8 +ENGINE_MAX_TURNS_PER_ORG=3 +#: The tests gate's own budget, and how many merge conflicts before the +#: manager takes over. +ENGINE_TESTS_TIMEOUT_SECONDS=600 +ENGINE_MAX_MERGE_CONFLICTS=3 +#: Per-task credit ceiling (0 = no ceiling). An organization that runs out is +#: paused, not left spinning. +ENGINE_MAX_TASK_CREDITS=0 ENGINE_MAX_CYCLES=6 ENGINE_MAX_EMPLOYEES=20 ENGINE_TICK_SECONDS=10 diff --git a/apps/api/app/application/container_engine.py b/apps/api/app/application/container_engine.py index f0161a3..497cb11 100644 --- a/apps/api/app/application/container_engine.py +++ b/apps/api/app/application/container_engine.py @@ -8,9 +8,18 @@ model" note land in the same transaction as the organization. from __future__ import annotations +from typing import TYPE_CHECKING + from sqlalchemy.ext.asyncio import AsyncSession from app.application.ports.workforce_bootstrap import WorkforceBootstrapPort +from app.domain.sandbox.models import SessionStatus +from app.domain.workforce.models import Task + +if TYPE_CHECKING: + from app.application.engine.manager import ManagerTurns + from app.application.engine.runner import TurnRunner + from app.application.engine.scheduler import Scheduler class _EngineWorkforceBootstrap: @@ -70,4 +79,192 @@ def build_workforce_bootstrap(session: AsyncSession) -> WorkforceBootstrapPort: return _EngineWorkforceBootstrap(session) -__all__ = ["build_workforce_bootstrap"] +# --------------------------------------------------------------------------- +# the scheduler +# --------------------------------------------------------------------------- + + +class _EngineSandboxSessions: + """The sandbox session the engine's turns run in. + + The engine does not *create* sandboxes: a workspace and a running session + belong to the project and are started through the sandbox flow. The engine + only reuses the active session of the task's project. When there is none it + returns ``""``, and the tests gate treats that as red and sends the work + back — never "merge untested because there was no container". + """ + + #: A session in one of these states can still run a command. + ACTIVE = frozenset( + { + SessionStatus.STARTING, + SessionStatus.RUNNING, + SessionStatus.PAUSED, + } + ) + + def __init__(self, session: AsyncSession) -> None: + self.session = session + + async def for_task(self, task: Task) -> str: + from app.infrastructure.repositories.sandbox import ( + SandboxSessionRepository, + WorkspaceRepository, + ) + + if not task.project_id: + return "" + workspaces = await WorkspaceRepository(self.session).list_for_project( + task.project_id + ) + sessions = SandboxSessionRepository(self.session) + for workspace in workspaces: + for sandbox_session in await sessions.list_for_workspace(workspace.id): + if ( + sandbox_session.status in self.ACTIVE + and sandbox_session.external_session_id + ): + return sandbox_session.external_session_id + return "" + + +def _build_manager( + session: AsyncSession, + *, + runner: TurnRunner, + provider, + grants, + memory, +) -> ManagerTurns: + from app.application.engine.manager import ManagerTurns + from app.infrastructure.repositories.ai_control import ( + AiModelRepository, + ModelCredentialRepository, + RateCardRepository, + ) + from app.infrastructure.repositories.audit import UsageLedgerRepository + from app.infrastructure.repositories.engine import ( + EmployeeProfileRepository, + PipelineStateRepository, + ) + from app.infrastructure.repositories.workforce import ( + EmployeeRepository, + MessageRepository, + TaskRepository, + ThreadRepository, + ) + + return ManagerTurns( + provider=provider, + employees=EmployeeRepository(session), + messages=MessageRepository(session), + tasks=TaskRepository(session), + threads=ThreadRepository(session), + models=AiModelRepository(session), + credentials=ModelCredentialRepository(session), + rate_cards=RateCardRepository(session), + ledger=UsageLedgerRepository(session), + grants=grants, + memory=memory, + profiles=EmployeeProfileRepository(session), + states=PipelineStateRepository(session), + runner=runner, + ) + + +def build_scheduler(session: AsyncSession) -> Scheduler: + """The engine's driver, assembled on the caller's session. + + Same rule as the bootstrap: ``container.py`` is frozen, so the engine wires + itself here. Every port the scheduler touches is built from the session or + from the existing composition helpers, so the worker and the API cannot + disagree about which adapter backs a port. + """ + from app.application.container import get_model_provider + from app.application.container_delivery import ( + build_delivery_hooks, + build_repo_workspace, + ) + from app.application.container_org import build_memory, build_model_grants + from app.application.engine.events import RunFeed + from app.application.engine.gates import GateRunner + from app.application.engine.runner import TurnRunner + from app.application.engine.scheduler import Scheduler + from app.infrastructure.adapters.registry import get_sandbox_provider + from app.infrastructure.repositories.ai_control import ( + AiModelRepository, + ModelCredentialRepository, + RateCardRepository, + ) + from app.infrastructure.repositories.app_profile import AppProfileRepository + from app.infrastructure.repositories.audit import UsageLedgerRepository + from app.infrastructure.repositories.billing import CreditWalletRepository + from app.infrastructure.repositories.engine import ( + EmployeeProfileRepository, + EngineTurnRepository, + GateResultRepository, + PipelineStateRepository, + TaskSpendRepository, + ) + from app.infrastructure.repositories.workforce import ( + EmployeeRepository, + RunRepository, + RunStepRepository, + TaskRepository, + ToolCallRepository, + ) + + provider = get_model_provider() + sandbox = get_sandbox_provider() + grants = build_model_grants(session) + memory = build_memory(session) + feed = RunFeed() + + runner = TurnRunner( + provider=provider, + models=AiModelRepository(session), + credentials=ModelCredentialRepository(session), + rate_cards=RateCardRepository(session), + ledger=UsageLedgerRepository(session), + grants=grants, + sandbox=sandbox, + memory=memory, + turns=EngineTurnRepository(session), + run_steps=RunStepRepository(session), + tool_calls=ToolCallRepository(session), + runs=RunRepository(session), + states=PipelineStateRepository(session), + feed=feed, + ) + manager = _build_manager( + session, runner=runner, provider=provider, grants=grants, memory=memory + ) + gate_runner = GateRunner( + results=GateResultRepository(session), + feed=feed, + profiles=AppProfileRepository(session), + ) + sandbox_sessions = _EngineSandboxSessions(session) + + return Scheduler( + states=PipelineStateRepository(session), + tasks=TaskRepository(session), + employees=EmployeeRepository(session), + profiles=EmployeeProfileRepository(session), + runs=RunRepository(session), + turns=EngineTurnRepository(session), + gates=GateResultRepository(session), + manager=manager, + runner=runner, + gate_runner=gate_runner, + repo_workspace=build_repo_workspace(session), + sandbox=sandbox, + memory=memory, + delivery=build_delivery_hooks(session), + wallets=CreditWalletRepository(session), + spend=TaskSpendRepository(session), + sandbox_session_for=sandbox_sessions.for_task, + ) + + +__all__ = ["build_scheduler", "build_workforce_bootstrap"] diff --git a/apps/api/app/application/engine/gates.py b/apps/api/app/application/engine/gates.py new file mode 100644 index 0000000..a0e6267 --- /dev/null +++ b/apps/api/app/application/engine/gates.py @@ -0,0 +1,287 @@ +"""The gate chain that decides whether finished work may move forward. + +A developer's ``WORK_REPORT`` is a claim, not evidence. Four gates check it: + +* **diff guard** — the diff is real, is not empty for a code task, and touches + nothing the platform forbids (``.git``, CI configuration, symlinks that point + outside the repo, files over a megabyte). +* **no-progress** — the same diff twice in a row means the agent is spinning; + the task goes to the manager instead of looping forever. +* **tests** — the app profile's test command runs in the sandbox. Red sends the + work back to the developer with the failure tail; green lets it continue. +* **green** — the tester's ``TEST_PASS`` is only believed when the tests gate is + green, or when the tester explicitly *waives* it with a reason. A waiver is + always stored, never inferred (PLAN-v2 §3). + +Every verdict is a ``gate_results`` row **and** a ``task.gate`` event, because a +gate that only exists in a log is a gate nobody can audit. +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass + +from app.application.engine.events import RunFeed +from app.application.ports.sandbox_provider import SandboxProviderPort +from app.core.config import settings +from app.domain.workforce.engine import GateName, GateResult +from app.domain.workforce.models import Task +from app.domain.workforce.pipeline import TaskKind +from app.infrastructure.repositories.app_profile import AppProfileRepository +from app.infrastructure.repositories.engine import GateResultRepository + +#: Path prefixes a diff may never touch: git internals and CI configuration. +CI_PATHS: tuple[str, ...] = (".github/", ".gitlab-ci.yml", ".gitea/", ".circleci/") + +#: Fail a diff that carries a file larger than this (one megabyte). +MAX_FILE_BYTES = 1_000_000 + +#: How much of the test output is stored and shown. A traceback is at the end. +TEST_OUTPUT_CHARS = 8_000 + + +def diff_hash(text: str) -> str: + """A stable fingerprint of a diff, for the no-progress gate.""" + return hashlib.sha256(text.encode("utf-8", "replace")).hexdigest() + + +def forbidden_paths(changed_files: list[str]) -> list[str]: + """Changed files the diff guard refuses, in the plan's own words.""" + bad: list[str] = [] + for path in changed_files: + normalised = path.strip() + # Strip a leading "./" *prefix*, not the set {".", "/"}: ``lstrip("./")`` + # would turn ".github/ci.yml" into "github/ci.yml" and hide it from the + # guard entirely. + if normalised.startswith("./"): + normalised = normalised[2:] + if normalised.startswith(".git/") or normalised == ".git": + bad.append(path) + continue + if any(normalised == ci.rstrip("/") or normalised.startswith(ci) for ci in CI_PATHS): + bad.append(path) + return bad + + +@dataclass(slots=True, kw_only=True) +class GateOutcome: + """One gate verdict, and whether it sends the task back to the developer.""" + + gate: GateName + ok: bool + detail: str = "" + #: True when the caller must move the task back to implementation. + sends_back: bool = False + + +class GateRunner: + """Runs the gates and leaves a row + an event behind for each verdict.""" + + def __init__( + self, + *, + results: GateResultRepository, + feed: RunFeed, + profiles: AppProfileRepository | None = None, + ) -> None: + self.results = results + self.feed = feed + self.profiles = profiles + + # ------------------------------------------------------------------ # + + async def _record(self, *, task: Task, outcome: GateOutcome) -> GateOutcome: + await self.results.create( + GateResult( + task_id=task.id, + gate=outcome.gate, + ok=outcome.ok, + detail=outcome.detail, + ) + ) + await self.feed.gate( + task_id=task.id, + organization_id=task.organization_id, + gate=str(outcome.gate), + ok=outcome.ok, + detail=outcome.detail, + ) + return outcome + + async def record(self, *, task: Task, outcome: GateOutcome) -> GateOutcome: + """Write a verdict the caller already reasoned about. + + The scheduler records merge conflicts through this instead of the + gate-specific helpers, because a conflict is a diff-guard verdict that + comes from git rather than from hashing the diff. + """ + return await self._record(task=task, outcome=outcome) + + # -- diff guard --------------------------------------------------------- # + + async def diff_guard(self, *, task: Task, diff, code_task: bool) -> GateOutcome: + """Refuse a diff that is empty, forbidden or unreasonably large.""" + problems: list[str] = [] + if diff.is_empty and code_task: + problems.append("تفاوت خالی است: هیچ تغییری ثبت نشده.") + forbidden = forbidden_paths(list(diff.changed_files)) + if forbidden: + problems.append( + "تغییر در مسیرهای ممنوع: " + ", ".join(sorted(forbidden)) + ) + if diff.escaping_symlinks: + problems.append( + "پیوندهای نمادین به بیرون مخزن: " + ", ".join(diff.escaping_symlinks) + ) + if diff.largest_file_bytes > MAX_FILE_BYTES: + problems.append( + f"فایلی بزرگ‌تر از حد مجاز: {diff.largest_file_bytes} بایت." + ) + detail = "\n".join(problems) + return await self._record( + task=task, + outcome=GateOutcome( + gate=GateName.DIFF_GUARD, + ok=not problems, + detail=detail or "تفاوت قابل قبول است.", + sends_back=bool(problems), + ), + ) + + # -- no-progress -------------------------------------------------------- # + + async def no_progress( + self, *, task: Task, previous_hash: str, same_count: int, diff_text: str + ) -> tuple[GateOutcome, str, int]: + """Hash the diff; the same hash twice in a row means no progress. + + Returns the verdict and the new ``(hash, same_count)`` the caller must + persist, so the decision and the counter live in one place. + """ + current = diff_hash(diff_text) + count = same_count + 1 if current and current == previous_hash else 1 + stuck = count >= 2 and bool(current) + outcome = await self._record( + task=task, + outcome=GateOutcome( + gate=GateName.NO_PROGRESS, + ok=not stuck, + detail=( + "دو تفاوت یکسان پشت سر هم: پیشرفتی حاصل نشده." + if stuck + else "پیشرفت تازه‌ای ثبت شد." + ), + sends_back=stuck, + ), + ) + return outcome, current, count + + # -- tests -------------------------------------------------------------- # + + async def test_command_for(self, task: Task) -> str: + """The app profile's test command, or ``""`` when there is none.""" + if self.profiles is None or not task.project_id: + return "" + profile = await self.profiles.get( + organization_id=task.organization_id, project_id=task.project_id + ) + return (profile.test_command or "").strip() if profile else "" + + async def tests( + self, *, task: Task, sandbox: SandboxProviderPort, session_external_id: str + ) -> GateOutcome: + """Run the app profile's tests in the sandbox: red or green.""" + command = await self.test_command_for(task) + if not command: + return await self._record( + task=task, + outcome=GateOutcome( + gate=GateName.TESTS, + ok=True, + detail="دستور تستی برای این برنامه تعریف نشده است.", + ), + ) + if not session_external_id: + return await self._record( + task=task, + outcome=GateOutcome( + gate=GateName.TESTS, + ok=False, + detail="نشست اجرا در دسترس نیست؛ تست‌ها اجرا نشدند.", + sends_back=True, + ), + ) + result = await sandbox.run_command( + session_external_id=session_external_id, + command=command, + timeout_seconds=settings.engine_tests_timeout_seconds, + max_output_bytes=TEST_OUTPUT_CHARS, + ) + ok = result.exit_code == 0 + output = (result.stdout + ("\n" + result.stderr if result.stderr else "")).strip() + tail = output[-TEST_OUTPUT_CHARS:] + detail = ( + f"تست‌ها سبز شدند (کد خروج {result.exit_code})." + if ok + else f"تست‌ها قرمز شدند (کد خروج {result.exit_code}).\n{tail}" + ) + return await self._record( + task=task, + outcome=GateOutcome( + gate=GateName.TESTS, ok=ok, detail=detail, sends_back=not ok + ), + ) + + # -- green -------------------------------------------------------------- # + + async def green( + self, *, task: Task, waived: bool, waiver_reason: str + ) -> GateOutcome: + """Accept a ``TEST_PASS`` only with a green tests gate, or a waiver.""" + passed = await self.results.latest_for(task.id, str(GateName.TESTS)) + tests_ok = passed.ok if passed is not None else False + reason = waiver_reason.strip() + if tests_ok: + return await self._record( + task=task, + outcome=GateOutcome( + gate=GateName.GREEN, ok=True, detail="دروازهٔ تست سبز است." + ), + ) + if waived and reason: + return await self._record( + task=task, + outcome=GateOutcome( + gate=GateName.GREEN, + ok=True, + detail=f"با اعلام صرف‌نظر: {reason}", + ), + ) + return await self._record( + task=task, + outcome=GateOutcome( + gate=GateName.GREEN, + ok=False, + detail="تست‌ها سبز نیستند و صرف‌نظری با دلیل ثبت نشده است.", + sends_back=True, + ), + ) + + # -- helpers the scheduler uses ----------------------------------------- # + + @staticmethod + def code_task(kind: TaskKind | str) -> bool: + """``analysis`` produces prose, not a diff: an empty diff is fine.""" + return TaskKind(kind) is not TaskKind.ANALYSIS + + +__all__ = [ + "CI_PATHS", + "MAX_FILE_BYTES", + "TEST_OUTPUT_CHARS", + "GateOutcome", + "GateRunner", + "diff_hash", + "forbidden_paths", +] diff --git a/apps/api/app/application/engine/manager.py b/apps/api/app/application/engine/manager.py index cab85a0..5074445 100644 --- a/apps/api/app/application/engine/manager.py +++ b/apps/api/app/application/engine/manager.py @@ -158,6 +158,9 @@ class ManagerOutcome: message: str = "" employee_id: str | None = None pipeline: PipelineKey | None = None + #: The runner's error vocabulary, so the scheduler can tell a transient + #: provider failure from a permanent one without re-classifying a string. + error_kind: str = "" def _fuzzy_match(name: str, candidates: list[str]) -> str | None: @@ -654,7 +657,10 @@ class ManagerTurns: ) if not result.ok: return ManagerOutcome( - acted=False, message=result.error, stage=Stage.PENDING_MANAGER_REVIEW + acted=False, + message=result.error, + stage=Stage.PENDING_MANAGER_REVIEW, + error_kind=result.error_kind, ) parsed = parse_reply(result.reply_text, role=RoleKey.MANAGER, mode=Mode.TRIAGE) assign = parsed.first(TagName.ASSIGN) @@ -769,7 +775,10 @@ class ManagerTurns: ) if not result.ok: return ManagerOutcome( - acted=False, message=result.error, stage=Stage.PENDING_FINAL_REVIEW + acted=False, + message=result.error, + stage=Stage.PENDING_FINAL_REVIEW, + error_kind=result.error_kind, ) parsed = parse_reply(result.reply_text, role=RoleKey.MANAGER, mode=Mode.FINAL) if parsed.has(TagName.COMPLETE): @@ -822,7 +831,10 @@ class ManagerTurns: ) if not result.ok: return ManagerOutcome( - acted=False, message=result.error, stage=Stage.ERROR_REVIEW + acted=False, + message=result.error, + stage=Stage.ERROR_REVIEW, + error_kind=result.error_kind, ) parsed = parse_reply(result.reply_text, role=RoleKey.MANAGER, mode=Mode.ERROR) resolve = parsed.first(TagName.RESOLVE_ERROR) diff --git a/apps/api/app/application/engine/scheduler.py b/apps/api/app/application/engine/scheduler.py new file mode 100644 index 0000000..b25d5a3 --- /dev/null +++ b/apps/api/app/application/engine/scheduler.py @@ -0,0 +1,801 @@ +"""The scheduler: one tick moves every ready task one step forward. + +A tick is deliberately boring. It finds the tasks that are *waiting for a turn* +(``STAGE_ROLE`` names an actor and the stage is not terminal), keeps the ones +whose dependencies have completed and whose organization can still pay, picks a +free employee of the stage's role, runs exactly one turn, and applies the +verdict. Everything after ``WORK_REPORT`` is a gate, and everything after the +pipeline is a platform merge — the AI never runs git (PLAN-v2 D6). + +Two properties matter more than throughput: + +* **Idempotence under a lost race.** The stage moves through the optimistic + ``advance`` from ``engine/state.py``. Two workers ticking the same task both + try; exactly one wins and the other sees a ``ConflictError`` and moves on. + That is why the caps below are *selection* caps: several worker processes may + each run a tick, and the compare-and-set keeps them honest. +* **No infinite loops.** A task is sent back only by incrementing its cycle, and + a task at ``max_cycles`` goes to the manager's error turn instead of grinding + forever. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field + +from app.application.engine.gates import GateOutcome, GateRunner +from app.application.engine.manager import ManagerOutcome, ManagerTurns, _pick_granted_model +from app.application.engine.protocol import Tag, TagName +from app.application.engine.runner import TurnRequest, TurnRunner +from app.application.engine.state import advance, park_dependents +from app.application.ports.delivery_hooks import DeliveryHooks, NullDeliveryHooks +from app.application.ports.memory import MemoryPort, NullMemory, TaskOutcome +from app.application.ports.repo_workspace import RepoWorkspacePort, WorktreeRef +from app.application.ports.sandbox_provider import SandboxProviderPort +from app.core.config import settings +from app.core.errors import ConflictError +from app.domain.base import utcnow +from app.domain.workforce.engine import ( + MAX_TRANSIENT_RETRIES, + ErrorKind, + GateName, + is_transient, +) +from app.domain.workforce.models import Employee, RoleKey, Run, RunStatus, Task, TaskStatus +from app.domain.workforce.pipeline import ( + Outcome, + Stage, + TaskPipelineState, + coarse_status, + next_stage, + requires_worktree, +) +from app.infrastructure.repositories.billing import CreditWalletRepository +from app.infrastructure.repositories.engine import ( + EmployeeProfileRepository, + EngineTurnRepository, + GateResultRepository, + PipelineStateRepository, + StaleStageError, + TaskSpendRepository, +) +from app.infrastructure.repositories.workforce import ( + EmployeeRepository, + RunRepository, + TaskRepository, +) + +#: Stages that wait for an AI turn. ``merge_request`` is not here: it is git, +#: and git is the platform's job, not a model's. +TURN_STAGES: tuple[Stage, ...] = ( + Stage.PENDING_MANAGER_REVIEW, + Stage.PENDING_ANALYSIS, + Stage.PENDING_IMPLEMENTATION, + Stage.PENDING_REVIEW, + Stage.PENDING_TEST, + Stage.PENDING_FINAL_REVIEW, + Stage.ERROR_REVIEW, +) +#: The stage the platform executes itself. +MERGE_STAGES: tuple[Stage, ...] = (Stage.MERGE_REQUEST,) + +#: The commit author's e-mail. The platform owns git, so the platform owns the +#: address; the *name* is the employee's, which is what a reviewer reads. +ENGINE_COMMIT_EMAIL = "platform@smart-org.local" + +#: Persian reason shown when work pauses because the money ran out. +BUDGET_PAUSED_MESSAGE = "اعتبار سازمان برای ادامهٔ کار کافی نیست؛ کار موقتاً متوقف شد." + +#: The stage a turn has to be at for each manager mode. +_MANAGER_STAGES: frozenset[Stage] = frozenset( + {Stage.PENDING_MANAGER_REVIEW, Stage.PENDING_FINAL_REVIEW, Stage.ERROR_REVIEW} +) + + +@dataclass(slots=True) +class TickReport: + """What one tick did. The worker logs it; the tests read it.""" + + selected: int = 0 + acted: int = 0 + skipped: int = 0 + failed: int = 0 + merged: int = 0 + paused: int = 0 + decisions: dict[str, int] = field(default_factory=dict) + notes: list[str] = field(default_factory=list) + + def note(self, text: str) -> None: + self.notes.append(text) + + def count(self, key: str) -> None: + self.decisions[key] = self.decisions.get(key, 0) + 1 + + +class Scheduler: + """Runs one turn for every ready task. One instance per session.""" + + def __init__( + self, + *, + states: PipelineStateRepository, + tasks: TaskRepository, + employees: EmployeeRepository, + profiles: EmployeeProfileRepository, + runs: RunRepository, + turns: EngineTurnRepository, + gates: GateResultRepository, + manager: ManagerTurns, + runner: TurnRunner, + gate_runner: GateRunner, + repo_workspace: RepoWorkspacePort, + sandbox: SandboxProviderPort | None = None, + memory: MemoryPort | None = None, + delivery: DeliveryHooks | None = None, + wallets: CreditWalletRepository | None = None, + spend: TaskSpendRepository | None = None, + sandbox_session_for: Callable[[Task], Awaitable[str]] | None = None, + ) -> None: + self.states = states + self.tasks = tasks + self.employees = employees + self.profiles = profiles + self.runs = runs + self.turns = turns + self.gates = gates + self.manager = manager + self.runner = runner + self.gate_runner = gate_runner + self.repo_workspace = repo_workspace + self.sandbox = sandbox + self.memory = memory or NullMemory() + self.delivery = delivery or NullDeliveryHooks() + self.wallets = wallets + self.spend = spend + self.sandbox_session_for = sandbox_session_for + + # ------------------------------------------------------------------ # + # the tick # + # ------------------------------------------------------------------ # + + async def tick(self, *, limit: int = 50) -> TickReport: + """Move every ready task one step. A failure never aborts the tick.""" + report = TickReport() + turn_states = await self.states.list_at_stages(TURN_STAGES, limit=limit) + merge_states = await self.states.list_at_stages(MERGE_STAGES, limit=limit) + + selected = self._select(turn_states) + report.selected = len(selected) + for state in selected: + await self._guarded(self._step, state, report) + + for state in self._select(merge_states): + await self._guarded(self._merge_step, state, report) + return report + + async def _guarded(self, step, state: TaskPipelineState, report: TickReport) -> None: + """One task must never stop the tick: catch, record, continue.""" + try: + await step(state, report) + except ConflictError: + # Another worker moved this task first; that is a normal outcome. + report.skipped += 1 + report.note(f"task {state.task_id}: موتور دیگری زودتر حرکت کرد") + except Exception as exc: # noqa: BLE001 - the tick keeps going + report.failed += 1 + report.note(f"task {state.task_id}: {type(exc).__name__}: {exc}") + + def _select(self, states: list[TaskPipelineState]) -> list[TaskPipelineState]: + """Apply the caps: per organization first, then globally.""" + per_org: dict[str, int] = {} + chosen: list[TaskPipelineState] = [] + for state in states: + count = per_org.get(state.organization_id, 0) + if count >= settings.engine_max_turns_per_org: + continue + if len(chosen) >= settings.engine_max_concurrent_turns: + break + per_org[state.organization_id] = count + 1 + chosen.append(state) + return chosen + + async def _step(self, state: TaskPipelineState, report: TickReport) -> None: + task = await self.tasks.get(state.task_id) + if task is None or task.status in ( + TaskStatus.DONE, + TaskStatus.FAILED, + TaskStatus.CANCELLED, + ): + report.skipped += 1 + return + + reason = await self._budget_reason(task) + if reason is not None: + await self._pause(state, task, reason) + report.paused += 1 + return + + if not await self._dependencies_done(state): + report.skipped += 1 + report.note(f"task {task.id}: وابستگی‌ها هنوز تمام نشده‌اند") + return + + actor = state.actor() + if actor is None: + report.skipped += 1 + return + employee = await self._employee_for(state, actor.role) + if employee is None: + report.skipped += 1 + report.note(f"task {task.id}: کارمند آزادی برای نقش {actor.role.value} نیست") + return + + await self.profiles.set_busy(employee_id=employee.id, task_id=task.id) + try: + if state.stage in _MANAGER_STAGES: + await self._manager_step(state, task, employee, report) + else: + await self._work_step(state, task, employee, report) + finally: + await self.profiles.set_busy(employee_id=employee.id, task_id=None) + + # ------------------------------------------------------------------ # + # manager stages # + # ------------------------------------------------------------------ # + + async def _manager_step( + self, state: TaskPipelineState, task: Task, employee: Employee, report: TickReport + ) -> None: + run = await self._open_run(task, employee) + try: + if state.stage is Stage.PENDING_MANAGER_REVIEW: + outcome = await self.manager.triage(task=task, run_id=run.id, employee=employee) + elif state.stage is Stage.PENDING_FINAL_REVIEW: + outcome = await self.manager.final_review( + task=task, + run_id=run.id, + employee=employee, + spec=await self._last_reply(task.id, RoleKey.PLANNER), + diff_text=(await self._diff_for(task, state))[0], + test_output=await self._gate_detail(task.id, str(GateName.TESTS)), + gate_summary=await self._gate_summary(task.id), + ) + else: + outcome = await self.manager.error_review( + task=task, + run_id=run.id, + employee=employee, + error_summary=await self._error_context(task), + ) + except Exception as exc: # noqa: BLE001 - one task must not stop the tick + await self._close_run(run, ok=False, error=str(exc)) + report.failed += 1 + report.note(f"task {task.id}: {type(exc).__name__}: {exc}") + return + await self._close_run(run, ok=outcome.acted, error="" if outcome.acted else outcome.message) + await self._settle(state, task, outcome, report) + + async def _settle( + self, + state: TaskPipelineState, + task: Task, + outcome: ManagerOutcome, + report: TickReport, + ) -> None: + """Turn a manager verdict into a tick result (or a failure).""" + report.count(outcome.decision or "idle") + if outcome.acted: + report.acted += 1 + if outcome.stage in (Stage.CANCELLED, Stage.FAILED): + await park_dependents(self.states, task_id=task.id) + return + if outcome.error_kind: + await self._retry_or_fail(state, task, outcome.message, outcome.error_kind) + report.failed += 1 + return + report.skipped += 1 + report.note(f"task {task.id}: تصمیمی گرفته نشد ({outcome.message[:120]})") + + # ------------------------------------------------------------------ # + # work stages (planner / developer / reviewer / tester) # + # ------------------------------------------------------------------ # + + async def _work_step( + self, state: TaskPipelineState, task: Task, employee: Employee, report: TickReport + ) -> None: + actor = state.actor() + if actor is None: # pragma: no cover - _step checked this + report.skipped += 1 + return + worktree = await self._worktree_for(task, state) + run = await self._open_run(task, employee) + request = TurnRequest( + organization_id=task.organization_id, + task=task, + role=actor.role, + mode=actor.mode, + stage=state.stage, + run_id=run.id, + employee=employee, + model_id=await self.manager._model_id_for(employee, task.organization_id), + sandbox_root=worktree.sandbox_path if worktree else "/workspace", + session_external_id=await self._sandbox_session(task), + extra_context=await self._work_context(task, state), + user_id=task.created_by_user_id, + ) + result = await self.runner.run_turn(request) + if not result.ok: + await self._close_run(run, ok=False, error=result.error) + await self._retry_or_fail(state, task, result.error, result.error_kind) + report.failed += 1 + return + await self._close_run(run, ok=True) + await self._apply_work_outcome(state, task, employee, result.tags, report, worktree) + + async def _apply_work_outcome( + self, + state: TaskPipelineState, + task: Task, + employee: Employee, + tags: tuple[Tag, ...], + report: TickReport, + worktree: WorktreeRef | None, + ) -> None: + names = {tag.name.value for tag in tags} + if state.stage is Stage.PENDING_ANALYSIS: + # The planner's plan *is* the deliverable for analysis_only. + await self._advance_or_complete(state, task) + elif state.stage is Stage.PENDING_IMPLEMENTATION: + await self._after_implementation(state, task, employee, report, worktree) + return + elif state.stage is Stage.PENDING_REVIEW: + if "REVIEW_CHANGES" in names: + await self._send_back(state, task, "بازبین اصلاحاتی خواست.") + report.failed += 1 + return + await self._advance_or_complete(state, task) + elif state.stage is Stage.PENDING_TEST: + await self._after_tester(state, task, tags, report) + return + else: # pragma: no cover - TURN_STAGES has no other work stage + report.skipped += 1 + report.note(f"task {task.id}: مرحلهٔ ناشناخته {state.stage.value}") + return + report.acted += 1 + + # -- the gate chain ---------------------------------------------------- # + + async def _after_implementation( + self, + state: TaskPipelineState, + task: Task, + employee: Employee, + report: TickReport, + worktree: WorktreeRef | None, + ) -> None: + if worktree is None: + await self._fail_task( + state, task, "درخت کاری برای این کار ساخته نشد.", ErrorKind.MODEL + ) + report.failed += 1 + return + diff = await self.repo_workspace.diff(worktree=worktree) + code_task = GateRunner.code_task(state.kind) + + guard = await self.gate_runner.diff_guard(task=task, diff=diff, code_task=code_task) + if not guard.ok: + await self._send_back(state, task, guard.detail) + report.failed += 1 + return + await self._save_branch(state, worktree) + + progress, new_hash, same_count = await self.gate_runner.no_progress( + task=task, + previous_hash=state.last_diff_hash, + same_count=state.same_diff_count, + diff_text=diff.text, + ) + await self._patch(state, last_diff_hash=new_hash, same_diff_count=same_count) + if not progress.ok: + await self._send_back(state, task, progress.detail) + report.failed += 1 + return + + if code_task and self.sandbox is not None: + tests = await self.gate_runner.tests( + task=task, + sandbox=self.sandbox, + session_external_id=await self._sandbox_session(task), + ) + if not tests.ok: + await self._escalate_after_repeated_red(task, employee) + await self._send_back(state, task, tests.detail) + report.failed += 1 + return + await self._advance_or_complete(state, task) + report.acted += 1 + + async def _after_tester( + self, + state: TaskPipelineState, + task: Task, + tags: tuple[Tag, ...], + report: TickReport, + ) -> None: + names = {tag.name.value for tag in tags} + if TagName.TEST_FAIL.value in names: + await self._send_back(state, task, "آزمون‌گر شکست تست‌ها را گزارش کرد.") + report.failed += 1 + return + waiver = next( + (tag for tag in tags if tag.name is TagName.TEST_PASS), None + ) + reason = waiver.attr("waive", "").strip() if waiver is not None else "" + gate = await self.gate_runner.green(task=task, waived=bool(reason), waiver_reason=reason) + if not gate.ok: + await self._send_back(state, task, gate.detail) + report.failed += 1 + return + if reason: + await self._patch(state, waiver_reason=reason) + await self._advance_or_complete(state, task) + report.acted += 1 + + # ------------------------------------------------------------------ # + # merge (git is the platform's) # + # ------------------------------------------------------------------ # + + async def _merge_step(self, state: TaskPipelineState, report: TickReport) -> None: + task = await self.tasks.get(state.task_id) + if task is None or task.status in ( + TaskStatus.DONE, + TaskStatus.FAILED, + TaskStatus.CANCELLED, + ): + report.skipped += 1 + return + worktree = await self._worktree_for(task, state) + if worktree is None: + await self._fail_task(state, task, "درخت کاری برای ادغام وجود ندارد.", ErrorKind.MODEL) + report.failed += 1 + return + employee = await self._task_employee(task) + author = f"{employee.name if employee else 'موتور'} <{ENGINE_COMMIT_EMAIL}>" + sha = await self.repo_workspace.commit( + worktree=worktree, message=f"task/{task.id}: {task.title}", author=author + ) + if sha is None: + await self._fail_task(state, task, "هیچ تغییری برای ثبت وجود ندارد.", ErrorKind.MODEL) + report.failed += 1 + return + + synced = await self.repo_workspace.sync_with_base(worktree=worktree) + if not synced.ok: + await self._merge_conflict(state, task, synced.conflict_files, report) + return + if synced.base_changed and self.sandbox is not None: + tests = await self.gate_runner.tests( + task=task, + sandbox=self.sandbox, + session_external_id=await self._sandbox_session(task), + ) + if not tests.ok: + await self._send_back(state, task, tests.detail) + report.failed += 1 + return + + merged = await self.repo_workspace.merge_into( + project_id=task.project_id or "", branch=worktree.branch, target="dev" + ) + if not merged.ok: + await self._merge_conflict(state, task, merged.conflict_files, report) + return + pushed = await self.repo_workspace.push(project_id=task.project_id or "", branch="dev") + if not pushed.ok: + await self._fail_task( + state, task, f"فرستادن شاخه ناموفق بود: {pushed.message}", ErrorKind.NETWORK + ) + report.failed += 1 + return + + final_sha = pushed.sha or merged.sha or sha + await self._move(state, task, Stage.COMPLETED) + await self.repo_workspace.remove_worktree(worktree=worktree) + if task.project_id: + await self.delivery.on_branch_pushed( + organization_id=task.organization_id, + project_id=task.project_id, + branch="dev", + sha=final_sha, + source="engine", + ) + await self._remember_outcome(task, state, final_sha) + report.merged += 1 + report.acted += 1 + + async def _merge_conflict( + self, + state: TaskPipelineState, + task: Task, + files: list[str], + report: TickReport, + ) -> None: + detail = "تعارض ادغام در: " + ", ".join(files or ["نامشخص"]) + await self.gate_runner.record( + task=task, + outcome=GateOutcome(gate=GateName.DIFF_GUARD, ok=False, detail=detail), + ) + attempts = state.retry_count + 1 + if attempts >= settings.engine_max_merge_conflicts: + await self._fail_task( + state, task, "تعارض ادغام پس از چند تلاش برطرف نشد.", ErrorKind.PROTOCOL + ) + report.failed += 1 + return + await self._patch(state, retry_count=attempts) + await self._send_back(state, task, detail) + report.failed += 1 + + # ------------------------------------------------------------------ # + # moving a task # + # ------------------------------------------------------------------ # + + async def _advance_or_complete(self, state: TaskPipelineState, task: Task) -> None: + await self._move(state, task, next_stage(state.pipeline, state.stage, Outcome.OK)) + + async def _send_back(self, state: TaskPipelineState, task: Task, detail: str) -> None: + """Cycle once and return to the developer, or ask the manager.""" + cycles = state.cycle + 1 + await self._patch(state, cycle=cycles) + if cycles >= state.max_cycles: + await self._move(state, task, Stage.ERROR_REVIEW) + return + await self._move(state, task, Stage.PENDING_IMPLEMENTATION) + + async def _move(self, state: TaskPipelineState, task: Task, target: Stage) -> None: + if target is state.stage: + return + await advance(self.states, task_id=task.id, expected_from=state.stage, to=target) + await self.tasks.set_status(task.id, coarse_status(target)) + if target in (Stage.FAILED, Stage.CANCELLED): + # A dead task keeps no worktree behind it, and its dependents must + # not wait for work that will never finish (agybot bug M1). + await self._release_worktree(task, state) + await park_dependents(self.states, task_id=task.id) + + async def _release_worktree(self, task: Task, state: TaskPipelineState) -> None: + worktree = await self._worktree_for(task, state) + if worktree is not None: + await self.repo_workspace.remove_worktree(worktree=worktree) + + async def _patch(self, state: TaskPipelineState, **fields: object) -> None: + try: + updated = await self.states.patch( + task_id=state.task_id, expected_version=state.version, **fields + ) + except StaleStageError: + # Another worker moved the row first; the next tick sees the truth. + return + # Keep the caller's snapshot current so a second patch in the same step + # locks against the version this write produced, not the one before it. + state.version = updated.version + + # ------------------------------------------------------------------ # + # budget # + # ------------------------------------------------------------------ # + + async def _pause(self, state: TaskPipelineState, task: Task, reason: str) -> None: + await self._move(state, task, Stage.PAUSED) + await self._post_note(task, reason) + + async def _budget_reason(self, task: Task) -> str | None: + if self.wallets is not None: + wallet = await self.wallets.get_for_org(task.organization_id) + # No wallet means no budget was ever set: a fresh org must not be + # frozen out of its first task. + if wallet is not None and wallet.balance <= 0: + return BUDGET_PAUSED_MESSAGE + if self.spend is not None and settings.engine_max_task_credits > 0: + used = await self.spend.spend_for_task(task.id) + if used >= settings.engine_max_task_credits: + return "سقف اعتبار این کار پر شده است؛ کار متوقف شد." + return None + + async def _retry_or_fail( + self, state: TaskPipelineState, task: Task, message: str, kind: str + ) -> None: + """A transient failure gets two more tries; then the manager decides.""" + if not is_transient(kind): + await self._fail_task(state, task, message, ErrorKind(kind)) + return + attempts = state.retry_count + 1 + if attempts > MAX_TRANSIENT_RETRIES: + await self._fail_task(state, task, message, ErrorKind.UNAVAILABLE) + return + await self._patch(state, retry_count=attempts) + + async def _fail_task( + self, state: TaskPipelineState, task: Task, reason: str, kind: ErrorKind + ) -> None: + if state.stage is Stage.ERROR_REVIEW: + # The error turn tried and could not act: the task is failed. + await self._move(state, task, Stage.FAILED) + else: + await self._move(state, task, Stage.ERROR_REVIEW) + await self._post_note(task, reason) + + # ------------------------------------------------------------------ # + # small helpers # + # ------------------------------------------------------------------ # + + async def _open_run(self, task: Task, employee: Employee) -> Run: + return await self.runs.create( + Run( + task_id=task.id, + employee_id=employee.id, + status=RunStatus.RUNNING, + started_at=utcnow(), + ) + ) + + async def _close_run(self, run: Run, *, ok: bool, error: str = "") -> None: + await self.runs.set_status( + run.id, + RunStatus.COMPLETED if ok else RunStatus.FAILED, + error=error or None, + finished_at=utcnow(), + ) + + async def _dependencies_done(self, state: TaskPipelineState) -> bool: + for dependency in state.depends_on: + other = await self.states.get(dependency) + if other is None: + continue + if other.stage is not Stage.COMPLETED: + return False + return True + + async def _employee_for(self, state: TaskPipelineState, role: RoleKey) -> Employee | None: + roster = [ + e + for e in await self.employees.list_for_org(state.organization_id, limit=200) + if e.is_available and e.role_key is role + ] + free: list[tuple[int, Employee]] = [] + for employee in roster: + profile = await self.profiles.get(employee.id) + if profile is None or profile.busy_task_id is not None: + continue + free.append((await self.tasks.count_open_for_employee(employee.id), employee)) + if not free: + return None + free.sort(key=lambda pair: pair[0]) + return free[0][1] + + async def _task_employee(self, task: Task) -> Employee | None: + if not task.assigned_employee_id: + return None + return await self.employees.get_for_org(task.organization_id, task.assigned_employee_id) + + async def _worktree_for(self, task: Task, state: TaskPipelineState) -> WorktreeRef | None: + if not requires_worktree(state.pipeline) or not task.project_id: + return None + return await self.repo_workspace.create_task_worktree( + project_id=task.project_id, task_id=task.id, base="dev" + ) + + async def _save_branch(self, state: TaskPipelineState, worktree: WorktreeRef) -> None: + if state.branch != worktree.branch: + await self._patch(state, branch=worktree.branch) + + async def _sandbox_session(self, task: Task) -> str: + if self.sandbox_session_for is None: + return "" + return await self.sandbox_session_for(task) + + async def _diff_for(self, task: Task, state: TaskPipelineState) -> tuple[str, list[str]]: + worktree = await self._worktree_for(task, state) + if worktree is None: + return "", [] + diff = await self.repo_workspace.diff(worktree=worktree) + return diff.text, list(diff.changed_files) + + async def _last_reply(self, task_id: str, role: RoleKey) -> str: + turns = await self.turns.list_for_task(task_id) + for turn in reversed(turns): + if turn.role is role and turn.ok: + return turn.reply_excerpt + return "" + + async def _gate_detail(self, task_id: str, gate: str) -> str: + row = await self.gates.latest_for(task_id, gate) + return row.detail if row else "" + + async def _gate_summary(self, task_id: str) -> str: + rows = await self.gates.list_for_task(task_id, limit=40) + if not rows: + return "" + return "\n".join( + f"{row.gate.value}: {'سبز' if row.ok else 'قرمز'}" for row in rows[-6:] + ) + + async def _work_context(self, task: Task, state: TaskPipelineState) -> str: + parts: list[str] = [] + spec = await self._last_reply(task.id, RoleKey.PLANNER) + if state.stage in ( + Stage.PENDING_IMPLEMENTATION, + Stage.PENDING_REVIEW, + Stage.PENDING_TEST, + ) and spec: + parts.append("برنامهٔ اجرا:\n" + spec) + if state.stage is Stage.PENDING_IMPLEMENTATION: + review = await self._last_reply(task.id, RoleKey.REVIEWER) + if review: + parts.append("نظر بازبین:\n" + review) + tests = await self._gate_detail(task.id, str(GateName.TESTS)) + if tests: + parts.append("آخرین نتیجهٔ تست:\n" + tests) + elif state.stage is Stage.PENDING_REVIEW: + diff_text, files = await self._diff_for(task, state) + parts.append("فایل‌های تغییریافته: " + ", ".join(files)) + parts.append("تفاوت:\n" + diff_text[:8000]) + return "\n\n".join(p for p in parts if p) + + async def _error_context(self, task: Task) -> str: + tests = await self._gate_detail(task.id, str(GateName.TESTS)) + return tests or "کار به بازبینی خطا رسید." + + async def _escalate_after_repeated_red(self, task: Task, employee: Employee) -> None: + """A second red run: move the developer onto the strongest granted model. + + One escalation only. A model that is already the strongest, or an org + with nothing stronger granted, is left alone — the cycle guard is what + eventually stops the loop. + """ + rows = await self.gates.list_for_task(task.id, limit=40) + reds = sum(1 for row in rows if row.gate is GateName.TESTS and not row.ok) + if reds < 2: + return + granted = await self.manager.grants.granted_models(organization_id=task.organization_id) + strong = _pick_granted_model(granted, "strong", fallback=False) + if not strong: + return + profile = await self.profiles.get(employee.id) + if profile is None or profile.ai_model_id == strong: + return + await self.profiles.set_model(employee_id=employee.id, ai_model_id=strong) + + async def _post_note(self, task: Task, content: str) -> None: + if not content: + return + await self.manager._post_system_message( + organization_id=task.organization_id, + project_id=task.project_id, + content=content, + ) + + async def _remember_outcome(self, task: Task, state: TaskPipelineState, sha: str) -> None: + try: + await self.memory.on_task_finished( + organization_id=task.organization_id, + task=TaskOutcome( + task_id=task.id, + title=task.title, + status="completed", + summary=f"شاخهٔ dev با شناسهٔ {sha[:12]} به‌روزرسانی شد.", + cycles=state.cycle, + ), + ) + except Exception: # noqa: BLE001 - memory must never fail a task + return + + +__all__ = [ + "BUDGET_PAUSED_MESSAGE", + "ENGINE_COMMIT_EMAIL", + "MERGE_STAGES", + "TURN_STAGES", + "Scheduler", + "TickReport", +] diff --git a/apps/api/app/core/config.py b/apps/api/app/core/config.py index 87d9021..d630e07 100644 --- a/apps/api/app/core/config.py +++ b/apps/api/app/core/config.py @@ -149,6 +149,16 @@ class Settings(BaseSettings): #: the turn timeout, hundreds of paid calls later. engine_max_protocol_retries: int = 3 engine_turn_timeout_seconds: int = 900 + #: Scheduler concurrency: turns running at once, globally and per org. + engine_max_concurrent_turns: int = 8 + engine_max_turns_per_org: int = 3 + #: The tests gate's own budget, and how often a merge may conflict before + #: the manager takes over. + engine_tests_timeout_seconds: int = 600 + engine_max_merge_conflicts: int = 3 + #: Per-task credit ceiling (0 = no ceiling). Checked before every turn: + #: an organization that ran out of credit is paused, not left spinning. + engine_max_task_credits: float = 0.0 engine_max_cycles: int = 6 engine_max_employees: int = 20 engine_tick_seconds: float = 10.0 diff --git a/apps/api/app/infrastructure/repositories/engine.py b/apps/api/app/infrastructure/repositories/engine.py index 1ad7b29..8ffa959 100644 --- a/apps/api/app/infrastructure/repositories/engine.py +++ b/apps/api/app/infrastructure/repositories/engine.py @@ -34,6 +34,7 @@ from app.domain.workforce.pipeline import ( TaskKind, TaskPipelineState, ) +from app.infrastructure.db.models import RunRow, UsageEventRow from app.infrastructure.db.models_engine import ( EmployeeProfileRow, EngineTurnRow, @@ -485,3 +486,22 @@ class GateResultRepository: ) row = (await self.session.execute(stmt)).scalar_one_or_none() return _gate(row) if row else None + + +class TaskSpendRepository: + """Per-task spend, read from the usage ledger (never written here). + + The scheduler's budget guard needs "how much has this task cost so far", + which the ledger can answer by joining its events to the task's runs. The + ledger stays the source of truth; this only reads it. + """ + + def __init__(self, session: AsyncSession) -> None: + self.session = session + + async def spend_for_task(self, task_id: str) -> float: + run_ids = select(RunRow.id).where(RunRow.task_id == task_id) + stmt = select(func.coalesce(func.sum(UsageEventRow.total_price), 0.0)).where( + UsageEventRow.run_id.in_(run_ids) + ) + return float((await self.session.execute(stmt)).scalar_one()) diff --git a/apps/api/app/workers/__main__.py b/apps/api/app/workers/__main__.py index 8a00807..9247cb7 100644 --- a/apps/api/app/workers/__main__.py +++ b/apps/api/app/workers/__main__.py @@ -32,8 +32,11 @@ def _build_parser() -> argparse.ArgumentParser: parser.add_argument( "--interval", type=float, - default=300.0, - help="seconds between passes when --loop is set (default: 300)", + default=None, + help=( + "seconds between passes when --loop is set " + "(default: 300, or the engine tick for ai-runtime)" + ), ) parser.add_argument("--limit", type=int, default=10, help="max rows per pass (default: 10)") parser.add_argument( @@ -121,9 +124,20 @@ async def _main_async(args: argparse.Namespace) -> int: if args.loop: install_signal_handlers(stop) + interval = args.interval + if interval is None: + # The engine's driver wants to tick often; every other worker is a + # slower rollup and the historical 300s default suits it. + if args.worker == "ai-runtime": + from app.core.config import settings + + interval = settings.engine_tick_seconds + else: + interval = 300.0 + outcomes = await run_periodically( one_pass, - interval_seconds=max(1.0, args.interval), + interval_seconds=max(1.0, interval), stop=stop, run_once=not args.loop, worker_name=args.worker, diff --git a/apps/api/app/workers/ai_runtime.py b/apps/api/app/workers/ai_runtime.py index 110728c..173fa6e 100644 --- a/apps/api/app/workers/ai_runtime.py +++ b/apps/api/app/workers/ai_runtime.py @@ -14,6 +14,7 @@ the service's business. from __future__ import annotations from app.application.container import build_workforce_service +from app.application.container_engine import build_scheduler from app.application.ports.realtime import emit_worker_status from app.core.errors import AppError from app.domain.audit.models import ErrorSeverity, ErrorSource @@ -24,12 +25,25 @@ from app.workers.base import WorkerOutcome, worker_session class AiRuntimeWorker: - """Drains the queue of tasks that nobody has run yet.""" + """Drains the queue of tasks that nobody has run yet. + + The worker drives **two** engines in one pass, oldest first: + + * the legacy run engine — runnable tasks handed to + ``WorkforceService.start_run``; + * the workforce engine (Phase 4) — one ``Scheduler.tick()``, which moves + every ready task one stage forward (a turn, a gate, or a platform merge). + + Both run on the same session, so a run started here and a stage advanced + here commit together. ``run_once`` (the CLI default) makes one pass — the + cron shape; ``--loop`` keeps ticking (set the interval to + ``ENGINE_TICK_SECONDS``). + """ name = "ai-runtime" async def drain(self, *, limit: int = 10) -> WorkerOutcome: - """Run up to ``limit`` queued tasks, oldest first. + """Run up to ``limit`` queued tasks, then one engine tick. A task that fails is recorded and the pass continues: one poisoned task must not block the queue behind it. @@ -37,66 +51,104 @@ class AiRuntimeWorker: outcome = WorkerOutcome(name=self.name) async with worker_session() as session: - tasks = TaskRepository(session) - workforce = build_workforce_service(session) - - for task in await tasks.list_runnable(limit=limit): - # Re-read through the service so the tenant check and the - # "already closed" guard both apply. The worker gets no - # privileged bypass. - try: - current = await workforce.get_task_or_404( - organization_id=task.organization_id, task_id=task.id - ) - except AppError: - outcome.skipped += 1 - outcome.note(f"task {task.id}: vanished before the run") - continue - - if current.status not in (TaskStatus.NEW, TaskStatus.TRIAGED): - outcome.skipped += 1 - continue - - try: - run = await workforce.start_run( - organization_id=task.organization_id, - task_id=task.id, - user_id=task.created_by_user_id or "", - employee_id=task.assigned_employee_id, - ) - except Exception as exc: # noqa: BLE001 - keep draining - outcome.failed += 1 - outcome.note(f"task {task.id}: {type(exc).__name__}: {exc}") - # Persist with the tenant attached: a task that fails to run - # is a tenant-visible problem, and the console must be able - # to show that organization what happened. - await record_error( - kind="task_run_failed", - message=str(exc) or type(exc).__name__, - exc=exc, - source=ErrorSource.WORKER, - severity=ErrorSeverity.ERROR, - organization_id=task.organization_id, - user_id=task.created_by_user_id, - metadata={ - "worker": self.name, - "task_id": task.id, - "employee_id": task.assigned_employee_id, - }, - ) - continue - - outcome.processed += 1 - outcome.note(f"task {task.id} -> run {run.id} ({run.status})") - # This worker is its own process: without a push here, a run - # started by the queue is invisible until someone reloads. - await emit_worker_status( - organization_id=task.organization_id, - worker=self.name, - message=f"کار «{current.title}» در حال اجراست (اجرا {run.id}).", - ) + await self._drain_runs(session, outcome, limit=limit) + await self._tick_engine(session, outcome, limit=limit) return outcome + async def _drain_runs(self, session, outcome: WorkerOutcome, *, limit: int) -> None: + tasks = TaskRepository(session) + workforce = build_workforce_service(session) + + for task in await tasks.list_runnable(limit=limit): + # Re-read through the service so the tenant check and the + # "already closed" guard both apply. The worker gets no + # privileged bypass. + try: + current = await workforce.get_task_or_404( + organization_id=task.organization_id, task_id=task.id + ) + except AppError: + outcome.skipped += 1 + outcome.note(f"task {task.id}: vanished before the run") + continue + + if current.status not in (TaskStatus.NEW, TaskStatus.TRIAGED): + outcome.skipped += 1 + continue + + try: + run = await workforce.start_run( + organization_id=task.organization_id, + task_id=task.id, + user_id=task.created_by_user_id or "", + employee_id=task.assigned_employee_id, + ) + except Exception as exc: # noqa: BLE001 - keep draining + outcome.failed += 1 + outcome.note(f"task {task.id}: {type(exc).__name__}: {exc}") + # Persist with the tenant attached: a task that fails to run + # is a tenant-visible problem, and the console must be able + # to show that organization what happened. + await record_error( + kind="task_run_failed", + message=str(exc) or type(exc).__name__, + exc=exc, + source=ErrorSource.WORKER, + severity=ErrorSeverity.ERROR, + organization_id=task.organization_id, + user_id=task.created_by_user_id, + metadata={ + "worker": self.name, + "task_id": task.id, + "employee_id": task.assigned_employee_id, + }, + ) + continue + + outcome.processed += 1 + outcome.note(f"task {task.id} -> run {run.id} ({run.status})") + # This worker is its own process: without a push here, a run + # started by the queue is invisible until someone reloads. + await emit_worker_status( + organization_id=task.organization_id, + worker=self.name, + message=f"کار «{current.title}» در حال اجراست (اجرا {run.id}).", + ) + + async def _tick_engine(self, session, outcome: WorkerOutcome, *, limit: int) -> None: + """One scheduler tick: move every ready task one step. + + A tick failure is recorded against the worker and the pass ends — the + run drain has already committed its own work, and a broken tick must + not look like a clean one. + """ + try: + scheduler = build_scheduler(session) + report = await scheduler.tick(limit=max(limit, 10)) + except Exception as exc: # noqa: BLE001 - a worker must survive a bad tick + outcome.failed += 1 + outcome.note(f"engine tick: {type(exc).__name__}: {exc}") + await record_error( + kind="engine_tick_failed", + message=str(exc) or type(exc).__name__, + exc=exc, + source=ErrorSource.WORKER, + severity=ErrorSeverity.ERROR, + metadata={"worker": self.name}, + ) + return + + outcome.processed += report.acted + report.merged + outcome.skipped += report.skipped + outcome.failed += report.failed + if report.selected or report.merged or report.paused: + outcome.note( + f"engine tick: selected={report.selected} acted={report.acted} " + f"merged={report.merged} paused={report.paused} failed={report.failed}" + ) + for note in report.notes: + outcome.note(note) + __all__ = ["AiRuntimeWorker"] diff --git a/apps/api/tests/engine_fakes.py b/apps/api/tests/engine_fakes.py index 4d1976c..c917a92 100644 --- a/apps/api/tests/engine_fakes.py +++ b/apps/api/tests/engine_fakes.py @@ -225,6 +225,8 @@ class RecordingMemory: def __init__(self, context: str = "") -> None: self.context = context self.queries: list[dict] = [] + self.recorded: list[dict] = [] + self.finished: list[dict] = [] async def build_context( self, *, organization_id: str, role: str, query: str, budget_chars: int = 2500 @@ -244,9 +246,18 @@ class RecordingMemory: evidence_task_id: str | None = None, key: str | None = None, ) -> str | None: + self.recorded.append( + { + "organization_id": organization_id, + "kind": kind, + "content": content, + "source": source, + } + ) return None async def on_task_finished(self, *, organization_id: str, task) -> None: + self.finished.append({"organization_id": organization_id, "task": task}) return None diff --git a/apps/api/tests/test_engine_flow.py b/apps/api/tests/test_engine_flow.py new file mode 100644 index 0000000..4b822e2 --- /dev/null +++ b/apps/api/tests/test_engine_flow.py @@ -0,0 +1,667 @@ +"""End-to-end: the scheduler walks a task from a queued idea to merged code. + +This is the 4.6 flow test. The repositories, the scheduler, the manager and the +turn runner are all real; only the model (scripted), the sandbox (scripted) and +git (in-memory ``FakeRepoWorkspace``) are fakes. What it defends: + +* a task advances ``analysis → implementation → test → final review → merge`` + one stage per tick, and the coarse ``tasks.status`` follows the stage +* the platform — never the model — runs git: ``commit``/``merge_into``/``push`` + were called, the commit author carries the platform address, and the model + was never asked to run a command +* every gate leaves a ``gate_results`` row (diff guard, no-progress, tests, + green) +* red tests and a forbidden diff send the work *back* with cycle + 1 instead of + merging it +* ``analysis_only`` never creates a worktree +* an empty wallet pauses the task before a single turn is spent +* a transient provider failure is retried, then handed to the manager +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from dataclasses import dataclass + +import pytest_asyncio + +from app.application.engine.events import RunFeed +from app.application.engine.gates import GateRunner +from app.application.engine.manager import ManagerTurns +from app.application.engine.runner import TurnRunner +from app.application.engine.scheduler import BUDGET_PAUSED_MESSAGE, ENGINE_COMMIT_EMAIL, Scheduler +from app.application.ports import CommandResult +from app.application.ports.delivery_hooks import NullDeliveryHooks +from app.application.ports.repo_workspace import DiffResult, FakeRepoWorkspace, MergeOutcome +from app.core.errors import ProviderUnavailableError +from app.domain.base import new_id +from app.domain.workforce.engine import EmployeeProfile +from app.domain.workforce.models import Employee, RoleKey, TaskStatus +from app.domain.workforce.pipeline import PipelineKey, Stage, TaskKind, TaskPipelineState +from app.infrastructure.db.models import ( + AiModelRow, + AiProviderRow, + OrganizationRow, + ProjectRow, + TaskRow, + UserRow, +) +from app.infrastructure.repositories.ai_control import ( + AiModelRepository, + ModelCredentialRepository, + RateCardRepository, +) +from app.infrastructure.repositories.app_profile import AppProfileRepository +from app.infrastructure.repositories.audit import UsageLedgerRepository +from app.infrastructure.repositories.engine import ( + EmployeeProfileRepository, + EngineTurnRepository, + GateResultRepository, + PipelineStateRepository, +) +from app.infrastructure.repositories.workforce import ( + EmployeeRepository, + MessageRepository, + RunRepository, + RunStepRepository, + TaskRepository, + ThreadRepository, + ToolCallRepository, +) +from tests.engine_fakes import ( + FakeGrants, + FakeSandbox, + RecordingMemory, + ScriptedModel, + ScriptedReply, +) + + +@dataclass +class _World: + organization_id: str + user_id: str + project_id: str + strong_id: str + + +class _EmptyWallet: + """A wallet with no credit, and no way to top it up in a test.""" + + balance = 0.0 + + +class _Wallets: + def __init__(self, wallet: _EmptyWallet | None) -> None: + self._wallet = wallet + + async def get_for_org(self, organization_id: str): + return self._wallet + + +class _EmptyWalletRepo(_Wallets): + def __init__(self) -> None: + super().__init__(_EmptyWallet()) + + +@pytest_asyncio.fixture +async def world(session_factory, seeded) -> AsyncIterator[_World]: + async with session_factory() as session: + user = UserRow(email=f"flow-{new_id()[:8]}@example.com", password_hash="x") + session.add(user) + await session.flush() + org = OrganizationRow( + name="سازمان جریان", slug=f"flow-{user.id[:6]}", owner_user_id=user.id + ) + session.add(org) + await session.flush() + project = ProjectRow( + id=new_id(), + organization_id=org.id, + name="app", + slug="app", + test_command="pytest -q", + ) + session.add(project) + provider_id = (await session.execute(AiProviderRow.__table__.select().limit(1))).first().id + strong = await AiModelRepository(session).create( + AiModelRow( + provider_id=provider_id, + internal_name=f"strong-{new_id()[:4]}", + display_name_for_admin="مدل قوی", + capabilities=["text", "code", "reasoning"], + input_price_per_1m=0.5, + output_price_per_1m=3.0, + is_active=True, + ) + ) + await session.commit() + yield _World( + organization_id=org.id, + user_id=user.id, + project_id=project.id, + strong_id=strong.id, + ) + + +async def _employee(session, world: _World, name: str, role: RoleKey) -> Employee: + employee = await EmployeeRepository(session).create( + Employee(organization_id=world.organization_id, name=name, role_key=role) + ) + await EmployeeProfileRepository(session).upsert( + EmployeeProfile(employee_id=employee.id, ai_model_id=world.strong_id, roles=[role.value]) + ) + return employee + + +async def _team(session, world: _World) -> dict[RoleKey, Employee]: + return { + RoleKey.MANAGER: await _employee(session, world, "مدیر", RoleKey.MANAGER), + RoleKey.PLANNER: await _employee(session, world, "برنامه‌ریز", RoleKey.PLANNER), + RoleKey.DEVELOPER: await _employee(session, world, "توسعه‌دهنده", RoleKey.DEVELOPER), + RoleKey.REVIEWER: await _employee(session, world, "بازبین", RoleKey.REVIEWER), + RoleKey.TESTER: await _employee(session, world, "آزمون‌گر", RoleKey.TESTER), + } + + +async def _task(session, world: _World, *, stage: Stage, pipeline=PipelineKey.PLAN_CODE_TEST, + kind=TaskKind.FEATURE) -> TaskRow: + row = TaskRow( + organization_id=world.organization_id, + title="کار جریان", + project_id=world.project_id, + created_by_user_id=world.user_id, + ) + session.add(row) + await session.flush() + await PipelineStateRepository(session).create( + TaskPipelineState( + task_id=row.id, + organization_id=world.organization_id, + project_id=world.project_id, + pipeline=pipeline, + kind=kind, + stage=stage, + ) + ) + await session.flush() + return row + + +def _scheduler( + session, + world: _World, + *, + scripted, + sandbox: FakeSandbox, + workspace: FakeRepoWorkspace, + memory: RecordingMemory, + delivery, + wallets=None, +) -> Scheduler: + grants = FakeGrants({world.strong_id: "strong"}) + runner = TurnRunner( + provider=scripted, + models=AiModelRepository(session), + credentials=ModelCredentialRepository(session), + rate_cards=RateCardRepository(session), + ledger=UsageLedgerRepository(session), + grants=grants, + sandbox=sandbox, + memory=memory, + turns=EngineTurnRepository(session), + run_steps=RunStepRepository(session), + tool_calls=ToolCallRepository(session), + runs=RunRepository(session), + states=PipelineStateRepository(session), + feed=RunFeed(), + ) + manager = ManagerTurns( + provider=scripted, + employees=EmployeeRepository(session), + messages=MessageRepository(session), + tasks=TaskRepository(session), + threads=ThreadRepository(session), + models=AiModelRepository(session), + credentials=ModelCredentialRepository(session), + rate_cards=RateCardRepository(session), + ledger=UsageLedgerRepository(session), + grants=grants, + memory=memory, + profiles=EmployeeProfileRepository(session), + states=PipelineStateRepository(session), + runner=runner, + feed=RunFeed(), + ) + gate_runner = GateRunner( + results=GateResultRepository(session), + feed=RunFeed(), + profiles=AppProfileRepository(session), + ) + + async def _session_for(_task) -> str: + return "sess-1" + + return Scheduler( + states=PipelineStateRepository(session), + tasks=TaskRepository(session), + employees=EmployeeRepository(session), + profiles=EmployeeProfileRepository(session), + runs=RunRepository(session), + turns=EngineTurnRepository(session), + gates=GateResultRepository(session), + manager=manager, + runner=runner, + gate_runner=gate_runner, + repo_workspace=workspace, + sandbox=sandbox, + memory=memory, + delivery=delivery, + wallets=wallets, + sandbox_session_for=_session_for, + ) + + +async def _gates(session, task_id: str) -> dict[str, bool]: + rows = await GateResultRepository(session).list_for_task(task_id) + return {row.gate.value: row.ok for row in rows} + + +def _calls(workspace: FakeRepoWorkspace, name: str) -> list[dict]: + return [kwargs for called, kwargs in workspace.calls if called == name] + + +async def _run(scheduler: Scheduler, session, ticks: int) -> None: + for _ in range(ticks): + await scheduler.tick() + await session.flush() + + +# --------------------------------------------------------------------------- +# the whole pipeline +# --------------------------------------------------------------------------- + + +async def test_a_task_runs_from_triage_to_merged_code(world, session_factory): + scripted = ScriptedModel( + [ + ScriptedReply(content='[[ASSIGN employee="" pipeline="plan_code_review_test"/]]'), + ScriptedReply(content="[[PLAN_SPEC]]پلان کار[[/PLAN_SPEC]]"), + ScriptedReply(content="[[WORK_REPORT]]پیاده‌سازی شد[[/WORK_REPORT]]"), + ScriptedReply(content="[[REVIEW_PASS]]قابل قبول است[[/REVIEW_PASS]]"), + ScriptedReply(content="[[TEST_PASS]]سبز است[[/TEST_PASS]]"), + ScriptedReply(content="[[COMPLETE/]]"), + ] + ) + sandbox = FakeSandbox() + workspace = FakeRepoWorkspace() + memory = RecordingMemory() + delivery = NullDeliveryHooks() + + async with session_factory() as session: + await _team(session, world) + row = await _task(session, world, stage=Stage.PENDING_MANAGER_REVIEW) + workspace.diffs[row.id] = DiffResult( + changed_files=["app/main.py"], text="--- a\n+++ b\n@@\n+x\n" + ) + await session.commit() + scheduler = _scheduler( + session, + world, + scripted=scripted, + sandbox=sandbox, + workspace=workspace, + memory=memory, + delivery=delivery, + ) + await _run(scheduler, session, 7) + state = await PipelineStateRepository(session).get(row.id) + task = await TaskRepository(session).get(row.id) + gates = await _gates(session, row.id) + + assert state is not None and state.stage is Stage.COMPLETED + assert state.branch == f"task/{row.id}" + assert task is not None and task.status is TaskStatus.DONE + # Five worker turns plus one triage and one final verdict; the merge is the + # platform's, so it costs no model call. + assert scripted.calls == 6 + assert gates["diff_guard"] is True + assert gates["no_progress"] is True + assert gates["tests"] is True + assert gates["green"] is True + + commits = _calls(workspace, "commit") + assert len(commits) == 1 and ENGINE_COMMIT_EMAIL in commits[0]["author"] + assert _calls(workspace, "merge_into")[0]["target"] == "dev" + assert _calls(workspace, "push")[0]["branch"] == "dev" + assert len(_calls(workspace, "remove_worktree")) == 1 + assert delivery.pushed and delivery.pushed[0]["source"] == "engine" + assert len(memory.finished) == 1 + + # The model was never asked to run git: no shell command reached the sandbox + # for git, and the only command it saw was the test command. + assert all("git" not in call["command"] for call in sandbox.executed) + + +async def test_the_scheduler_never_runs_git_through_the_model(world, session_factory): + scripted = ScriptedModel([ScriptedReply(content="[[WORK_REPORT]]x[[/WORK_REPORT]]")]) + sandbox = FakeSandbox() + workspace = FakeRepoWorkspace() + async with session_factory() as session: + await _team(session, world) + row = await _task(session, world, stage=Stage.PENDING_IMPLEMENTATION) + workspace.diffs[row.id] = DiffResult(changed_files=["a.py"], text="d") + await session.commit() + scheduler = _scheduler( + session, + world, + scripted=scripted, + sandbox=sandbox, + workspace=workspace, + memory=RecordingMemory(), + delivery=NullDeliveryHooks(), + ) + await _run(scheduler, session, 1) + assert scripted.all_text() # a prompt was built + assert all("git" not in call["command"] for call in sandbox.executed) + + +# --------------------------------------------------------------------------- +# gates send work back +# --------------------------------------------------------------------------- + + +async def test_red_tests_send_the_work_back_then_green_lets_it_through( + world, session_factory +): + scripted = ScriptedModel( + [ + ScriptedReply(content="[[WORK_REPORT]]تلاش یک[[/WORK_REPORT]]"), + ScriptedReply(content="[[WORK_REPORT]]تلاش دو[[/WORK_REPORT]]"), + ] + ) + sandbox = FakeSandbox( + commands={ + "pytest": [ + CommandResult(exit_code=1, stdout="", stderr="AssertionError"), + CommandResult(exit_code=0, stdout="ok"), + ] + } + ) + workspace = FakeRepoWorkspace() + async with session_factory() as session: + await _team(session, world) + row = await _task(session, world, stage=Stage.PENDING_IMPLEMENTATION) + workspace.diffs[row.id] = DiffResult(changed_files=["a.py"], text="diff-1") + await session.commit() + scheduler = _scheduler( + session, + world, + scripted=scripted, + sandbox=sandbox, + workspace=workspace, + memory=RecordingMemory(), + delivery=NullDeliveryHooks(), + ) + await _run(scheduler, session, 1) + after_red = await PipelineStateRepository(session).get(row.id) + gates_red = await _gates(session, row.id) + + # A different diff on the second attempt, or the no-progress gate fires. + workspace.diffs[row.id] = DiffResult(changed_files=["a.py", "b.py"], text="diff-2") + await _run(scheduler, session, 1) + after_green = await PipelineStateRepository(session).get(row.id) + gates_green = await _gates(session, row.id) + + assert after_red is not None and after_red.stage is Stage.PENDING_IMPLEMENTATION + assert after_red.cycle == 1 + assert gates_red["tests"] is False + assert after_green is not None and after_green.stage is Stage.PENDING_TEST + assert gates_green["tests"] is True + + +async def test_a_forbidden_diff_never_reaches_git(world, session_factory): + scripted = ScriptedModel([ScriptedReply(content="[[WORK_REPORT]]x[[/WORK_REPORT]]")]) + sandbox = FakeSandbox() + workspace = FakeRepoWorkspace() + async with session_factory() as session: + await _team(session, world) + row = await _task(session, world, stage=Stage.PENDING_IMPLEMENTATION) + workspace.diffs[row.id] = DiffResult( + changed_files=[".github/workflows/ci.yml"], text="ci" + ) + await session.commit() + scheduler = _scheduler( + session, + world, + scripted=scripted, + sandbox=sandbox, + workspace=workspace, + memory=RecordingMemory(), + delivery=NullDeliveryHooks(), + ) + await _run(scheduler, session, 1) + state = await PipelineStateRepository(session).get(row.id) + gates = await _gates(session, row.id) + + assert state is not None and state.stage is Stage.PENDING_IMPLEMENTATION + assert state.cycle == 1 + assert gates["diff_guard"] is False + assert _calls(workspace, "commit") == [] + + +async def test_a_tester_failure_sends_the_work_back(world, session_factory): + scripted = ScriptedModel([ScriptedReply(content="[[TEST_FAIL]]قرمز[[/TEST_FAIL]]")]) + workspace = FakeRepoWorkspace() + async with session_factory() as session: + await _team(session, world) + row = await _task(session, world, stage=Stage.PENDING_TEST) + await session.commit() + scheduler = _scheduler( + session, + world, + scripted=scripted, + sandbox=FakeSandbox(), + workspace=workspace, + memory=RecordingMemory(), + delivery=NullDeliveryHooks(), + ) + await _run(scheduler, session, 1) + state = await PipelineStateRepository(session).get(row.id) + assert state is not None and state.stage is Stage.PENDING_IMPLEMENTATION + assert state.cycle == 1 + + +async def test_a_test_pass_without_green_tests_is_not_believed(world, session_factory): + scripted = ScriptedModel([ScriptedReply(content="[[TEST_PASS]]سبز[[/TEST_PASS]]")]) + workspace = FakeRepoWorkspace() + async with session_factory() as session: + await _team(session, world) + row = await _task(session, world, stage=Stage.PENDING_TEST) + await session.commit() + scheduler = _scheduler( + session, + world, + scripted=scripted, + sandbox=FakeSandbox(), + workspace=workspace, + memory=RecordingMemory(), + delivery=NullDeliveryHooks(), + ) + await _run(scheduler, session, 1) + state = await PipelineStateRepository(session).get(row.id) + gates = await _gates(session, row.id) + assert state is not None and state.stage is Stage.PENDING_IMPLEMENTATION + assert gates["green"] is False + + +# --------------------------------------------------------------------------- +# analysis, budget, dependencies, failures +# --------------------------------------------------------------------------- + + +async def test_analysis_only_finishes_without_a_worktree(world, session_factory): + scripted = ScriptedModel([ScriptedReply(content="[[PLAN_SPEC]]گزارش[[/PLAN_SPEC]]")]) + workspace = FakeRepoWorkspace() + async with session_factory() as session: + await _team(session, world) + row = await _task( + session, + world, + stage=Stage.PENDING_ANALYSIS, + pipeline=PipelineKey.ANALYSIS_ONLY, + kind=TaskKind.ANALYSIS, + ) + await session.commit() + scheduler = _scheduler( + session, + world, + scripted=scripted, + sandbox=FakeSandbox(), + workspace=workspace, + memory=RecordingMemory(), + delivery=NullDeliveryHooks(), + ) + await _run(scheduler, session, 1) + state = await PipelineStateRepository(session).get(row.id) + task = await TaskRepository(session).get(row.id) + + assert state is not None and state.stage is Stage.COMPLETED + assert task is not None and task.status is TaskStatus.DONE + assert _calls(workspace, "create_task_worktree") == [] + assert _calls(workspace, "commit") == [] + + +async def test_an_empty_wallet_pauses_the_task_before_a_turn_is_spent( + world, session_factory +): + scripted = ScriptedModel([ScriptedReply(content="[[ASSIGN/]]")]) + workspace = FakeRepoWorkspace() + async with session_factory() as session: + await _team(session, world) + row = await _task(session, world, stage=Stage.PENDING_MANAGER_REVIEW) + await session.commit() + scheduler = _scheduler( + session, + world, + scripted=scripted, + sandbox=FakeSandbox(), + workspace=workspace, + memory=RecordingMemory(), + delivery=NullDeliveryHooks(), + wallets=_EmptyWalletRepo(), + ) + await _run(scheduler, session, 1) + state = await PipelineStateRepository(session).get(row.id) + task = await TaskRepository(session).get(row.id) + + assert state is not None and state.stage is Stage.PAUSED + assert task is not None and task.status is TaskStatus.WAITING_APPROVAL + assert scripted.calls == 0 + assert BUDGET_PAUSED_MESSAGE + + +async def test_a_task_waits_for_its_dependencies(world, session_factory): + scripted = ScriptedModel( + [ + ScriptedReply(content='[[TEST_PASS waive="مشکل شناخته‌شده"]]سبز[[/TEST_PASS]]'), + ScriptedReply(content="[[WORK_REPORT]]done[[/WORK_REPORT]]"), + ] + ) + workspace = FakeRepoWorkspace() + async with session_factory() as session: + await _team(session, world) + first = await _task(session, world, stage=Stage.PENDING_TEST) + second = await _task(session, world, stage=Stage.PENDING_IMPLEMENTATION) + state = await PipelineStateRepository(session).get(second.id) + assert state is not None + await PipelineStateRepository(session).patch( + task_id=second.id, expected_version=state.version, depends_on=[first.id] + ) + workspace.diffs[second.id] = DiffResult(changed_files=["b.py"], text="diff-b") + await session.commit() + scheduler = _scheduler( + session, + world, + scripted=scripted, + sandbox=FakeSandbox(), + workspace=workspace, + memory=RecordingMemory(), + delivery=NullDeliveryHooks(), + ) + await _run(scheduler, session, 1) + blocked = await PipelineStateRepository(session).get(second.id) + + # The dependency finished, so the next tick lets the work run. + dependency = await PipelineStateRepository(session).get(first.id) + assert dependency is not None + await PipelineStateRepository(session).patch( + task_id=first.id, + expected_version=dependency.version, + stage=str(Stage.COMPLETED), + ) + await _run(scheduler, session, 1) + unblocked = await PipelineStateRepository(session).get(second.id) + + assert blocked is not None and blocked.stage is Stage.PENDING_IMPLEMENTATION + assert unblocked is not None and unblocked.stage is Stage.PENDING_TEST + + +async def test_a_transient_failure_is_retried_then_handed_to_the_manager( + world, session_factory +): + scripted = ScriptedModel( + [ + ScriptedReply(raises=ProviderUnavailableError("429 rate limit")), + ScriptedReply(raises=ProviderUnavailableError("429 rate limit")), + ScriptedReply(raises=ProviderUnavailableError("429 rate limit")), + ] + ) + workspace = FakeRepoWorkspace() + async with session_factory() as session: + await _team(session, world) + row = await _task(session, world, stage=Stage.PENDING_IMPLEMENTATION) + workspace.diffs[row.id] = DiffResult(changed_files=["a.py"], text="d") + await session.commit() + scheduler = _scheduler( + session, + world, + scripted=scripted, + sandbox=FakeSandbox(), + workspace=workspace, + memory=RecordingMemory(), + delivery=NullDeliveryHooks(), + ) + await _run(scheduler, session, 3) + state = await PipelineStateRepository(session).get(row.id) + + assert state is not None and state.stage is Stage.ERROR_REVIEW + assert state.retry_count == 2 + + +async def test_a_merge_conflict_sends_the_work_back_for_another_pass( + world, session_factory +): + scripted = ScriptedModel() + workspace = FakeRepoWorkspace() + workspace.sync_outcomes = [MergeOutcome(ok=False, conflict_files=["app/main.py"])] + async with session_factory() as session: + await _team(session, world) + row = await _task(session, world, stage=Stage.MERGE_REQUEST) + workspace.diffs[row.id] = DiffResult(changed_files=["app/main.py"], text="d") + await session.commit() + scheduler = _scheduler( + session, + world, + scripted=scripted, + sandbox=FakeSandbox(), + workspace=workspace, + memory=RecordingMemory(), + delivery=NullDeliveryHooks(), + ) + await _run(scheduler, session, 1) + state = await PipelineStateRepository(session).get(row.id) + + assert state is not None and state.stage is Stage.PENDING_IMPLEMENTATION + assert state.retry_count == 1 and state.cycle == 1 + diff --git a/apps/api/tests/test_engine_gates.py b/apps/api/tests/test_engine_gates.py new file mode 100644 index 0000000..5df26c2 --- /dev/null +++ b/apps/api/tests/test_engine_gates.py @@ -0,0 +1,216 @@ +"""The gate chain: what blocks work, and what leaves a record behind. + +Every gate is tested through the *real* repository, because a gate that writes +no ``gate_results`` row is a gate nobody can audit. The sandbox is the only +fake: no command ever leaves the test process. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator + +import pytest_asyncio + +from app.application.engine.events import RunFeed +from app.application.engine.gates import ( + MAX_FILE_BYTES, + GateRunner, + diff_hash, + forbidden_paths, +) +from app.application.ports import CommandResult +from app.application.ports.repo_workspace import DiffResult +from app.domain.base import new_id +from app.domain.workforce.engine import GateName, GateResult +from app.domain.workforce.models import Task +from app.domain.workforce.pipeline import TaskKind +from app.infrastructure.db.models import OrganizationRow, ProjectRow, TaskRow, UserRow +from app.infrastructure.repositories.app_profile import AppProfileRepository +from app.infrastructure.repositories.engine import GateResultRepository +from tests.engine_fakes import FakeSandbox + + +@pytest_asyncio.fixture +async def task(session_factory, seeded) -> AsyncIterator[Task]: + async with session_factory() as session: + user = UserRow(email=f"g-{new_id()[:8]}@example.com", password_hash="x") + session.add(user) + await session.flush() + org = OrganizationRow( + name="سازمان", slug=f"g-{user.id[:6]}", owner_user_id=user.id + ) + session.add(org) + await session.flush() + row = TaskRow(organization_id=org.id, title="کار دِرِوازه") + session.add(row) + await session.flush() + await session.commit() + yield Task(id=row.id, organization_id=org.id, title=row.title) + + +@pytest_asyncio.fixture +async def runner(session_factory, task: Task) -> AsyncIterator[GateRunner]: + async with session_factory() as session: + yield GateRunner( + results=GateResultRepository(session), + feed=RunFeed(), + profiles=AppProfileRepository(session), + ) + + +# --------------------------------------------------------------------------- +# pure helpers +# --------------------------------------------------------------------------- + + +def test_diff_hash_is_stable_and_sensitive_to_content(): + assert diff_hash("same") == diff_hash("same") + assert diff_hash("same") != diff_hash("same!") + + +def test_forbidden_paths_flag_git_internals_and_ci(): + changed = [".git/config", ".github/workflows/ci.yml", ".gitlab-ci.yml", "app/main.py"] + assert forbidden_paths(changed) == [ + ".git/config", + ".github/workflows/ci.yml", + ".gitlab-ci.yml", + ] + + +def test_forbidden_paths_allow_ordinary_source_files(): + assert forbidden_paths(["app/main.py", "README.md", "tests/test_app.py"]) == [] + + +def test_code_task_excludes_analysis(): + assert GateRunner.code_task(TaskKind.FEATURE) is True + assert GateRunner.code_task(TaskKind.ANALYSIS) is False + + +# --------------------------------------------------------------------------- +# diff guard +# --------------------------------------------------------------------------- + + +async def test_diff_guard_fails_an_empty_code_diff(runner, task, session_factory): + outcome = await runner.diff_guard(task=task, diff=DiffResult(), code_task=True) + assert outcome.ok is False + assert outcome.sends_back is True + async with session_factory() as session: + row = await GateResultRepository(session).latest_for( + task.id, str(GateName.DIFF_GUARD) + ) + assert row is not None and row.ok is False + + +async def test_diff_guard_allows_an_empty_analysis_diff(runner, task): + outcome = await runner.diff_guard(task=task, diff=DiffResult(), code_task=False) + assert outcome.ok is True + + +async def test_diff_guard_flags_forbidden_paths_symlinks_and_big_files(runner, task): + diff = DiffResult( + changed_files=["app/main.py", ".github/workflows/ci.yml"], + escaping_symlinks=["link-to-outside"], + largest_file_bytes=MAX_FILE_BYTES + 1, + ) + outcome = await runner.diff_guard(task=task, diff=diff, code_task=True) + assert outcome.ok is False + assert outcome.sends_back is True + assert ".github" in outcome.detail + + +# --------------------------------------------------------------------------- +# no progress +# --------------------------------------------------------------------------- + + +async def test_no_progress_blocks_the_same_diff_twice(runner, task): + first, fingerprint, count = await runner.no_progress( + task=task, previous_hash="", same_count=0, diff_text="--- a\n+++ b\n" + ) + assert first.ok is True and count == 1 + + second, fingerprint2, count2 = await runner.no_progress( + task=task, + previous_hash=fingerprint, + same_count=count, + diff_text="--- a\n+++ b\n", + ) + assert second.ok is False and second.sends_back is True + assert fingerprint2 == fingerprint and count2 == 2 + + +async def test_no_progress_accepts_a_new_diff(runner, task): + _, first_hash, _ = await runner.no_progress( + task=task, previous_hash="", same_count=0, diff_text="a" + ) + outcome, _, count = await runner.no_progress( + task=task, previous_hash=first_hash, same_count=3, diff_text="b" + ) + assert outcome.ok is True and count == 1 + + +# --------------------------------------------------------------------------- +# tests gate +# --------------------------------------------------------------------------- + + +async def test_tests_gate_is_green_when_no_command_is_defined(runner, task): + outcome = await runner.tests(task=task, sandbox=FakeSandbox(), session_external_id="s") + assert outcome.ok is True + + +async def test_tests_gate_reports_red_with_the_failure_tail( + runner, task, session_factory +): + async with session_factory() as session: + project = ProjectRow( + id=new_id(), + organization_id=task.organization_id, + name="app", + slug="app", + test_command="pytest -q", + ) + session.add(project) + await session.commit() + project_id = project.id + task.project_id = project_id + + sandbox = FakeSandbox( + commands={ + "pytest": [ + CommandResult(exit_code=1, stdout="", stderr="AssertionError: boom") + ] + } + ) + outcome = await runner.tests(task=task, sandbox=sandbox, session_external_id="s") + assert outcome.ok is False and outcome.sends_back is True + assert "AssertionError" in outcome.detail + + +# --------------------------------------------------------------------------- +# green gate +# --------------------------------------------------------------------------- + + +async def test_green_gate_needs_a_passing_tests_row_or_a_waiver(runner, task): + blocked = await runner.green(task=task, waived=False, waiver_reason="") + assert blocked.ok is False and blocked.sends_back is True + + waived = await runner.green(task=task, waived=True, waiver_reason="known flake") + assert waived.ok is True + + +async def test_green_gate_is_green_once_tests_pass(runner, task, session_factory): + async with session_factory() as session: + await GateResultRepository(session).create( + GateResult( + task_id=task.id, + gate=GateName.TESTS, + ok=True, + detail="green", + ) + ) + await session.commit() + outcome = await runner.green(task=task, waived=False, waiver_reason="") + assert outcome.ok is True diff --git a/docs/plan/reports/agent-2-phase-4.md b/docs/plan/reports/agent-2-phase-4.md new file mode 100644 index 0000000..a34ba42 --- /dev/null +++ b/docs/plan/reports/agent-2-phase-4.md @@ -0,0 +1,197 @@ +# Agent 2 — Phase 4 report: the scheduler (one tick → one step → merged code) + +**Date:** 2026-10-14 · **Branch:** `dev` · **Status:** done, all checks green + +## What was done + +### 4.1 The tick — `app/application/engine/scheduler.py` (new) + +`Scheduler.tick()` is the engine's driver. One tick: + +1. reads every task whose stage **waits for a turn** (`TURN_STAGES`) and every task + ready to **merge** (`MERGE_STAGES = (merge_request,)`), +2. applies the selection caps, then +3. runs **one** step per selected task — a turn, or the platform's merge. + +A step never aborts the tick: `_guarded` catches, records and continues. A lost +optimistic race surfaces as `ConflictError` and is counted as *skipped*, because +"another worker moved it first" is a normal outcome, not a failure. + +**Selection caps, not execution caps.** `_select` honours +`engine_max_turns_per_org` (3) and `engine_max_concurrent_turns` (8) by *choosing* +that many tasks per tick. Turns themselves run one at a time inside a tick: an +`AsyncSession` is not safe to share across concurrent turns, and correctness +across worker processes comes from the optimistic `advance` (two workers ticking +the same task both try; exactly one wins). This is a deliberate deviation from a +literal reading of "run 8 turns at once" — see *Deviations*. + +`TickReport` (selected / acted / skipped / failed / merged / paused, decisions, +notes) is what the worker logs and the tests assert on. + +### 4.2 Worktree lifecycle + +`_worktree_for` calls `RepoWorkspacePort.create_task_worktree(project_id, task_id, +base="dev")` for any pipeline that `requires_worktree` (i.e. not `analysis_only`) +and a task with a `project_id`; otherwise `None`. The branch is saved onto the task +state (`engine.state` → `task_pipeline_state.branch`) the moment the developer's +diff is accepted, so a merged task records *what* it merged. + +The worktree is removed on **completion, cancellation and failure** (TODO 4.2): +`_merge_step` removes it after a successful push, and `_move` calls +`_release_worktree` whenever the target is `failed`/`cancelled`. It is *kept* +during `error_review`, so an `error_review` retry still has the code to move +forward. + +### 4.3 The gate chain — `app/application/engine/gates.py` (new) + +A `WORK_REPORT` is a claim, not evidence. Four gates check it, and **every verdict +is a `gate_results` row and a `task.gate` event** — a gate that only lives in a log +is unauditable. + +- **diff guard** — empty diff for a code task; `.git`/CI paths; symlinks escaping + the repo; files over `MAX_FILE_BYTES` (1 MiB). +- **no-progress** — hashes the diff; the same hash twice in a row means the agent + is spinning, so the task goes back to the developer (and eventually to the + manager via `max_cycles`). `no_progress` returns the new `(hash, count)`, which + the scheduler persists on the state. +- **tests** — the app profile's `test_command` runs in the sandbox with + `engine_tests_timeout_seconds`; the failure tail is stored and handed back. No + command defined → green (nothing to run is not a failure). +- **green** — the tester's `TEST_PASS` is only believed when the tests gate is + green, or when the tester **waives** it with a reason. A waiver is stored on the + state (`waiver_reason`) and in the gate row; it is never inferred. + +`GateRunner.record(task, outcome)` is the public entry the scheduler uses to write +a verdict it reasoned about itself (a merge conflict is a diff-guard verdict that +comes from git, not from hashing). + +`_escalate_after_repeated_red`: a **second** red `tests` row moves the developer +onto the strongest *granted* model (one escalation only; no stronger model granted +→ left alone). + +### 4.4 Commit / merge / push — the platform owns git + +The AI never runs git. After `final_review → COMPLETE` the task reaches +`merge_request`, and `_merge_step` does the git: + +1. `commit` in the worktree, author `" "` + (`ENGINE_COMMIT_EMAIL`, matching the delivery adapter), +2. `sync_with_base`, +3. `merge_into(branch, target="dev")`, +4. `push(branch="dev")` → `diff_guard` note, +5. `move → completed`, `remove_worktree`, `DeliveryHooks.on_branch_pushed(source="engine")`, +6. `MemoryPort.on_task_finished`. + +A merge conflict is retried by sending the task back; after +`engine_max_merge_conflicts` (3) the manager takes over (`error_review`). A failed +push is `ErrorKind.NETWORK` → `error_review`. + +### 4.5 Budget + +Before any turn, `_budget_reason` checks the org's wallet. A wallet **present** with +`balance <= 0` pauses the task (`Stage.PAUSED`, status `waiting_approval`) with the +Persian note `BUDGET_PAUSED_MESSAGE` — no turn is spent. A **missing** wallet means +no budget was ever configured, and a fresh org must not be frozen out of its first +task (so `get_for_org`, never `ensure`, and `None` is not a pause). An optional +per-task ceiling (`engine_max_task_credits > 0`) reads projected spend from +`TaskSpendRepository.spend_for_task` (a read-only join of the usage ledger to the +task's runs). + +### 4.6 Flow test — `tests/test_engine_flow.py` (new) + +The end-to-end test: real repositories, real scheduler, real manager, real turn +runner; only the model (scripted), the sandbox (scripted) and git +(`FakeRepoWorkspace`) are fakes. The happy path walks +`manager_review → analysis → implementation → test → final_review → merge → +completed` one tick at a time and asserts the stage, the coarse status, the gate +rows, the git calls (author, target, branch), the removal of the worktree, the +delivery hook and the memory callback. Eleven tests cover: the happy path; git +never flowing through the model; red tests → back → then green; a forbidden diff +never reaching git; a tester failure; a `TEST_PASS` with no green tests; the +`analysis_only` pipeline (no worktree ever); the budget pause; dependencies; a +transient provider failure retried then handed to the manager; and a merge conflict. + +### 4.7 Wiring — `container_engine.build_scheduler`, the worker, `.env.example` + +- `build_scheduler(session)` assembles the runner, manager, gate runner, git + workspace, sandbox, delivery hooks, wallet and per-task spend repos on the + caller's session — the same rule as the bootstrap (`container.py` is frozen, so + the engine wires itself in my file). +- `sandbox_session_for` reuses the **active** sandbox session of the task's project + (`_EngineSandboxSessions`). The engine does not create sandboxes; when there is + none, the tests gate goes red and sends the work back — never "merge untested + because there was no container". +- `workers/ai_runtime.py` now drives **both** engines in one pass: the legacy run + queue, then one `Scheduler.tick()`. A broken tick is recorded against the worker + and never looks like a clean pass. `__main__.py` defaults the `ai-runtime` + interval to `engine_tick_seconds`; the one-pass (cron) default is unchanged. +- `.env.example` gains the six new `ENGINE_*` settings. + +### Bug found and fixed by the tests + +`forbidden_paths` normalised with `path.lstrip("./")`, which strips **all** leading +`.` and `/` characters: `.git/config` became `git/config` and +`.github/workflows/ci.yml` became `github/...`, so the diff guard would have +silently allowed changes to git internals and CI configuration. It now strips a +single leading `./` prefix. Caught by `test_forbidden_paths_flag_git_internals_and_ci`. + +### Other support changes + +- `ManagerOutcome.error_kind` — the manager carries the runner's error vocabulary + so the scheduler can tell a transient provider failure from a permanent one + without re-classifying a string. +- `TaskSpendRepository` (`repositories/engine.py`) — `spend_for_task`, read-only. +- `Scheduler._patch` refreshes the caller's snapshot version after each write, so a + second patch in the same step locks against the version the first write produced + instead of silently losing the write to `StaleStageError`. +- `tests/engine_fakes.py`: `RecordingMemory` now records `record` and + `on_task_finished` (it previously returned `None` and asserted nothing). + +## Tests + +`tests/test_engine_gates.py` (new, **13 tests**): pure helpers (`diff_hash`, +`forbidden_paths`, `code_task`); the diff guard's four refusals; no-progress; the +tests gate with and without a command; the green gate's waiver rule. + +`tests/test_engine_flow.py` (new, **11 tests**): see 4.6. + +Together **+24 tests**, all through the real repositories. + +## Checks (PLAN-v2 §7.4) + +``` +cd apps/api && .venv/bin/python -m pytest -q → 746 passed, 1 skipped +cd apps/api && .venv/bin/ruff check . → All checks passed! +cd apps/api && .venv/bin/mypy app → Success: no issues found in 160 source files +``` + +No web files were touched; no migration was needed (Phase 1's `a2b3c4d5e6f2` already +has the engine tables). + +## Deviations & notes + +1. **Concurrency is a selection cap, not 8 live turns.** A tick runs at most one + turn per task, sequentially. An `AsyncSession` is not concurrency-safe, so + running several turns on one session in parallel would be a corruption bug, not + a speed feature. `engine_max_concurrent_turns` / `engine_max_turns_per_org` + therefore bound how many tasks are *chosen*, and multi-process safety comes from + the optimistic `advance`. Recorded because it is a deliberate reading of 4.1. +2. **The engine does not create sandbox workspaces.** It reuses the active session + of the task's project; creating one automatically is a product decision the plan + does not make. Absent session → tests gate red → work sent back. +3. **The commit author e-mail is fixed** (`platform@smart-org.local`) and matches the + delivery adapter; the *name* is the employee's, which is what a reviewer reads. +4. `no_progress`/diff verdicts are persisted on the task state + (`last_diff_hash`, `same_diff_count`), so progress is judged across ticks and + across worker processes, not within one process. + +## Open questions / CHANGE REQUESTs + +- None new. The org automation policy (`automation_enabled`) is still behind + `ModelGrantPort` as planned. + +## Next + +Phase 5 — the web side: the engine board, the task page (live log, gates, diff), +`/workforce`, `/dashboard` + `/traces`, `src/test/engine-*.test.tsx`, and the final +workstream report.