smart-org/apps/api/app/application/engine/scheduler.py

"""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.model_grants import OrgPolicyView
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)

        # One policy read per organization for the whole tick, shared by both lists.
        policies: dict[str, OrgPolicyView] = {}
        turn_states = await self._drop_switched_off(turn_states, policies, report)
        merge_states = await self._drop_switched_off(merge_states, policies, report)

        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 _drop_switched_off(
        self,
        states: list[TaskPipelineState],
        policies: dict[str, OrgPolicyView],
        report: TickReport,
    ) -> list[TaskPipelineState]:
        """Keep only the states whose organization has automation switched on.

        The owner's switch stops everything up to the merge. A skipped task keeps
        its stage, so switching automation back on resumes it where it was.
        """
        kept: list[TaskPipelineState] = []
        for state in states:
            if state.organization_id not in policies:
                policies[state.organization_id] = await self.manager.grants.get_org_policy(
                    state.organization_id
                )
            if not policies[state.organization_id].automation_enabled:
                report.skipped += 1
                report.note(f"task {state.task_id}: اتوماسیون این سازمان خاموش است")
                continue
            kept.append(state)
        return kept

    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:
        """Why this task may not spend a turn now, or ``None`` when it may.

        Usage is post-paid: the ledger feeds invoices and nothing debits a wallet
        from usage. A zero balance is therefore the normal starting state of every
        org, not "out of money", so only an overdrawn wallet (below zero) pauses
        work. The owner's spend lever is the org's monthly credit budget.
        """
        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. A zero balance is not a pause either.
            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 "سقف اعتبار این کار پر شده است؛ کار متوقف شد."
        if self.spend is not None:
            policy = await self.manager.grants.get_org_policy(task.organization_id)
            budget = policy.monthly_credit_budget
            if budget is not None:
                # The month is counted in UTC, the clock the ledger stamps rows with.
                month_start = utcnow().replace(day=1, hour=0, minute=0, second=0, microsecond=0)
                spent = await self.spend.spend_for_org_since(task.organization_id, month_start)
                if spent >= budget:
                    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",
]