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

"""The manager AI: orders become tasks, tasks become team work.

This is the layer that turns a user's sentence into the engine's machinery, and
it is deliberately the *only* place that interprets the manager's tags:

* **3.1 chat** — the user talks to the manager; the manager replies and may
  emit ``CREATE_TASK`` (≤ 3), ``PAUSE_TASK`` / ``RESUME_TASK`` / ``CANCEL_TASK``
  and ``HIRE`` / ``UPDATE_EMPLOYEE`` / ``FIRE``. Tasks get a pipeline state at
  ``pending_manager_review`` and are linked to the thread. The prompt asks for
  **atomic tasks that do not touch the same files**, and the chat's model calls
  are billed like every other call.
* **3.2 team** — hiring picks a *granted* model for the requested neutral tier,
  respects the employee cap and the "last manager" rule, and publishes
  ``team.changed``. Users choosing from the UI give a role and a tier only;
  model identity never crosses the API.
* **3.3 triage** — ``ASSIGN`` (pipeline + assignee), ``REJECT`` or ``ASK_USER``.
  Assignee selection defaults to the **least busy** active employee with the
  role, and a name is matched fuzzily **inside the org only**.
* **3.4 final / error** — ``COMPLETE`` → ``merge_request``, ``REVISE`` → back to
  the developer with ``cycle + 1``; ``RESOLVE_ERROR`` retries with a hint,
  ``CANCEL`` ends the task.
* **3.5 knowledge** — ``KNOWLEDGE`` tags are recorded through ``MemoryPort``.
* **3.6 bootstrap** — a new org gets manager + planner + developer + tester on
  the strongest / balanced / fast granted tiers; with nothing granted the
  employees are created inactive and a system message says so. Idempotent.

Every query here is tenant-scoped: the manager of org A can never assign an
employee of org B, even by exact name.
"""

from __future__ import annotations

import contextlib
import time
from dataclasses import dataclass

from app.application.engine.events import emit_team_changed
from app.application.engine.prompt import TurnContext, TurnMessage, build_turn_prompt
from app.application.engine.protocol import (
    AssignTag,
    ParsedReply,
    TagName,
    assign_of,
    knowledge_items,
    parse_reply,
    task_drafts,
    team_changes,
)
from app.application.engine.runner import (
    BillingScope,
    TurnRequest,
    TurnResult,
    TurnRunner,
    call_model,
    classify_error,
)
from app.application.engine.state import advance
from app.application.ports import ModelProviderPort
from app.application.ports.memory import MemoryPort, NullMemory
from app.application.ports.model_grants import ModelGrantPort
from app.application.ports.realtime import emit_task_created
from app.core.config import settings
from app.core.errors import ConflictError, NotFoundError, ValidationError
from app.domain.workforce.engine import EmployeeProfile, ErrorKind
from app.domain.workforce.models import (
    Employee,
    EmployeeStatus,
    Message,
    MessageRole,
    RoleKey,
    Task,
    TaskStatus,
    Thread,
)
from app.domain.workforce.pipeline import (
    Mode,
    PipelineKey,
    Risk,
    Stage,
    TaskKind,
    TaskPipelineState,
    coarse_status,
    resolve_pipeline,
)
from app.infrastructure.repositories.ai_control import (
    AiModelRepository,
    ModelCredentialRepository,
    RateCardRepository,
)
from app.infrastructure.repositories.audit import UsageLedgerRepository
from app.infrastructure.repositories.billing import CreditWalletRepository
from app.infrastructure.repositories.engine import (
    EmployeeProfileRepository,
    PipelineStateRepository,
    StaleStageError,
)
from app.infrastructure.repositories.workforce import (
    EmployeeRepository,
    MessageRepository,
    TaskRepository,
    ThreadRepository,
)

#: How long a chat turn may take. Chat is interactive, so it is much shorter
#: than a task turn's budget.
CHAT_TURN_TIMEOUT_SECONDS = 120
#: How many times a chat reply may be corrected for protocol errors.
CHAT_MAX_ATTEMPTS = 2

#: The message a new organization sees when the owner has granted nothing yet.
NO_MODEL_GRANTED_MESSAGE = "منتظر تخصیص مدل توسط مالک سامانه"

#: The reply a zero-credit organization gets instead of a model call (prepaid, D-B).
CHAT_NO_CREDIT_MESSAGE = (
    "اعتبار سازمان صفر است؛ مدیر پس از افزودن اعتبار توسط مالک سامانه پاسخ می‌دهد."
)

#: Bootstrap tiers per role, in the plan's order (strongest / balanced / fast).
BOOTSTRAP_TIERS: dict[RoleKey, str] = {
    RoleKey.MANAGER: "strong",
    RoleKey.PLANNER: "balanced",
    RoleKey.DEVELOPER: "balanced",
    RoleKey.TESTER: "fast",
}
#: When a tier is not granted, fall back down this ladder rather than failing.
_TIER_FALLBACK: dict[str, tuple[str, ...]] = {
    "strong": ("strong", "balanced", "fast"),
    "balanced": ("balanced", "fast"),
    "fast": ("fast", "balanced"),
}

#: Roles a manager may hire through chat.
HIRABLE_ROLES: tuple[RoleKey, ...] = (
    RoleKey.MANAGER,
    RoleKey.PLANNER,
    RoleKey.DEVELOPER,
    RoleKey.REVIEWER,
    RoleKey.TESTER,
)

#: Persian labels, so a prompt can show a human-readable roster.
_ROLE_LABELS: dict[RoleKey, str] = {
    RoleKey.MANAGER: "مدیر",
    RoleKey.PLANNER: "برنامه‌ریز",
    RoleKey.DEVELOPER: "برنامه‌نویس",
    RoleKey.REVIEWER: "بازبین",
    RoleKey.TESTER: "آزمونگر",
    RoleKey.RESEARCHER: "پژوهشگر",
    RoleKey.CONTENT_CREATOR: "تولیدکنندهٔ محتوا",
    RoleKey.SUPPORT: "پشتیبان",
}

#: The system thread's title, where "no model granted" style notes live.
SYSTEM_THREAD_TITLE = "پیام‌های سامانه"


@dataclass(slots=True, kw_only=True)
class ManagerOutcome:
    """What a manager turn decided, in the engine's own words."""

    acted: bool
    decision: str = "idle"
    stage: Stage | None = None
    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:
    """Fuzzy name matching **inside one organization**.

    Exact match first, then a case-insensitive containment either way. Ties are
    resolved by the caller's ordering (least busy first), never by luck.
    """
    wanted = name.strip().lower()
    if not wanted:
        return None
    for candidate in candidates:
        if candidate.lower() == wanted:
            return candidate
    for candidate in candidates:
        if wanted in candidate.lower():
            return candidate
    for candidate in candidates:
        if candidate.lower() in wanted:
            return candidate
    return None


def _pick_granted_model(granted, tier: str, *, fallback: bool = True) -> str | None:
    """The granted model for a tier.

    ``fallback=False`` is the strict mode an explicit hire uses: the tier the
    caller asked for, or nothing. The default is the bootstrap's "whatever is
    granted" behaviour, where a degraded team beats no team.
    """
    for candidate in _TIER_FALLBACK.get(tier, (tier,)) if fallback else (tier,):
        for model in granted:
            if model.tier == candidate:
                return model.ai_model_id
    return None


def _bootstrap_name(role: RoleKey) -> str:
    return f"{_ROLE_LABELS.get(role, role.value)} یک"


class ManagerTurns:
    """The manager's voice: chat, triage, final verdict, error handling, team."""

    def __init__(
        self,
        *,
        provider: ModelProviderPort,
        employees: EmployeeRepository,
        messages: MessageRepository,
        tasks: TaskRepository,
        threads: ThreadRepository,
        models: AiModelRepository,
        credentials: ModelCredentialRepository,
        rate_cards: RateCardRepository,
        ledger: UsageLedgerRepository,
        grants: ModelGrantPort,
        memory: MemoryPort | None = None,
        profiles: EmployeeProfileRepository | None = None,
        states: PipelineStateRepository | None = None,
        runner: TurnRunner | None = None,
        feed=None,
        wallets: CreditWalletRepository | None = None,
    ) -> None:
        self.provider = provider
        self.employees = employees
        self.messages = messages
        self.tasks = tasks
        self.threads = threads
        self.models = models
        self.credentials = credentials
        self.rate_cards = rate_cards
        self.ledger = ledger
        self.grants = grants
        self.memory = memory or NullMemory()
        self.profiles = profiles
        self.states = states
        self.runner = runner
        self.feed = feed
        self.wallets = wallets

    # ------------------------------------------------------------------ #
    # 3.6 bootstrap                                                       #
    # ------------------------------------------------------------------ #

    async def bootstrap_team(self, *, organization_id: str, project_id: str) -> None:
        """Create the starting team of a new organization. Idempotent.

        With no granted model the employees still exist — inactive — so the
        organization sees its team and the reason it cannot work yet, rather
        than an empty page and a mystery.
        """
        if await self.employees.list_for_org(organization_id, limit=1):
            return
        granted = await self.grants.granted_models(organization_id=organization_id)
        for role in BOOTSTRAP_TIERS:
            model_id = _pick_granted_model(granted, BOOTSTRAP_TIERS[role])
            employee = await self.employees.create(
                Employee(
                    organization_id=organization_id,
                    name=_bootstrap_name(role),
                    role_key=role,
                    project_id=project_id,
                    status=(
                        EmployeeStatus.ACTIVE if model_id else EmployeeStatus.PAUSED
                    ),
                )
            )
            if self.profiles is not None:
                await self.profiles.upsert(
                    EmployeeProfile(
                        employee_id=employee.id,
                        ai_model_id=model_id,
                        roles=[role.value],
                        hired_by="manager",
                    )
                )
        if not granted:
            await self._post_system_message(
                organization_id=organization_id,
                project_id=project_id,
                content=NO_MODEL_GRANTED_MESSAGE,
            )

    # ------------------------------------------------------------------ #
    # 3.2 team management                                                 #
    # ------------------------------------------------------------------ #

    async def hire(
        self,
        *,
        organization_id: str,
        role: RoleKey,
        tier: str,
        name: str = "",
        project_id: str | None = None,
        hired_by: str = "manager",
    ) -> Employee:
        """Hire one employee on a granted model of the requested tier."""
        if role not in HIRABLE_ROLES:
            raise ValidationError("این نقش قابل استخدام نیست.")
        if not tier:
            raise ValidationError("سطح کارمند را انتخاب کنید.")
        # The owner's per-org policy may lower the global cap, never raise it.
        policy = await self.grants.get_org_policy(organization_id)
        cap = min(settings.engine_max_employees, policy.max_employees)
        count = await self.employees.count_for_org(organization_id)
        if count >= cap:
            raise ConflictError(f"سقف تعداد کارمندان ({cap}) پر شده است.")
        granted = await self.grants.granted_models(organization_id=organization_id)
        # A hire is an explicit request for a tier: no silent downgrade, or the
        # manager would think it hired a strong employee and got a fast one.
        model_id = _pick_granted_model(granted, tier, fallback=False)
        if model_id is None:
            raise ConflictError("مدلی از این سطح به سازمان شما اختصاص داده نشده است.")
        employee = await self.employees.create(
            Employee(
                organization_id=organization_id,
                name=(name or _bootstrap_name(role)).strip(),
                role_key=role,
                project_id=project_id or await self._project_id_for(organization_id),
            )
        )
        if self.profiles is not None:
            await self.profiles.upsert(
                EmployeeProfile(
                    employee_id=employee.id,
                    ai_model_id=model_id,
                    roles=[role.value],
                    hired_by=hired_by,
                )
            )
        await emit_team_changed(
            organization_id=organization_id, change="hired", employee_id=employee.id
        )
        return employee

    async def fire(self, *, organization_id: str, employee_id: str) -> Employee:
        """Archive one employee. The last manager may not be fired."""
        employee = await self.employees.get_for_org(organization_id, employee_id)
        if employee is None:
            raise NotFoundError("کارمند یافت نشد.")
        if employee.role_key is RoleKey.MANAGER:
            managers = [
                e
                for e in await self.employees.list_for_org(organization_id)
                if e.role_key is RoleKey.MANAGER and e.status is EmployeeStatus.ACTIVE
            ]
            if len(managers) <= 1:
                raise ConflictError("آخرین مدیر سازمان را نمی‌توان حذف کرد.")
        employee.status = EmployeeStatus.ARCHIVED
        updated = await self.employees.update(employee)
        if self.profiles is not None:
            await self.profiles.set_busy(employee_id=employee_id, task_id=None)
        await emit_team_changed(
            organization_id=organization_id, change="fired", employee_id=employee_id
        )
        return updated

    async def update_tier(
        self, *, organization_id: str, employee_id: str, tier: str
    ) -> Employee:
        """Re-point an employee at a granted model of another tier."""
        employee = await self.employees.get_for_org(organization_id, employee_id)
        if employee is None:
            raise NotFoundError("کارمند یافت نشد.")
        granted = await self.grants.granted_models(organization_id=organization_id)
        model_id = _pick_granted_model(granted, tier, fallback=False)
        if model_id is None:
            raise ConflictError("مدلی از این سطح به سازمان شما اختصاص داده نشده است.")
        if self.profiles is not None:
            await self.profiles.set_model(employee_id=employee_id, ai_model_id=model_id)
        await emit_team_changed(
            organization_id=organization_id, change="updated", employee_id=employee_id
        )
        return employee

    # ------------------------------------------------------------------ #
    # 3.1 manager chat                                                    #
    # ------------------------------------------------------------------ #

    async def chat(
        self,
        *,
        organization_id: str,
        thread_id: str,
        content: str,
        user_id: str,
        employee_id: str | None = None,
    ) -> tuple[Message, Message]:
        """The manager hears the order, answers it, and may act on it.

        The reply the user reads is the prose *outside* the tags; the tags are
        executed beside it. That separation is what lets one message create a
        task and still read like a conversation.
        """
        thread = await self.threads.get_for_org(organization_id, thread_id)
        if thread is None:
            raise NotFoundError("گفتگو یافت نشد.")
        if not content.strip():
            raise ValidationError("متن پیام نمی‌تواند خالی باشد.")

        user_message = await self.messages.create(
            Message(thread_id=thread.id, role=MessageRole.USER, content=content)
        )
        # Prepaid: a reply costs a model call, so an org with no credit gets the
        # note instead. No model is called and no tag is executed.
        if await self._out_of_credit(organization_id):
            note = await self.messages.create(
                Message(
                    thread_id=thread.id,
                    role=MessageRole.ASSISTANT,
                    content=CHAT_NO_CREDIT_MESSAGE,
                )
            )
            return user_message, note
        employee = await self._manager_of(organization_id, employee_id)
        extra = "\n\n".join(
            part
            for part in (
                await self._roster_text(organization_id),
                await self._tasks_text(organization_id),
                await self._conversation_text(thread),
            )
            if part
        )

        reply_text = ""
        parsed: ParsedReply | None = None
        deadline = time.monotonic() + CHAT_TURN_TIMEOUT_SECONDS
        context = self._chat_context(
            organization_id=organization_id,
            thread=thread,
            content=content,
            extra=extra,
            history=[],
        )
        for _attempt in range(CHAT_MAX_ATTEMPTS):
            model = await self._model_for(employee, organization_id)
            if model is None:
                await self._system_note(
                    thread, "مدیری با مدل فعلاً در دسترس نیست."
                )
                break
            try:
                response = await call_model(
                    provider=self.provider,
                    credentials=self.credentials,
                    rate_cards=self.rate_cards,
                    ledger=self.ledger,
                    model=model,
                    prompt=build_turn_prompt(context),
                    scope=BillingScope(
                        organization_id=organization_id,
                        project_id=thread.project_id,
                        employee_id=employee.id if employee else None,
                        user_id=user_id,
                    ),
                )
            except Exception as exc:  # noqa: BLE001 - chat must always answer
                await self._system_note(
                    thread, f"مدیر موقتاً پاسخ نمی‌دهد ({classify_error(exc).value})."
                )
                break
            reply_text = response.content
            parsed = parse_reply(
                response.content, role=RoleKey.MANAGER, mode=Mode.CHAT
            )
            if not parsed.errors or parsed.tags:
                break
            # A second chance, with the protocol errors quoted back to it.
            context = self._chat_context(
                organization_id=organization_id,
                thread=thread,
                content=content,
                extra=extra,
                history=[TurnMessage(kind="result", text="\n".join(parsed.errors))],
            )
            if time.monotonic() > deadline:
                break

        actions: list[str] = []
        if parsed is not None and parsed.tags:
            actions = await self._dispatch_chat_tags(
                parsed=parsed,
                organization_id=organization_id,
                thread=thread,
                user_id=user_id,
            )
        await self._record_knowledge(parsed, organization_id=organization_id)

        body = (parsed.prose if parsed else "") or reply_text
        if actions:
            body = f"{body}\n\n" + "\n".join(f"• {action}" for action in actions)
        assistant_message = await self.messages.create(
            Message(
                thread_id=thread.id,
                role=MessageRole.ASSISTANT,
                content=body.strip() or "دستور دریافت شد.",
                employee_id=employee.id if employee else None,
            )
        )
        return user_message, assistant_message

    async def _out_of_credit(self, organization_id: str) -> bool:
        """True when a wired wallet shows no credit left (a missing wallet counts as none)."""
        if self.wallets is None:
            return False
        wallet = await self.wallets.get_for_org(organization_id)
        return wallet is None or wallet.balance <= 0

    def _chat_context(
        self,
        *,
        organization_id: str,
        thread: Thread,
        content: str,
        extra: str,
        history: list[TurnMessage],
    ) -> TurnContext:
        return TurnContext(
            organization_id=organization_id,
            task_id=thread.id,
            project_id=thread.project_id,
            task_title="گفتگو با مدیر سازمان",
            task_description=content,
            stage=Stage.PENDING_MANAGER_REVIEW,
            role=RoleKey.MANAGER,
            mode=Mode.CHAT,
            stage_context=extra,
            history=history,
        )

    async def _dispatch_chat_tags(
        self,
        *,
        parsed: ParsedReply,
        organization_id: str,
        thread: Thread,
        user_id: str,
    ) -> list[str]:
        """Execute the chat's tags, returning one Persian line per action."""
        actions: list[str] = []
        for draft in task_drafts(parsed):
            kind = (
                TaskKind(draft.kind)
                if draft.kind in {k.value for k in TaskKind}
                else TaskKind.FEATURE
            )
            # A thread made by the web UI has no project; the org's project is the fallback.
            project_id = thread.project_id or await self._project_id_for(organization_id)
            task = await self.tasks.create(
                Task(
                    organization_id=organization_id,
                    title=draft.title[:200],
                    description=draft.description,
                    project_id=project_id,
                    thread_id=thread.id,
                    created_by_user_id=user_id,
                )
            )
            if self.states is not None:
                await self.states.create(
                    TaskPipelineState(
                        task_id=task.id,
                        organization_id=organization_id,
                        project_id=project_id,
                        pipeline=resolve_pipeline(
                            manager_choice=draft.pipeline or None,
                            task_kind=kind,
                            has_active_reviewer=await self._has_active_reviewer(organization_id),
                        ),
                        kind=kind,
                        risk=Risk.MEDIUM,
                        stage=Stage.PENDING_MANAGER_REVIEW,
                        depends_on=list(draft.depends_on),
                    )
                )
            await emit_task_created(
                task_id=task.id,
                organization_id=organization_id,
                title=task.title,
                status=TaskStatus.NEW.value,
            )
            actions.append(f"کار ساخته شد: {task.title}")

        for change in team_changes(parsed):
            try:
                if change.action is TagName.HIRE:
                    role = (
                        RoleKey(change.role)
                        if change.role in {r.value for r in RoleKey}
                        else RoleKey.DEVELOPER
                    )
                    employee = await self.hire(
                        organization_id=organization_id,
                        role=role,
                        tier=change.tier or "balanced",
                        name=change.name,
                        project_id=thread.project_id,
                    )
                    actions.append(f"کارمند استخدام شد: {employee.name}")
                elif change.action in (TagName.FIRE, TagName.UPDATE_EMPLOYEE):
                    employee_id = change.employee_id or await self._employee_id_by_name(
                        organization_id, change.name
                    )
                    if change.action is TagName.FIRE:
                        employee = await self.fire(
                            organization_id=organization_id, employee_id=employee_id
                        )
                        actions.append(f"کارمند حذف شد: {employee.name}")
                    else:
                        await self.update_tier(
                            organization_id=organization_id,
                            employee_id=employee_id,
                            tier=change.tier or "balanced",
                        )
                        actions.append("سطح کارمند تغییر کرد.")
            except (ConflictError, NotFoundError, ValidationError) as exc:
                actions.append(str(exc))

        for tag in parsed.tags:
            if tag.name not in (TagName.PAUSE_TASK, TagName.RESUME_TASK, TagName.CANCEL_TASK):
                continue
            task_id = tag.attr("task_id", "").strip()
            state = await self._state_for(organization_id, task_id)
            if state is None or self.states is None:
                actions.append("کاری با این شناسه پیدا نشد.")
                continue
            if tag.name is TagName.CANCEL_TASK:
                target = Stage.CANCELLED
                done = "کار لغو شد."
            elif tag.name is TagName.PAUSE_TASK:
                target = Stage.PAUSED
                done = "کار متوقف شد."
            else:
                target = Stage.PENDING_MANAGER_REVIEW
                done = "کار از سر گرفته شد."
            try:
                await advance(
                    self.states,
                    task_id=task_id,
                    expected_from=state.stage,
                    to=target,
                )
            except ConflictError as exc:
                actions.append(str(exc))
                continue
            if self.tasks is not None:
                await self.tasks.set_status(task_id, coarse_status(target))
            actions.append(done)
        return actions

    # ------------------------------------------------------------------ #
    # 3.3 triage                                                          #
    # ------------------------------------------------------------------ #

    async def triage(
        self,
        *,
        task: Task,
        run_id: str,
        employee: Employee,
        stage_context: str = "",
    ) -> ManagerOutcome:
        """The manager's triage turn: assign, reject, or ask the user."""
        roster = await self._roster_text(task.organization_id)
        result = await self._run(
            await self._request(
                task=task,
                employee=employee,
                role=RoleKey.MANAGER,
                mode=Mode.TRIAGE,
                stage=Stage.PENDING_MANAGER_REVIEW,
                run_id=run_id,
                extra_context="\n\n".join(p for p in (stage_context, roster) if p),
            ),
            employee,
        )
        if not result.ok:
            return ManagerOutcome(
                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)
        if assign is not None:
            return await self._apply_assign(task, assign)
        reject = parsed.first(TagName.REJECT)
        if reject is not None:
            await self._move(task, Stage.CANCELLED)
            return ManagerOutcome(
                acted=True,
                decision="rejected",
                stage=Stage.CANCELLED,
                message=reject.attr("reason", ""),
            )
        ask = parsed.first(TagName.ASK_USER)
        if ask is not None:
            await self._move(task, Stage.PENDING_USER_DECISION)
            if task.thread_id:
                await self.messages.create(
                    Message(
                        thread_id=task.thread_id,
                        role=MessageRole.ASSISTANT,
                        content=ask.body.strip(),
                        employee_id=employee.id,
                    )
                )
            return ManagerOutcome(
                acted=True,
                decision="asked_user",
                stage=Stage.PENDING_USER_DECISION,
                message=ask.body.strip(),
            )
        return ManagerOutcome(
            acted=False, stage=Stage.PENDING_MANAGER_REVIEW, message=result.reply_text
        )

    async def _apply_assign(self, task: Task, tag) -> ManagerOutcome:
        """Apply an ``ASSIGN``: pick the assignee and the pipeline."""
        try:
            assign: AssignTag = assign_of(tag)
        except ValueError as exc:
            return ManagerOutcome(
                acted=False, message=str(exc), stage=Stage.PENDING_MANAGER_REVIEW
            )
        assignee = await self._pick_assignee(
            task.organization_id, assign.employee or ""
        )
        if assignee is None:
            return ManagerOutcome(
                acted=False,
                decision="idle",
                message="کارمند مناسبی برای این کار پیدا نشد.",
                stage=Stage.PENDING_MANAGER_REVIEW,
            )
        state = await self._state_for(task.organization_id, task.id)
        pipeline = resolve_pipeline(
            manager_choice=assign.pipeline or None,
            task_kind=state.kind if state else None,
            has_active_reviewer=await self._has_active_reviewer(task.organization_id),
        )
        first_stage = (
            Stage.PENDING_ANALYSIS
            if pipeline is not PipelineKey.CODE_TEST
            else Stage.PENDING_IMPLEMENTATION
        )
        await self.tasks.set_assignee(task.id, assignee.id)
        if state is not None and self.states is not None:
            # A lost race here is fine: the advance below still moves the task,
            # and the pipeline column is a hint the next triage re-reads anyway.
            with contextlib.suppress(StaleStageError):
                await self.states.patch(
                    task_id=task.id,
                    expected_version=state.version,
                    pipeline=str(pipeline),
                )
        await self._move(task, first_stage)
        return ManagerOutcome(
            acted=True,
            decision="assigned",
            stage=first_stage,
            employee_id=assignee.id,
            pipeline=pipeline,
        )

    # ------------------------------------------------------------------ #
    # 3.4 final verdict and error handling                                #
    # ------------------------------------------------------------------ #

    async def final_review(
        self,
        *,
        task: Task,
        run_id: str,
        employee: Employee,
        spec: str = "",
        diff_text: str = "",
        test_output: str = "",
        gate_summary: str = "",
    ) -> ManagerOutcome:
        """The manager's final verdict: merge or send it back."""
        extra = "\n\n".join(p for p in (spec, diff_text, test_output, gate_summary) if p)
        result = await self._run(
            await self._request(
                task=task,
                employee=employee,
                role=RoleKey.MANAGER,
                mode=Mode.FINAL,
                stage=Stage.PENDING_FINAL_REVIEW,
                run_id=run_id,
                extra_context=extra,
            ),
            employee,
        )
        if not result.ok:
            return ManagerOutcome(
                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):
            await self._move(task, Stage.MERGE_REQUEST)
            return ManagerOutcome(
                acted=True, decision="completed", stage=Stage.MERGE_REQUEST
            )
        revise = parsed.first(TagName.REVISE)
        if revise is not None:
            state = await self._state_for(task.organization_id, task.id)
            if state is not None and self.states is not None:
                await self.states.patch(
                    task_id=task.id, expected_version=state.version, cycle=state.cycle + 1
                )
            await self._move(task, Stage.PENDING_IMPLEMENTATION)
            return ManagerOutcome(
                acted=True,
                decision="revised",
                stage=Stage.PENDING_IMPLEMENTATION,
                message=revise.body.strip(),
            )
        return ManagerOutcome(
            acted=False, stage=Stage.PENDING_FINAL_REVIEW, message=result.reply_text
        )

    async def error_review(
        self,
        *,
        task: Task,
        run_id: str,
        employee: Employee,
        error_summary: str,
        hint: str = "",
    ) -> ManagerOutcome:
        """The manager's error turn: retry with a hint, or give up."""
        detail = f"خطا:\n{error_summary}"
        if hint:
            detail += f"\n\nراهنمایی قبلی: {hint}"
        result = await self._run(
            await self._request(
                task=task,
                employee=employee,
                role=RoleKey.MANAGER,
                mode=Mode.ERROR,
                stage=Stage.ERROR_REVIEW,
                run_id=run_id,
                extra_context=detail,
            ),
            employee,
        )
        if not result.ok:
            return ManagerOutcome(
                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)
        if resolve is not None:
            await self._move(task, Stage.PENDING_IMPLEMENTATION)
            return ManagerOutcome(
                acted=True,
                decision="resolved",
                stage=Stage.PENDING_IMPLEMENTATION,
                message=resolve.attr("hint", ""),
            )
        cancel = parsed.first(TagName.CANCEL)
        if cancel is not None:
            await self._move(task, Stage.CANCELLED)
            return ManagerOutcome(
                acted=True,
                decision="cancelled",
                stage=Stage.CANCELLED,
                message=cancel.attr("reason", ""),
            )
        return ManagerOutcome(
            acted=False, stage=Stage.ERROR_REVIEW, message=result.reply_text
        )

    # ------------------------------------------------------------------ #
    # internals                                                           #
    # ------------------------------------------------------------------ #

    async def _run(self, request: TurnRequest, employee: Employee) -> TurnResult:
        if self.runner is None:
            return TurnResult(
                ok=False,
                error_kind=ErrorKind.UNAVAILABLE.value,
                error="موتور اجرا در دسترس نیست.",
            )
        return await self.runner.run_turn(request)

    async def _request(
        self,
        *,
        task: Task,
        employee: Employee,
        role: RoleKey,
        mode: Mode,
        stage: Stage,
        run_id: str,
        extra_context: str = "",
    ) -> TurnRequest:
        return TurnRequest(
            organization_id=task.organization_id,
            task=task,
            role=role,
            mode=mode,
            stage=stage,
            run_id=run_id,
            employee=employee,
            model_id=await self._model_id_for(employee, task.organization_id),
            extra_context=extra_context,
        )

    async def _model_id_for(
        self, employee: Employee | None, organization_id: str
    ) -> str | None:
        """The model id an employee runs on: its profile, or a granted tier."""
        if employee is None:
            return None
        model_id = getattr(employee, "model_id", None)
        if not model_id and self.profiles is not None:
            profile = await self.profiles.get(employee.id)
            model_id = profile.ai_model_id if profile else None
        if not model_id:
            granted = await self.grants.granted_models(organization_id=organization_id)
            model_id = _pick_granted_model(
                granted, BOOTSTRAP_TIERS.get(employee.role_key, "balanced")
            )
        return model_id

    async def _model_for(self, employee: Employee | None, organization_id: str):
        """Resolve the model an employee runs on, or ``None``.

        A manager without a granted model cannot answer: the caller says so in
        the thread rather than raising into the user's face.
        """
        if employee is None:
            return None
        model_id = await self._model_id_for(employee, organization_id)
        if not model_id:
            return None
        if not await self.grants.is_granted(
            organization_id=organization_id, ai_model_id=model_id
        ):
            return None
        return await self.models.get(model_id)

    async def _manager_of(
        self, organization_id: str, employee_id: str | None
    ) -> Employee | None:
        if employee_id:
            return await self.employees.get_for_org(organization_id, employee_id)
        for employee in await self.employees.list_for_org(organization_id):
            if employee.role_key is RoleKey.MANAGER and employee.is_available:
                return employee
        return None

    async def _pick_assignee(
        self, organization_id: str, wanted: str
    ) -> Employee | None:
        """The employee who takes the task: least busy, or the named one.

        The name is matched **inside this organization only** — a name from
        another tenant resolves to nothing, which is the isolation rule the
        tests enforce.
        """
        roster = [
            e
            for e in await self.employees.list_for_org(organization_id)
            if e.is_available and e.role_key is not RoleKey.MANAGER
        ]
        if not roster:
            return None
        if wanted:
            match = _fuzzy_match(wanted, [e.name for e in roster])
            for employee in roster:
                if employee.name == match:
                    return employee
            return None
        loads = [(await self.tasks.count_open_for_employee(e.id), e) for e in roster]
        loads.sort(key=lambda pair: pair[0])
        return loads[0][1]

    async def _employee_id_by_name(self, organization_id: str, name: str) -> str:
        roster = await self.employees.list_for_org(organization_id)
        match = _fuzzy_match(name, [e.name for e in roster])
        for employee in roster:
            if employee.name == match:
                return employee.id
        raise NotFoundError("کارمندی با این نام پیدا نشد.")

    async def _roster_text(self, organization_id: str) -> str:
        """The team as prompt text: names and roles, never model identity."""
        lines = ["تیم سازمان:"]
        for employee in await self.employees.list_for_org(organization_id):
            status = "فعال" if employee.is_available else "غیرفعال"
            label = _ROLE_LABELS.get(employee.role_key, employee.role_key.value)
            lines.append(f"- {employee.name} ({label}) — {status}")
        return "\n".join(lines)

    async def _tasks_text(self, organization_id: str) -> str:
        """The organization's open tasks, so the manager can act on them by id."""
        lines = ["کارهای باز:"]
        for task in await self.tasks.list_for_org(organization_id, limit=20):
            state = await self._state_for(organization_id, task.id)
            stage = state.stage.value if state else task.status.value
            lines.append(f"- {task.id}: {task.title} [{stage}]")
        return "\n".join(lines)

    async def _conversation_text(self, thread: Thread) -> str:
        history = await self.messages.list_for_thread(thread.id)
        lines = [
            f"{m.role.value}: {m.content}"
            for m in history[-12:]
            if m.role is not MessageRole.SYSTEM
        ]
        return ("گفتگو تا کنون:\n" + "\n".join(lines)) if lines else ""

    async def _state_for(self, organization_id: str, task_id: str) -> TaskPipelineState | None:
        if self.states is None:
            return None
        state = await self.states.get(task_id)
        if state is None or state.organization_id != organization_id:
            return None
        return state

    async def _move(self, task: Task, to: Stage) -> None:
        """Move the pipeline state and keep the coarse status truthful."""
        state = await self._state_for(task.organization_id, task.id)
        if state is not None and self.states is not None:
            try:
                await advance(
                    self.states, task_id=task.id, expected_from=state.stage, to=to
                )
            except ConflictError:
                return
        await self.tasks.set_status(task.id, coarse_status(to))

    async def _record_knowledge(
        self, parsed: ParsedReply | None, *, organization_id: str
    ) -> None:
        """3.5: a manager's KNOWLEDGE tags become org memory."""
        if parsed is None:
            return
        for item in knowledge_items(parsed):
            await self.memory.record(
                organization_id=organization_id,
                kind=item.kind if item.kind in ("rule", "fact") else "rule",
                content=item.content,
                source="manager",
                key=item.key or None,
            )

    async def _post_system_message(
        self, *, organization_id: str, project_id: str | None, content: str
    ) -> None:
        thread = await self._system_thread(organization_id, project_id)
        await self.messages.create(
            Message(thread_id=thread.id, role=MessageRole.SYSTEM, content=content)
        )

    async def _system_thread(
        self, organization_id: str, project_id: str | None
    ) -> Thread:
        for thread in await self.threads.list_for_org(organization_id, limit=20):
            if thread.title == SYSTEM_THREAD_TITLE:
                return thread
        return await self.threads.create(
            Thread(
                organization_id=organization_id,
                title=SYSTEM_THREAD_TITLE,
                project_id=project_id,
            )
        )

    async def _system_note(self, thread: Thread, content: str) -> None:
        await self.messages.create(
            Message(thread_id=thread.id, role=MessageRole.SYSTEM, content=content)
        )

    async def _project_id_for(self, organization_id: str) -> str | None:
        for employee in await self.employees.list_for_org(organization_id, limit=1):
            return employee.project_id
        return None

    async def _has_active_reviewer(self, organization_id: str) -> bool:
        """A review stage needs an active reviewer to act on it, or the task dead-ends."""
        return any(
            e.role_key is RoleKey.REVIEWER and e.is_available
            for e in await self.employees.list_for_org(organization_id)
        )


__all__ = [
    "BOOTSTRAP_TIERS",
    "CHAT_NO_CREDIT_MESSAGE",
    "CHAT_TURN_TIMEOUT_SECONDS",
    "HIRABLE_ROLES",
    "NO_MODEL_GRANTED_MESSAGE",
    "SYSTEM_THREAD_TITLE",
    "ManagerOutcome",
    "ManagerTurns",
]