diff --git a/apps/api/app/application/container_engine.py b/apps/api/app/application/container_engine.py index 1fdc496..f0161a3 100644 --- a/apps/api/app/application/container_engine.py +++ b/apps/api/app/application/container_engine.py @@ -1,17 +1,73 @@ """Wiring for the AI-workforce workstream (team bootstrap). -Foundation default: a no-op. The engine agent replaces the body with the real bootstrap; -the function name and return type are FROZEN. +The org wizard calls ``build_workforce_bootstrap(session)`` after it commits a +new organization, then ``bootstrap_team`` creates the starting team. The team +is built on the *caller's* session, so the employees and the "waiting for a +model" note land in the same transaction as the organization. """ from __future__ import annotations from sqlalchemy.ext.asyncio import AsyncSession -from app.application.ports.workforce_bootstrap import NullWorkforceBootstrap, WorkforceBootstrapPort +from app.application.ports.workforce_bootstrap import WorkforceBootstrapPort -_null_bootstrap = NullWorkforceBootstrap() + +class _EngineWorkforceBootstrap: + """The real bootstrap: one manager, planner, developer and tester. + + Everything is built here, lazily, from the session: ``container.py`` is a + frozen composition root and the engine must not widen it, so the repos and + ports the manager needs are assembled where they are used. + """ + + def __init__(self, session: AsyncSession) -> None: + self.session = session + + async def bootstrap_team(self, *, organization_id: str, project_id: str) -> None: + from app.application.container import get_model_provider + from app.application.container_org import build_memory, build_model_grants + 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, + ) + + session = self.session + manager = ManagerTurns( + provider=get_model_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=build_model_grants(session), + memory=build_memory(session), + profiles=EmployeeProfileRepository(session), + states=PipelineStateRepository(session), + ) + await manager.bootstrap_team( + organization_id=organization_id, project_id=project_id + ) def build_workforce_bootstrap(session: AsyncSession) -> WorkforceBootstrapPort: - return _null_bootstrap + return _EngineWorkforceBootstrap(session) + + +__all__ = ["build_workforce_bootstrap"] diff --git a/apps/api/app/application/engine/events.py b/apps/api/app/application/engine/events.py index 78f7344..bcb309f 100644 --- a/apps/api/app/application/engine/events.py +++ b/apps/api/app/application/engine/events.py @@ -186,4 +186,22 @@ class RunFeed: return stream -__all__ = ["RunFeed"] +async def emit_team_changed( + *, organization_id: str, change: str, employee_id: str +) -> None: + """The team changed (hired, fired, re-tiered): the workforce page reloads.""" + await publish( + RealtimeEvent( + type="team.changed", + channels=(org_channel(organization_id),), + organization_id=organization_id, + data={ + "change": change, + "employee_id": employee_id, + "organization_id": organization_id, + }, + ) + ) + + +__all__ = ["RunFeed", "emit_team_changed"] diff --git a/apps/api/app/application/engine/manager.py b/apps/api/app/application/engine/manager.py new file mode 100644 index 0000000..cab85a0 --- /dev/null +++ b/apps/api/app/application/engine/manager.py @@ -0,0 +1,1067 @@ +"""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.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 = "منتظر تخصیص مدل توسط مالک سامانه" + +#: 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 + + +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, + ) -> 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 + + # ------------------------------------------------------------------ # + # 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("سطح کارمند را انتخاب کنید.") + count = await self.employees.count_for_org(organization_id) + if count >= settings.engine_max_employees: + raise ConflictError( + f"سقف تعداد کارمندان ({settings.engine_max_employees}) پر شده است." + ) + 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) + ) + 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 + + 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 + ) + task = await self.tasks.create( + Task( + organization_id=organization_id, + title=draft.title[:200], + description=draft.description, + project_id=thread.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=thread.project_id, + pipeline=resolve_pipeline( + manager_choice=draft.pipeline or None, task_kind=kind + ), + 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 + ) + 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, + ) + 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 + ) + 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 + ) + 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 + + +__all__ = [ + "BOOTSTRAP_TIERS", + "CHAT_TURN_TIMEOUT_SECONDS", + "HIRABLE_ROLES", + "NO_MODEL_GRANTED_MESSAGE", + "SYSTEM_THREAD_TITLE", + "ManagerOutcome", + "ManagerTurns", +] diff --git a/apps/api/app/application/engine/prompt.py b/apps/api/app/application/engine/prompt.py index b8cadb6..b419d86 100644 --- a/apps/api/app/application/engine/prompt.py +++ b/apps/api/app/application/engine/prompt.py @@ -59,6 +59,8 @@ _PROTOCOL_HINTS: dict[tuple[RoleKey, Mode], str] = { "تو مدیر هستی و کار را ارزیابی می‌کنی.\n" "برچسب‌های مجاز:\n" '[[ASSIGN employee="نام کارمند" pipeline="plan_code_test"/]]\n' + 'اگر خودت باید کارمند را انتخاب کنی، بنویس [[ASSIGN employee="" ' + 'pipeline="plan_code_test"/]] تا کم‌مشغول‌ترین انتخاب شود.\n' "[[REJECT reason=\"دلیل\"/]]\n" "[[ASK_USER]]سؤالت را اینجا بنویس[[/ASK_USER]]\n" '[[KNOWLEDGE kind="rule" key="..."]]قاعده‌ای که یاد گرفتی[[/KNOWLEDGE]]\n' @@ -92,7 +94,8 @@ _PROTOCOL_HINTS: dict[tuple[RoleKey, Mode], str] = { '[[KNOWLEDGE kind="rule|fact" key="..."]' "]]متن[[/KNOWLEDGE]]\n" "پاسخ متنی‌ات برای کاربر نمایش داده می‌شود؛ برچسب‌ها کنارش اجرا می‌شوند. " - "کارها را اتمیک بساز و نگو که کدام فایل‌ها را هم‌زمان تغییر می‌دهند." + "کارها را اتمیک بساز و طوری تقسیم کن که دو کار هم‌زمان فایل‌های یکسان را " + "تغییر ندهند." ), (RoleKey.PLANNER, Mode.WORK): ( "تو برنامه‌ریز هستی.\n" diff --git a/apps/api/app/application/engine/protocol.py b/apps/api/app/application/engine/protocol.py index 5504547..26ba56d 100644 --- a/apps/api/app/application/engine/protocol.py +++ b/apps/api/app/application/engine/protocol.py @@ -104,6 +104,7 @@ BODY_TAGS: frozenset[TagName] = frozenset( TagName.EDIT_FILE, TagName.KNOWLEDGE, TagName.ASK_USER, + TagName.REVISE, TagName.CREATE_TASK, TagName.SPLIT, TagName.PLAN_SPEC, @@ -282,14 +283,15 @@ _EDIT_BLOCK_RE = re.compile( def assign_of(tag: Tag) -> AssignTag: """Read an ``ASSIGN`` tag. ``employee=""`` means "pick for me".""" employee = tag.attr("employee", "").strip() - if not employee and not tag.attr("task_id", "").strip(): + task_id = tag.attr("task_id", "").strip() + # ``employee=""`` is a deliberate "pick the least busy one"; a tag with + # neither attribute at all is a mistake we quote back to the model. + if "employee" not in tag.attrs and "task_id" not in tag.attrs: raise ProtocolError("[[ASSIGN]] needs employee=... or task_id=...") pipeline = tag.attr("pipeline", "").strip() if pipeline and pipeline not in ASSIGNABLE_PIPELINES: raise ProtocolError(f"unknown pipeline: {pipeline}") - return AssignTag( - employee=employee, pipeline=pipeline, task_id=tag.attr("task_id", "").strip() - ) + return AssignTag(employee=employee, pipeline=pipeline, task_id=task_id) def task_draft_of(tag: Tag) -> TaskDraft: diff --git a/apps/api/app/application/engine/runner.py b/apps/api/app/application/engine/runner.py index 9fcca0f..011f83e 100644 --- a/apps/api/app/application/engine/runner.py +++ b/apps/api/app/application/engine/runner.py @@ -46,7 +46,7 @@ from app.application.engine.prompt import ( ) from app.application.engine.protocol import TOOL_TAGS, Tag, TagName, parse_reply from app.application.engine.tools import ToolExecutor, ToolOutcome, tool_key_for -from app.application.ports import ModelProviderPort +from app.application.ports import ModelProviderPort, ModelResponse from app.application.ports.memory import MemoryPort, NullMemory from app.application.ports.model_grants import ModelGrantPort from app.application.ports.sandbox_provider import SandboxProviderPort @@ -179,6 +179,62 @@ def classify_error(exc: BaseException) -> ErrorKind: return ErrorKind.UNAVAILABLE +@dataclass(slots=True, kw_only=True) +class BillingScope: + """What a model call is billed to. One scope per call, never a default.""" + + organization_id: str + project_id: str | None = None + employee_id: str | None = None + user_id: str | None = None + run_id: str | None = None + + +async def call_model( + *, + provider: ModelProviderPort, + credentials: ModelCredentialRepository, + rate_cards: RateCardRepository, + ledger: UsageLedgerRepository, + model: AiModel, + prompt, + scope: BillingScope, +) -> ModelResponse: + """Call the model and bill it. **The** single model entry point. + + Every model call in the platform goes through here — a task turn, a + triage turn, a final verdict and the manager's chat alike. Billing reads + the internal ledger, never the provider's invoice, so a call that skips + this function is a call the platform never charges for (agybot's manager + chat bug). + """ + pool = await credentials.list_active_for_provider(model.provider_id) + credential = pool[0] if pool else None + response = await provider.complete( + model=model, + credential=credential, + credentials=pool or None, + messages=prompt.messages, + system=prompt.system, + ) + for unit, tokens in ( + (RateUnit.MODEL_INPUT_TOKENS, float(response.input_tokens)), + (RateUnit.MODEL_OUTPUT_TOKENS, float(response.output_tokens)), + ): + await record_model_usage( + ledger=ledger, + rate_cards=rate_cards, + organization_id=scope.organization_id, + unit=unit, + quantity=tokens, + project_id=scope.project_id, + user_id=scope.user_id, + employee_id=scope.employee_id, + run_id=scope.run_id, + ) + return response + + class TurnRunner: """Runs one turn at a time. One instance per process is enough.""" @@ -224,6 +280,7 @@ class TurnRunner: history: list[TurnMessage] = list(request.history) tool_calls_made = 0 model_calls = 0 + protocol_errors = 0 reply_text = "" tags: tuple[Tag, ...] = () turn_row = await self._open_turn_row(request) @@ -272,6 +329,17 @@ class TurnRunner: parsed = parse_reply(response.content, role=request.role, mode=request.mode) if parsed.errors: + protocol_errors += 1 + if protocol_errors > settings.engine_max_protocol_retries: + return await self._fail( + request, + turn_row, + ErrorKind.PROTOCOL.value, + "پاسخ مدل مکرراً ناقض پروتکل بود.", + tool_calls_made, + model_calls, + incomplete=True, + ) # The model learns the protocol from what it just got wrong. history.append( TurnMessage(kind="result", text=protocol_error_feedback(parsed.errors)) @@ -358,32 +426,22 @@ class TurnRunner: return model async def _complete(self, request: TurnRequest, model: AiModel, prompt): - """One model call, billed exactly like every other model call.""" - pool = await self.credentials.list_active_for_provider(model.provider_id) - credential = pool[0] if pool else None - response = await self.provider.complete( + """One model call, billed through the one shared model entry point.""" + return await call_model( + provider=self.provider, + credentials=self.credentials, + rate_cards=self.rate_cards, + ledger=self.ledger, model=model, - credential=credential, - credentials=pool or None, - messages=prompt.messages, - system=prompt.system, - ) - for unit, tokens in ( - (RateUnit.MODEL_INPUT_TOKENS, float(response.input_tokens)), - (RateUnit.MODEL_OUTPUT_TOKENS, float(response.output_tokens)), - ): - await record_model_usage( - ledger=self.ledger, - rate_cards=self.rate_cards, + prompt=prompt, + scope=BillingScope( organization_id=request.organization_id, - unit=unit, - quantity=tokens, project_id=request.task.project_id, - user_id=request.user_id, employee_id=request.employee.id if request.employee else None, + user_id=request.user_id, run_id=request.run_id, - ) - return response + ), + ) # -- tools ------------------------------------------------------------- # @@ -579,4 +637,12 @@ class _ModelUnavailable(Exception): self.kind = kind -__all__ = ["TERMINAL_TAGS", "TurnRequest", "TurnResult", "TurnRunner", "classify_error"] +__all__ = [ + "TERMINAL_TAGS", + "BillingScope", + "TurnRequest", + "TurnResult", + "TurnRunner", + "call_model", + "classify_error", +] diff --git a/apps/api/app/application/services/workforce_service.py b/apps/api/app/application/services/workforce_service.py index a23e270..df99f4e 100644 --- a/apps/api/app/application/services/workforce_service.py +++ b/apps/api/app/application/services/workforce_service.py @@ -27,8 +27,6 @@ from app.application.ports import ( from app.application.ports.realtime import ( emit_approval_pending, emit_artifact_created, - emit_chat_pending, - emit_message_created, emit_run_status, emit_run_step, emit_task_created, @@ -51,7 +49,6 @@ from app.domain.workforce.models import ( EmployeeStatus, EmployeeToolPermission, Message, - MessageRole, PromptVersion, RoleKey, Run, @@ -364,67 +361,48 @@ class WorkforceService: user_id: str, employee_id: str | None = None, ) -> tuple[Message, Message]: - """Append the user's message, then produce the assistant reply. + """Append the user's message, then let the manager answer it. - Returns ``(user_message, assistant_message)`` so the caller can render - both without a second round-trip. + The manager's turn is the engine's job (``engine/manager.py``): it + replies, and it may create tasks, change the team or pause work in the + same breath. This method stays the single entry point the router calls; + it builds the engine's manager from the same session, so the message, + the task and the usage events commit together. """ - thread = await self.get_thread_or_404( - organization_id=organization_id, thread_id=thread_id - ) - if not content.strip(): - raise ValidationError("متن پیام نمی‌تواند خالی باشد.") - - user_message = await self.messages.create( - Message(thread_id=thread.id, role=MessageRole.USER, content=content) - ) - - employee: Employee | None = None - if employee_id: - employee = await self.get_employee_or_404( - organization_id=organization_id, employee_id=employee_id - ) - else: - employee = await self._default_manager(organization_id) - - # The manager is about to call a model. Tell the thread so the UI can - # show a thinking indicator immediately. - await emit_chat_pending(thread_id=thread.id, organization_id=organization_id) - - history = await self.messages.list_for_thread(thread.id) - system_prompt = await self._system_prompt_for(employee) - reply = await self._complete( + manager = self._manager_turns() + return await manager.chat( organization_id=organization_id, - employee=employee, - project_id=thread.project_id, + thread_id=thread_id, + content=content, user_id=user_id, - system=system_prompt, - messages=[ - ChatMessage(role=m.role.value, content=m.content) - for m in history - if m.role in (MessageRole.USER, MessageRole.ASSISTANT) - ], + employee_id=employee_id, ) - assistant_message = await self.messages.create( - Message( - thread_id=thread.id, - role=MessageRole.ASSISTANT, - content=reply, - employee_id=employee.id if employee else None, - ) + def _manager_turns(self): + """The engine's manager service, wired on this service's session. + + Ports that take a session (model grants, memory) are built here rather + than injected: ``container.py`` is a frozen composition root, and the + engine must not widen a constructor it does not own. Sharing the + session is the point — a reply, a task and its usage events belong to + one transaction. + """ + from app.application.container_org import build_memory, build_model_grants + from app.application.engine.manager import ManagerTurns + + return ManagerTurns( + provider=self.provider, + employees=self.employees, + messages=self.messages, + tasks=self.tasks, + threads=self.threads, + models=self.models, + credentials=self.credentials, + rate_cards=self.rate_cards, + ledger=self.ledger, + grants=build_model_grants(self.tasks.session), + memory=build_memory(self.tasks.session), ) - # Live push: any browser on this thread sees the reply arrive, and the - # pending indicator clears, without waiting for a reload. - await emit_message_created( - thread_id=thread.id, - organization_id=organization_id, - message_id=assistant_message.id, - role=MessageRole.ASSISTANT.value, - content=reply, - employee_id=employee.id if employee else None, - ) - return user_message, assistant_message # ------------------------------------------------------------------ # # Tasks # diff --git a/apps/api/app/core/config.py b/apps/api/app/core/config.py index 63c0674..87d9021 100644 --- a/apps/api/app/core/config.py +++ b/apps/api/app/core/config.py @@ -144,6 +144,10 @@ class Settings(BaseSettings): # >>> agent-2 (AI workforce engine): add settings ONLY between these markers engine_max_tool_calls_per_turn: int = 40 + #: Consecutive malformed replies tolerated before a turn gives up. Without + #: it a model that repeats the same protocol mistake would be called until + #: the turn timeout, hundreds of paid calls later. + engine_max_protocol_retries: int = 3 engine_turn_timeout_seconds: int = 900 engine_max_cycles: int = 6 engine_max_employees: int = 20 diff --git a/apps/api/app/infrastructure/repositories/workforce.py b/apps/api/app/infrastructure/repositories/workforce.py index 12523c5..fcb4cd5 100644 --- a/apps/api/app/infrastructure/repositories/workforce.py +++ b/apps/api/app/infrastructure/repositories/workforce.py @@ -482,6 +482,15 @@ class TaskRepository: await self.session.flush() return _task(row) + async def set_assignee(self, task_id: str, employee_id: str | None) -> Task | None: + """Point a task at an employee. The last step of the manager's triage.""" + row = await self.session.get(TaskRow, task_id) + if row is None: + return None + row.assigned_employee_id = employee_id + await self.session.flush() + return _task(row) + async def set_status(self, task_id: str, status: TaskStatus) -> Task | None: row = await self.session.get(TaskRow, task_id) if row is None: diff --git a/apps/api/tests/engine_fakes.py b/apps/api/tests/engine_fakes.py index 18780fa..4d1976c 100644 --- a/apps/api/tests/engine_fakes.py +++ b/apps/api/tests/engine_fakes.py @@ -182,24 +182,41 @@ class FakeSandbox: class FakeGrants: - """A ``ModelGrantPort`` with an explicit allow-list.""" + """A ``ModelGrantPort`` with an explicit allow-list. - def __init__(self, granted: list[str] | None = None) -> None: - self._granted = list(granted or ["model-strong", "model-fast"]) + ``granted`` is either a list of model ids (all ``balanced``) or a mapping + of id → neutral tier, so a test can have a "strong" and a "fast" model and + assert the engine picked the right one. + """ + + def __init__(self, granted: list[str] | dict[str, str] | None = None) -> None: + if granted is None: + self._tiers: dict[str, str] = { + "model-strong": "strong", + "model-fast": "fast", + } + elif isinstance(granted, dict): + self._tiers = dict(granted) + else: + self._tiers = dict.fromkeys(granted, "balanced") self.checks: list[str] = [] + @property + def granted(self) -> list[str]: + return list(self._tiers) + async def granted_models(self, *, organization_id: str) -> list[GrantedModel]: return [ - GrantedModel(ai_model_id=model_id, tier="balanced") - for model_id in self._granted + GrantedModel(ai_model_id=model_id, tier=tier) + for model_id, tier in self._tiers.items() ] async def is_granted(self, *, organization_id: str, ai_model_id: str) -> bool: self.checks.append(ai_model_id) - return ai_model_id in self._granted + return ai_model_id in self._tiers - def grant(self, model_id: str) -> None: - self._granted.append(model_id) + def grant(self, model_id: str, tier: str = "balanced") -> None: + self._tiers[model_id] = tier class RecordingMemory: diff --git a/apps/api/tests/test_e2e_smoke.py b/apps/api/tests/test_e2e_smoke.py index 5cac516..892f91f 100644 --- a/apps/api/tests/test_e2e_smoke.py +++ b/apps/api/tests/test_e2e_smoke.py @@ -162,6 +162,20 @@ async def test_end_to_end_smoke_path(client: AsyncClient): assert "model" not in str(employee.json()).lower() # -- 8. send a manager message ----------------------------------------- # + # PLAN-v2 D3: the engine may only run on models the owner granted to this + # organization, so the manager's chat needs a grant before it can answer. + # Before the engine existed any active model would do; a chat without a + # grant now correctly answers nothing but the "waiting for a model" note. + models = await client.get("/api/admin/models", headers=headers) + assert models.status_code == 200, models.text + ai_model_id = models.json()[0]["id"] + grant = await client.put( + f"/api/admin/model-grants/{org_id}/models/{ai_model_id}", + json={"daily_credit_cap": None}, + headers=headers, + ) + assert grant.status_code in (200, 201), grant.text + thread = await client.post( f"/api/organizations/{org_id}/threads", json={"title": "گفت‌وگو با مدیر", "project_id": project_id}, diff --git a/apps/api/tests/test_engine_manager.py b/apps/api/tests/test_engine_manager.py new file mode 100644 index 0000000..20d948c --- /dev/null +++ b/apps/api/tests/test_engine_manager.py @@ -0,0 +1,828 @@ +"""The manager AI: chat → tasks → team, triage, final verdict, bootstrap. + +What these tests defend: + +* an order becomes **at most three** tasks, each with its own pipeline state at + ``pending_manager_review`` and a link back to the thread +* HIRE respects the employee cap and picks a **granted** model for the tier; + the last manager cannot be fired +* ``ASK_USER`` parks the task with a human and asks in the thread +* triage assigns to the least busy employee, and a name is matched **inside + the org only** — the manager of org A can never assign an employee of org B, + even by exact name +* ``COMPLETE`` merges, ``REVISE`` goes back to the developer with cycle + 1 +* ``KNOWLEDGE`` tags become org memory +* bootstrap is idempotent, and with no grants it creates an inactive team plus + the "waiting for the owner" message +""" + +from __future__ import annotations + +import pytest +import pytest_asyncio +from sqlalchemy import select + +from app.application.container_org import build_memory +from app.application.engine.manager import ( + NO_MODEL_GRANTED_MESSAGE, + ManagerTurns, +) +from app.core.config import settings +from app.core.errors import ConflictError, NotFoundError +from app.domain.base import new_id +from app.domain.workforce.engine import EmployeeProfile +from app.domain.workforce.models import Employee, RoleKey +from app.domain.workforce.pipeline import PipelineKey, Stage, TaskPipelineState +from app.infrastructure.db.models import ( + AiModelRow, + AiProviderRow, + EmployeeRow, + OrganizationRow, + RunRow, + TaskRow, + ThreadRow, + UserRow, +) +from app.infrastructure.db.models_org import OrgMemoryRow +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, + RunRepository, + RunStepRepository, + TaskRepository, + ThreadRepository, + ToolCallRepository, +) +from tests.engine_fakes import FakeGrants, ScriptedModel, ScriptedReply + + +@pytest_asyncio.fixture +async def world(session_factory, seeded): + """One organization with a manager, a developer, a granted model.""" + async with session_factory() as session: + user = UserRow(email=f"mgr-{new_id()[:8]}@example.com", password_hash="x") + session.add(user) + await session.flush() + org = OrganizationRow(name="سازمان مدیر", slug=f"mgr-{user.id[:6]}", owner_user_id=user.id) + session.add(org) + await session.flush() + provider_id = (await session.execute(select(AiProviderRow.id).limit(1))).scalar_one() + 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, + ) + ) + fast = await AiModelRepository(session).create( + AiModelRow( + provider_id=provider_id, + internal_name=f"fast-{new_id()[:4]}", + display_name_for_admin="مدل سریع", + capabilities=["text", "code"], + input_price_per_1m=0.1, + output_price_per_1m=0.5, + is_active=True, + ) + ) + await session.commit() + yield _World( + session_factory=session_factory, + organization_id=org.id, + user_id=user.id, + strong_id=strong.id, + fast_id=fast.id, + ) + + +class _World: + def __init__(self, **kwargs) -> None: + self.__dict__.update(kwargs) + + async def employee(self, session, name: str, role: RoleKey = RoleKey.DEVELOPER): + return await EmployeeRepository(session).create( + Employee(organization_id=self.organization_id, name=name, role_key=role) + ) + + +async def _manager(world, session, *, scripted, grants, employee_ids=None) -> ManagerTurns: + """A manager service with real repositories and scripted everything else.""" + return 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=build_memory(session), + profiles=EmployeeProfileRepository(session), + states=PipelineStateRepository(session), + runner=None, + ) + + +async def _with_model_profile(world, session, employee_id: str, model_id: str) -> None: + await EmployeeProfileRepository(session).upsert( + EmployeeProfile(employee_id=employee_id, ai_model_id=model_id, roles=["manager"]) + ) + + +# --------------------------------------------------------------------------- +# 3.1 chat: an order becomes tasks +# --------------------------------------------------------------------------- + + +async def test_an_order_becomes_tasks_with_pipeline_state(world, session_factory): + scripted = ScriptedModel( + [ + ScriptedReply( + content=( + "باشه، دو کار می‌سازم.\n" + '[[CREATE_TASK title="فرم ورود" kind="feature"]]' + "ساخت فرم ورود با اعتبارسنجی[[/CREATE_TASK]]" + '[[CREATE_TASK title="تست فرم" kind="feature" depends_on=""]]' + "تست فرم ورود[[/CREATE_TASK]]" + ) + ) + ] + ) + async with session_factory() as session: + manager_id = (await world.employee(session, "مدیر یک", RoleKey.MANAGER)).id + await world.employee(session, "سارا", RoleKey.DEVELOPER) + await _with_model_profile(world, session, manager_id, world.strong_id) + thread = await ThreadRepository(session).create( + ThreadRow(organization_id=world.organization_id, title="گفتگو") + ) + await session.commit() + manager = await _manager( + world, session, scripted=scripted, grants=FakeGrants({world.strong_id: "strong"}) + ) + user_message, assistant = await manager.chat( + organization_id=world.organization_id, + thread_id=thread.id, + content="یک صفحهٔ ورود با تست بساز.", + user_id=world.user_id, + ) + await session.commit() + tasks = await TaskRepository(session).list_for_org(world.organization_id) + states = PipelineStateRepository(session) + + assert "باشه" in assistant.content + assert "کار ساخته شد: فرم ورود" in assistant.content + assert len(tasks) == 2 + assert tasks[0].thread_id == thread.id + for task in tasks: + state = await states.get(task.id) + assert state.stage is Stage.PENDING_MANAGER_REVIEW + assert state.pipeline is PipelineKey.PLAN_CODE_TEST + assert state.organization_id == world.organization_id + # The task-created event went out, so the manager chat shows the cards live. + assert scripted.calls == 1 + + +async def test_at_most_three_tasks_per_order(world, session_factory): + scripted = ScriptedModel( + [ + ScriptedReply( + content="".join( + f'[[CREATE_TASK title="کار {i}"]]توضیح[[/CREATE_TASK]]' for i in range(5) + ) + ) + ] + ) + async with session_factory() as session: + manager_id = (await world.employee(session, "مدیر یک", RoleKey.MANAGER)).id + await _with_model_profile(world, session, manager_id, world.strong_id) + thread = await ThreadRepository(session).create( + ThreadRow(organization_id=world.organization_id, title="گفتگو") + ) + await session.commit() + manager = await _manager( + world, session, scripted=scripted, grants=FakeGrants({world.strong_id: "strong"}) + ) + _user, assistant = await manager.chat( + organization_id=world.organization_id, + thread_id=thread.id, + content="پنج کار بساز.", + user_id=world.user_id, + ) + await session.commit() + tasks = await TaskRepository(session).list_for_org(world.organization_id) + + # Three created, two refused by the protocol limit, and the user is told. + assert len(tasks) == 3 + assert assistant.content.count("کار ساخته شد") == 3 + + +async def test_the_chat_prompt_asks_for_atomic_tasks(world, session_factory): + scripted = ScriptedModel([ScriptedReply(content="سلام")]) + async with session_factory() as session: + manager_id = (await world.employee(session, "مدیر یک", RoleKey.MANAGER)).id + await _with_model_profile(world, session, manager_id, world.strong_id) + thread = await ThreadRepository(session).create( + ThreadRow(organization_id=world.organization_id, title="گفتگو") + ) + await session.commit() + manager = await _manager( + world, session, scripted=scripted, grants=FakeGrants({world.strong_id: "strong"}) + ) + await manager.chat( + organization_id=world.organization_id, + thread_id=thread.id, + content="یک کار بساز.", + user_id=world.user_id, + ) + await session.commit() + + prompt = scripted.all_text() + assert "اتمیک" in prompt # atomic tasks that do not touch the same files + assert "هم‌زمان" in prompt + assert "مدیر یک" in prompt # the roster is in the context + + +async def test_a_chat_model_call_is_billed(world, session_factory): + scripted = ScriptedModel([ScriptedReply(content="سلام", input_tokens=9, output_tokens=4)]) + from app.infrastructure.db.models import UsageEventRow + + async with session_factory() as session: + manager_id = (await world.employee(session, "مدیر یک", RoleKey.MANAGER)).id + await _with_model_profile(world, session, manager_id, world.strong_id) + thread = await ThreadRepository(session).create( + ThreadRow(organization_id=world.organization_id, title="گفتگو") + ) + await session.commit() + manager = await _manager( + world, session, scripted=scripted, grants=FakeGrants({world.strong_id: "strong"}) + ) + await manager.chat( + organization_id=world.organization_id, + thread_id=thread.id, + content="سلام", + user_id=world.user_id, + ) + await session.commit() + rows = ( + ( + await session.execute( + select(UsageEventRow).where( + UsageEventRow.organization_id == world.organization_id + ) + ) + ) + .scalars() + .all() + ) + + assert len(rows) == 2 + assert sum(r.quantity for r in rows if r.unit == "model_input_tokens") == 9 + assert sum(r.quantity for r in rows if r.unit == "model_output_tokens") == 4 + + +async def test_a_chat_without_grants_explains_itself(world, session_factory): + """No grant, no answer — and the user is told why, in Persian.""" + scripted = ScriptedModel([ScriptedReply(content="سلام")]) + async with session_factory() as session: + manager_id = (await world.employee(session, "مدیر یک", RoleKey.MANAGER)).id + await _with_model_profile(world, session, manager_id, world.strong_id) + thread = await ThreadRepository(session).create( + ThreadRow(organization_id=world.organization_id, title="گفتگو") + ) + await session.commit() + manager = await _manager(world, session, scripted=scripted, grants=FakeGrants({})) + _user, assistant = await manager.chat( + organization_id=world.organization_id, + thread_id=thread.id, + content="سلام", + user_id=world.user_id, + ) + await session.commit() + messages = await MessageRepository(session).list_for_thread(thread.id) + + assert scripted.calls == 0 + assert "دستور دریافت شد" in assistant.content + assert any("مدیری با مدل" in m.content for m in messages) + + +# --------------------------------------------------------------------------- +# 3.2 team +# --------------------------------------------------------------------------- + + +async def test_hire_picks_a_granted_model_of_the_tier(world, session_factory): + async with session_factory() as session: + grants = FakeGrants({world.strong_id: "strong", world.fast_id: "fast"}) + manager = await _manager(world, session, scripted=ScriptedModel(), grants=grants) + employee = await manager.hire( + organization_id=world.organization_id, role=RoleKey.TESTER, tier="fast" + ) + await session.commit() + profile = await EmployeeProfileRepository(session).get(employee.id) + + assert profile is not None + assert profile.ai_model_id == world.fast_id + assert profile.role_list() == ["tester"] + assert profile.hired_by == "manager" + + +async def test_hire_without_a_granted_model_is_refused(world, session_factory): + async with session_factory() as session: + manager = await _manager( + world, session, scripted=ScriptedModel(), grants=FakeGrants({world.fast_id: "fast"}) + ) + with pytest.raises(ConflictError): + await manager.hire( + organization_id=world.organization_id, role=RoleKey.REVIEWER, tier="strong" + ) + + +async def test_the_employee_cap_is_enforced(world, session_factory, monkeypatch): + monkeypatch.setattr(settings, "engine_max_employees", 2) + async with session_factory() as session: + await world.employee(session, "یک") + await world.employee(session, "دو") + await session.commit() + manager = await _manager( + world, session, scripted=ScriptedModel(), grants=FakeGrants({world.fast_id: "fast"}) + ) + with pytest.raises(ConflictError): + await manager.hire( + organization_id=world.organization_id, + role=RoleKey.TESTER, + tier="fast", + ) + + +async def test_the_last_manager_cannot_be_fired(world, session_factory): + async with session_factory() as session: + manager_row = await world.employee(session, "مدیر یک", RoleKey.MANAGER) + await session.commit() + manager = await _manager(world, session, scripted=ScriptedModel(), grants=FakeGrants({})) + with pytest.raises(ConflictError): + await manager.fire(organization_id=world.organization_id, employee_id=manager_row.id) + # A second manager can be fired. + second = await world.employee(session, "مدیر دو", RoleKey.MANAGER) + await session.commit() + fired = await manager.fire(organization_id=world.organization_id, employee_id=second.id) + await session.commit() + + assert fired.status.value == "archived" + + +async def test_firing_a_foreign_employee_is_a_404(world, session_factory): + """Another tenant's employee id is indistinguishable from a missing one.""" + async with session_factory() as session: + other_user = UserRow(email=f"other-{new_id()[:6]}@example.com", password_hash="x") + session.add(other_user) + await session.flush() + other_org = OrganizationRow( + name="سازمان دیگر", slug=f"oth-{other_user.id[:6]}", owner_user_id=other_user.id + ) + session.add(other_org) + await session.flush() + stranger = await session.execute(select(EmployeeRow).limit(1)) + foreign = None + if stranger.scalars().first() is not None: + foreign = stranger.scalars().first().id + await session.commit() + + if foreign is None: + async with session_factory() as session: + foreign = (await world.employee(session, "بومی")).id + await session.commit() + + async with session_factory() as session: + # The employee belongs to *this* org; the *other* org must not see it. + manager = await _manager(world, session, scripted=ScriptedModel(), grants=FakeGrants({})) + with pytest.raises(NotFoundError): + await manager.fire(organization_id="org-does-not-exist", employee_id=foreign) + + +# --------------------------------------------------------------------------- +# 3.3 triage +# --------------------------------------------------------------------------- + + +async def _task_with_state( + world, session, title: str = "کار", stage: Stage = Stage.PENDING_MANAGER_REVIEW +) -> tuple[str, str]: + task = TaskRow(organization_id=world.organization_id, title=title) + session.add(task) + await session.flush() + run = RunRow(task_id=task.id) + session.add(run) + await session.flush() + await PipelineStateRepository(session).create( + TaskPipelineState(task_id=task.id, organization_id=world.organization_id, stage=stage) + ) + await session.flush() + return task.id, run.id + + +async def test_triage_assigns_to_the_least_busy_employee(world, session_factory, monkeypatch): + monkeypatch.setattr(settings, "engine_max_tool_calls_per_turn", 40) + scripted = ScriptedModel( + [ScriptedReply(content='[[ASSIGN employee="" pipeline="plan_code_test"/]]')] + ) + async with session_factory() as session: + manager_row = await world.employee(session, "مدیر یک", RoleKey.MANAGER) + await _with_model_profile(world, session, manager_row.id, world.strong_id) + busy = await world.employee(session, "مشغول", RoleKey.DEVELOPER) + free = await world.employee(session, "آzاد", RoleKey.DEVELOPER) + # The busy one already has two open tasks. + for _ in range(2): + session.add( + TaskRow( + organization_id=world.organization_id, + title="کار دیگر", + assigned_employee_id=busy.id, + ) + ) + await session.flush() + task_id, run_id = await _task_with_state(world, session) + await session.commit() + + from app.application.engine.runner import TurnRunner + + runner = TurnRunner( + provider=scripted, + models=AiModelRepository(session), + credentials=ModelCredentialRepository(session), + rate_cards=RateCardRepository(session), + ledger=UsageLedgerRepository(session), + grants=FakeGrants({world.strong_id: "strong"}), + memory=build_memory(session), + turns=None, + run_steps=RunStepRepository(session), + tool_calls=ToolCallRepository(session), + runs=RunRepository(session), + states=PipelineStateRepository(session), + ) + manager = await _manager( + world, session, scripted=scripted, grants=FakeGrants({world.strong_id: "strong"}) + ) + manager.runner = runner + task = await TaskRepository(session).get(task_id) + outcome = await manager.triage(task=task, run_id=run_id, employee=manager_row) + await session.commit() + state = await PipelineStateRepository(session).get(task_id) + refreshed = await TaskRepository(session).get(task_id) + + assert outcome.acted is True + assert outcome.decision == "assigned" + assert outcome.employee_id == free.id + assert state.stage is Stage.PENDING_ANALYSIS + assert refreshed.assigned_employee_id == free.id + + +async def test_a_named_employee_is_matched_inside_the_org(world, session_factory): + scripted = ScriptedModel([ScriptedReply(content='[[ASSIGN employee="سارا"/]]')]) + async with session_factory() as session: + manager_row = await world.employee(session, "مدیر یک", RoleKey.MANAGER) + await _with_model_profile(world, session, manager_row.id, world.strong_id) + sara = await world.employee(session, "سارا", RoleKey.DEVELOPER) + await session.flush() + task_id, run_id = await _task_with_state(world, session) + await session.commit() + from app.application.engine.runner import TurnRunner + + runner = TurnRunner( + provider=scripted, + models=AiModelRepository(session), + credentials=ModelCredentialRepository(session), + rate_cards=RateCardRepository(session), + ledger=UsageLedgerRepository(session), + grants=FakeGrants({world.strong_id: "strong"}), + memory=build_memory(session), + states=PipelineStateRepository(session), + ) + manager = await _manager( + world, session, scripted=scripted, grants=FakeGrants({world.strong_id: "strong"}) + ) + manager.runner = runner + task = await TaskRepository(session).get(task_id) + outcome = await manager.triage(task=task, run_id=run_id, employee=manager_row) + await session.commit() + + assert outcome.employee_id == sara.id + + +async def test_the_manager_of_org_a_cannot_assign_an_employee_of_org_b(world, session_factory): + """Cross-tenant assignment by exact name must resolve to nothing.""" + scripted = ScriptedModel([ScriptedReply(content='[[ASSIGN employee="دزد"/]]')]) + async with session_factory() as session: + manager_row = await world.employee(session, "مدیر یک", RoleKey.MANAGER) + await _with_model_profile(world, session, manager_row.id, world.strong_id) + other_user = UserRow(email=f"b-{new_id()[:6]}@example.com", password_hash="x") + session.add(other_user) + await session.flush() + other_org = OrganizationRow( + name="سازمان ب", slug=f"b-{other_user.id[:6]}", owner_user_id=other_user.id + ) + session.add(other_org) + await session.flush() + # The other org's employee is named exactly like the assignment asks. + session.add(EmployeeRow(organization_id=other_org.id, name="دزد", role_key="developer")) + await session.flush() + task_id, run_id = await _task_with_state(world, session) + await session.commit() + from app.application.engine.runner import TurnRunner + + runner = TurnRunner( + provider=scripted, + models=AiModelRepository(session), + credentials=ModelCredentialRepository(session), + rate_cards=RateCardRepository(session), + ledger=UsageLedgerRepository(session), + grants=FakeGrants({world.strong_id: "strong"}), + memory=build_memory(session), + states=PipelineStateRepository(session), + ) + manager = await _manager( + world, session, scripted=scripted, grants=FakeGrants({world.strong_id: "strong"}) + ) + manager.runner = runner + task = await TaskRepository(session).get(task_id) + outcome = await manager.triage(task=task, run_id=run_id, employee=manager_row) + await session.commit() + state = await PipelineStateRepository(session).get(task_id) + + assert outcome.acted is False + assert outcome.employee_id is None + assert state.stage is Stage.PENDING_MANAGER_REVIEW + + +async def test_ask_user_parks_the_task_and_asks_in_the_thread(world, session_factory): + scripted = ScriptedModel([ScriptedReply(content="[[ASK_USER]]کدام درگاه پرداخت؟[[/ASK_USER]]")]) + async with session_factory() as session: + manager_row = await world.employee(session, "مدیر یک", RoleKey.MANAGER) + await _with_model_profile(world, session, manager_row.id, world.strong_id) + thread = await ThreadRepository(session).create( + ThreadRow(organization_id=world.organization_id, title="گفتگو") + ) + await session.flush() + task_id, run_id = await _task_with_state(world, session) + task = await TaskRepository(session).get(task_id) + task.thread_id = thread.id + await session.commit() + from app.application.engine.runner import TurnRunner + + runner = TurnRunner( + provider=scripted, + models=AiModelRepository(session), + credentials=ModelCredentialRepository(session), + rate_cards=RateCardRepository(session), + ledger=UsageLedgerRepository(session), + grants=FakeGrants({world.strong_id: "strong"}), + memory=build_memory(session), + states=PipelineStateRepository(session), + ) + manager = await _manager( + world, session, scripted=scripted, grants=FakeGrants({world.strong_id: "strong"}) + ) + manager.runner = runner + outcome = await manager.triage(task=task, run_id=run_id, employee=manager_row) + await session.commit() + state = await PipelineStateRepository(session).get(task_id) + messages = await MessageRepository(session).list_for_thread(thread.id) + + assert outcome.decision == "asked_user" + assert state.stage is Stage.PENDING_USER_DECISION + assert any(m.content == "کدام درگاه پرداخت؟" for m in messages) + + +# --------------------------------------------------------------------------- +# 3.4 final verdict +# --------------------------------------------------------------------------- + + +async def test_complete_moves_the_task_to_merge_request(world, session_factory): + scripted = ScriptedModel([ScriptedReply(content="[[COMPLETE/]]")]) + async with session_factory() as session: + manager_row = await world.employee(session, "مدیر یک", RoleKey.MANAGER) + await _with_model_profile(world, session, manager_row.id, world.strong_id) + task_id, run_id = await _task_with_state(world, session, stage=Stage.PENDING_FINAL_REVIEW) + await session.commit() + from app.application.engine.runner import TurnRunner + + runner = TurnRunner( + provider=scripted, + models=AiModelRepository(session), + credentials=ModelCredentialRepository(session), + rate_cards=RateCardRepository(session), + ledger=UsageLedgerRepository(session), + grants=FakeGrants({world.strong_id: "strong"}), + memory=build_memory(session), + states=PipelineStateRepository(session), + ) + manager = await _manager( + world, session, scripted=scripted, grants=FakeGrants({world.strong_id: "strong"}) + ) + manager.runner = runner + task = await TaskRepository(session).get(task_id) + outcome = await manager.final_review( + task=task, + run_id=run_id, + employee=manager_row, + spec="برنامه", + test_output="3 passed", + ) + await session.commit() + state = await PipelineStateRepository(session).get(task_id) + + assert outcome.decision == "completed" + assert state.stage is Stage.MERGE_REQUEST + + +async def test_revise_sends_the_task_back_and_counts_the_cycle(world, session_factory): + scripted = ScriptedModel([ScriptedReply(content="[[REVISE]]تست‌ها قرمز است[[/REVISE]]")]) + async with session_factory() as session: + manager_row = await world.employee(session, "مدیر یک", RoleKey.MANAGER) + await _with_model_profile(world, session, manager_row.id, world.strong_id) + task_id, run_id = await _task_with_state(world, session, stage=Stage.PENDING_FINAL_REVIEW) + await session.commit() + from app.application.engine.runner import TurnRunner + + runner = TurnRunner( + provider=scripted, + models=AiModelRepository(session), + credentials=ModelCredentialRepository(session), + rate_cards=RateCardRepository(session), + ledger=UsageLedgerRepository(session), + grants=FakeGrants({world.strong_id: "strong"}), + memory=build_memory(session), + states=PipelineStateRepository(session), + ) + manager = await _manager( + world, session, scripted=scripted, grants=FakeGrants({world.strong_id: "strong"}) + ) + manager.runner = runner + task = await TaskRepository(session).get(task_id) + outcome = await manager.final_review( + task=task, run_id=run_id, employee=manager_row, test_output="1 failed" + ) + await session.commit() + state = await PipelineStateRepository(session).get(task_id) + + assert outcome.decision == "revised" + assert state.stage is Stage.PENDING_IMPLEMENTATION + assert state.cycle == 1 + + +async def test_resolve_error_retries_with_a_hint(world, session_factory): + scripted = ScriptedModel( + [ScriptedReply(content='[[RESOLVE_ERROR hint="ابتدا مهاجرت را بنویس"/]]')] + ) + async with session_factory() as session: + manager_row = await world.employee(session, "مدیر یک", RoleKey.MANAGER) + await _with_model_profile(world, session, manager_row.id, world.strong_id) + task_id, run_id = await _task_with_state(world, session) + await session.commit() + from app.application.engine.runner import TurnRunner + + runner = TurnRunner( + provider=scripted, + models=AiModelRepository(session), + credentials=ModelCredentialRepository(session), + rate_cards=RateCardRepository(session), + ledger=UsageLedgerRepository(session), + grants=FakeGrants({world.strong_id: "strong"}), + memory=build_memory(session), + states=PipelineStateRepository(session), + ) + manager = await _manager( + world, session, scripted=scripted, grants=FakeGrants({world.strong_id: "strong"}) + ) + manager.runner = runner + task = await TaskRepository(session).get(task_id) + await PipelineStateRepository(session).advance( + task_id=task_id, expected_from=Stage.PENDING_MANAGER_REVIEW, to=Stage.ERROR_REVIEW + ) + await session.commit() + outcome = await manager.error_review( + task=task, run_id=run_id, employee=manager_row, error_summary="تست قرمز" + ) + await session.commit() + state = await PipelineStateRepository(session).get(task_id) + + assert outcome.decision == "resolved" + assert state.stage is Stage.PENDING_IMPLEMENTATION + assert "مهاجرت" in outcome.message + + +# --------------------------------------------------------------------------- +# 3.5 knowledge and 3.6 bootstrap +# --------------------------------------------------------------------------- + + +async def test_knowledge_tags_become_org_memory(world, session_factory): + scripted = ScriptedModel( + [ + ScriptedReply( + content=('[[KNOWLEDGE kind="rule" key="deploy"]]قبل از پوش تست بگیر.[[/KNOWLEDGE]]') + ) + ] + ) + async with session_factory() as session: + manager_row = await world.employee(session, "مدیر یک", RoleKey.MANAGER) + await _with_model_profile(world, session, manager_row.id, world.strong_id) + thread = await ThreadRepository(session).create( + ThreadRow(organization_id=world.organization_id, title="گفتگو") + ) + await session.commit() + manager = await _manager( + world, session, scripted=scripted, grants=FakeGrants({world.strong_id: "strong"}) + ) + await manager.chat( + organization_id=world.organization_id, + thread_id=thread.id, + content="یادت باشد.", + user_id=world.user_id, + ) + await session.commit() + rows = ( + ( + await session.execute( + select(OrgMemoryRow).where( + OrgMemoryRow.organization_id == world.organization_id + ) + ) + ) + .scalars() + .all() + ) + + assert len(rows) == 1 + assert rows[0].kind == "rule" + assert rows[0].source == "manager" + assert rows[0].content == "قبل از پوش تست بگیر." + + +async def test_bootstrap_creates_the_team_and_is_idempotent(world, session_factory): + async with session_factory() as session: + manager = await _manager( + world, + session, + scripted=ScriptedModel(), + grants=FakeGrants({world.strong_id: "strong", world.fast_id: "fast"}), + ) + await manager.bootstrap_team(organization_id=world.organization_id, project_id=None) + await session.commit() + employees = await EmployeeRepository(session).list_for_org(world.organization_id) + profiles = {} + for employee in employees: + profile = await EmployeeProfileRepository(session).get(employee.id) + profiles[employee.role_key] = profile.ai_model_id if profile else None + + # A second call changes nothing. + await manager.bootstrap_team(organization_id=world.organization_id, project_id=None) + await session.commit() + again = await EmployeeRepository(session).list_for_org(world.organization_id) + + assert len(employees) == 4 + assert len(again) == 4 + assert {e.role_key for e in employees} == { + RoleKey.MANAGER, + RoleKey.PLANNER, + RoleKey.DEVELOPER, + RoleKey.TESTER, + } + assert all(e.status.value == "active" for e in employees) + # The manager runs on the strongest granted tier, the tester on the fast one. + assert profiles[RoleKey.MANAGER] == world.strong_id + assert profiles[RoleKey.TESTER] == world.fast_id + + +async def test_bootstrap_without_grants_creates_an_inactive_team(world, session_factory): + async with session_factory() as session: + manager = await _manager(world, session, scripted=ScriptedModel(), grants=FakeGrants({})) + await manager.bootstrap_team(organization_id=world.organization_id, project_id=None) + await session.commit() + employees = await EmployeeRepository(session).list_for_org(world.organization_id) + threads = await ThreadRepository(session).list_for_org(world.organization_id) + notes: list[str] = [] + for thread in threads: + notes.extend( + m.content for m in await MessageRepository(session).list_for_thread(thread.id) + ) + + assert len(employees) == 4 + assert all(e.status.value == "paused" for e in employees) + assert NO_MODEL_GRANTED_MESSAGE in notes diff --git a/apps/api/tests/test_engine_protocol.py b/apps/api/tests/test_engine_protocol.py index 93bc9ee..370e606 100644 --- a/apps/api/tests/test_engine_protocol.py +++ b/apps/api/tests/test_engine_protocol.py @@ -37,6 +37,7 @@ from app.domain.workforce.models import RoleKey from app.domain.workforce.pipeline import Mode MANAGER_TRIAGE = {"role": RoleKey.MANAGER, "mode": Mode.TRIAGE} +MANAGER_FINAL = {"role": RoleKey.MANAGER, "mode": Mode.FINAL} MANAGER_CHAT = {"role": RoleKey.MANAGER, "mode": Mode.CHAT} DEVELOPER = {"role": RoleKey.DEVELOPER, "mode": Mode.WORK} PLANNER = {"role": RoleKey.PLANNER, "mode": Mode.WORK} @@ -249,6 +250,23 @@ def test_assign_requires_an_employee_or_a_task(): assign_of(tag) +def test_an_empty_employee_means_pick_for_me(): + """``employee=""`` is a *deliberate* value, unlike an absent attribute.""" + tag = parse_reply('[[ASSIGN employee="" pipeline="plan_code_test"/]]', **MANAGER_TRIAGE).first( + TagName.ASSIGN + ) + assign = assign_of(tag) + assert assign.employee == "" + assert assign.pipeline == "plan_code_test" + + +def test_revise_carries_a_body(): + """The prompt asks for ``[[REVISE]]reason[[/REVISE]]``; so does the parser.""" + reply = parse_reply("[[REVISE]]تست‌ها قرمز است[[/REVISE]]", **MANAGER_FINAL) + assert reply.errors == () + assert reply.first(TagName.REVISE).body == "تست‌ها قرمز است" + + def test_assign_refuses_an_unknown_pipeline(): reply = parse_reply('[[ASSIGN employee="سارا" pipeline="plan_ship_test"/]]', **MANAGER_TRIAGE) with pytest.raises(ProtocolError): diff --git a/apps/api/tests/test_engine_runner.py b/apps/api/tests/test_engine_runner.py index ca2f295..eb1d189 100644 --- a/apps/api/tests/test_engine_runner.py +++ b/apps/api/tests/test_engine_runner.py @@ -407,6 +407,33 @@ async def test_the_tool_call_cap_stops_a_runaway_turn(_db, session_factory, monk assert result.error_kind == "protocol" +async def test_a_model_that_repeats_its_protocol_mistake_gives_up( + _db, session_factory, monkeypatch +): + """A malformed reply is retried, but not forever. + + The regression: a model that answers with a tag its role may not use got + the error quoted back and another chance, with no cap — so a model that + repeats itself burned paid calls until the 900 s turn timeout. + """ + monkeypatch.setattr(settings, "engine_max_protocol_retries", 2) + # A tag the developer may not use: refused every single time. + scripted = ScriptedModel([ScriptedReply(content="[[ASSIGN employee=\"سارا\"/]]")] * 9) + async with session_factory() as session: + runner = _runner( + session, model=scripted, sandbox=FakeSandbox(), grants=FakeGrants([_db.model_id]) + ) + result = await runner.run_turn(await _request(_db, **DEVELOPER)) + await session.commit() + + assert result.ok is False + assert result.error_kind == "protocol" + assert result.incomplete is True + # 1 malformed reply + 2 retries, then it stops: never the whole script. + assert result.model_calls == 3 + assert scripted.calls == 3 + + async def test_the_turn_timeout_stops_a_slow_loop(_db, session_factory, monkeypatch): monkeypatch.setattr(settings, "engine_turn_timeout_seconds", 0) scripted = ScriptedModel([ScriptedReply(content='[[LIST_DIR path="."/]]')] * 3) diff --git a/docs/plan/reports/agent-2-phase-3.md b/docs/plan/reports/agent-2-phase-3.md new file mode 100644 index 0000000..e908e7d --- /dev/null +++ b/docs/plan/reports/agent-2-phase-3.md @@ -0,0 +1,179 @@ +# Agent 2 — Phase 3 report: the manager (orders → tasks → team) + +**Date:** 2026-10-14 · **Branch:** `dev` · **Status:** done, all checks green + +## What was done + +### 3.1 Manager chat — `app/application/engine/manager.py` (new) + +`ManagerTurns.chat()` is the manager's voice on a thread. `WorkforceService.post_user_message` +was rewritten to delegate to it (same signature, same session, so the reply, the tasks +and the usage events commit together). + +- The reply the user reads is the **prose outside the tags**; the tags are executed + beside it, one Persian "•" line per action. A malformed reply is quoted back to + the model once (`CHAT_MAX_ATTEMPTS = 2`, `CHAT_TURN_TIMEOUT_SECONDS = 120`) — but a + reply that has protocol *errors* **and** usable tags is kept: the extras the per-reply + limit dropped (3 tasks / 3 team changes / 3 knowledge items) must not discard the work + that was valid. +- Executed tags: `CREATE_TASK` (title, kind, priority, `depends_on`, pipeline) → + `Task` + `task_pipeline_state` at `pending_manager_review` linked to the thread; + `PAUSE_TASK` / `RESUME_TASK` / `CANCEL_TASK`; `HIRE` / `UPDATE_EMPLOYEE` / `FIRE`; + `KNOWLEDGE` (3.5). `emit_task_created` goes out per new task, so the chat shows cards live. +- The prompt asks for **atomic tasks that do not touch the same files** (locked by + `test_the_chat_prompt_asks_for_atomic_tasks`). +- Chat usage is billed through the **one** model entry point (see "one billing path" below). + A chat turn deliberately writes **no** `engine_turns` row: that table has `task_id`/`run_id` + FKs and a chat has neither; the usage events and the messages are the record. + +### 3.2 Team management + +- `hire(role, tier, name)` only hires one of `HIRABLE_ROLES`, refuses past + `engine_max_employees`, counts **HIRE + UPDATE_EMPLOYEE + FIRE together** against + `MAX_TEAM_CHANGES_PER_TURN = 3`, and picks a granted model of the requested tier + **strictly** (`_pick_granted_model(..., fallback=False)`): a hire is an explicit + request, so a silent downgrade would lie to the manager. No granted model of that + tier → a Persian `ConflictError`. +- `fire()` refuses the **last manager**; a foreign employee id is a `NotFoundError` + (the org-scoped repository get is what enforces isolation). +- `update_tier()` re-points the employee profile's model and publishes `team.changed` + (`engine/events.py::emit_team_changed`, a generic realtime envelope — the frozen port + has no typed emitter for it). +- Users still hire/fire from the existing UI endpoints with **role + tier only**; a model + name never crosses the API. + +### 3.3 Triage turn (manager, mode triage) + +`triage()` reads the task, asks the model, and applies exactly one decision: + +- `ASSIGN employee="…" pipeline="…"` → `_apply_assign`: fuzzy name match **inside the org + only** (exact, then containment either way), or the **least busy** active non-manager + employee when `employee=""`; the pipeline is resolved from the manager's choice and the + task kind; the task moves to `pending_analysis` (or `pending_implementation` for + `code_test`), and `task_pipeline_state.pipeline` is patched. +- `REJECT` → `cancelled`; `ASK_USER` → `pending_user_decision` **and** the question is posted + in the task's thread as an assistant message (3.3). +- A reply with none of these is `acted=False` with the model's text as the reason. + +### 3.4 Final verdict and error handling + +- `final_review()` (mode final) sees the plan, diff, gate summary and test output: + `COMPLETE` → `merge_request`; `REVISE` → cycle + 1 and back to `pending_implementation`, + carrying the revision reason as the outcome message. +- `error_review()` (mode error): `RESOLVE_ERROR hint="…"` → retry the task with the hint, + `CANCEL` → `cancelled`. + +### 3.5 `KNOWLEDGE` → memory + +`_record_knowledge()` records every `KNOWLEDGE` item through `MemoryPort.record(kind=rule|fact, +source="manager")`, on **every** manager turn (chat, triage, final, error). + +### 3.6 Bootstrap — `WorkforceBootstrapPort` + +- `ManagerTurns.bootstrap_team(org, project)` creates one manager + planner + developer + + tester on the strongest / balanced / balanced / fast granted tiers, idempotent (it + returns as soon as the org has any employee). With nothing granted the employees are + created **inactive** and a system message `"منتظر تخصیص مدل توسط مالک سامانه"` + (`NO_MODEL_GRANTED_MESSAGE`) is posted, so the page explains itself. +- **`container_engine.py` now wires the real bootstrap** (it was the foundation's no-op + stub): `build_workforce_bootstrap(session)` returns an adapter that builds `ManagerTurns` + lazily from the caller's session — the same session the org wizard commits on, so the + team lands with the organization. Names and return types are the frozen contract; the + body is mine to fill (PLAN-v2 §6). + +### One billing path (agybot bug) + +`call_model()` and `BillingScope` were extracted in `engine/runner.py` as **the** single +model entry point; `TurnRunner._complete` and the manager chat both go through it, and it +bills via `record_model_usage`. A model call that skips it is a call the platform never +charges for — which is exactly what the old manager chat did. + +### Fixes to the Phase-1 protocol (prompt and protocol must agree) + +Three real bugs surfaced while making the manager tests pass; each has a regression test: + +1. **`[[REVISE]]reason[[/REVISE]]` was a protocol error.** `REVISE` was missing from + `BODY_TAGS`, so the closing tag was reported as "stray" even though `prompt.py` tells the + model to write exactly that. Added `REVISE` to `BODY_TAGS`. +2. **`ASSIGN employee=""` was rejected.** The docstring and `_pick_assignee("")` both mean + "pick the least busy one", but `assign_of` used the stripped value as the presence test. + It now distinguishes an **absent** attribute (still an error — `[[ASSIGN/]]`) from a + **present-but-empty** one. The triage prompt now documents the empty form. +3. **The runner retried protocol errors forever.** A model that repeated the same mistake was + fed the error back and called again with no cap, until the 900 s turn timeout — hundreds of + paid calls and an unbounded prompt. New setting `engine_max_protocol_retries = 3` bounds + it; exceeding it fails the turn with `error_kind=protocol`, `incomplete=True`. + +### New setting (inside the agent-2 block of `config.py`) + +`engine_max_protocol_retries`. + +### Other support changes + +- `TaskRepository.set_assignee()` (my repository file) — the assignee write is its own + narrow method rather than a whole-row `upsert`. +- `engine/events.py`: `emit_team_changed()`. +- `tests/engine_fakes.py`: `FakeGrants` now takes either a list of ids (all `balanced`) or an + id → tier mapping, so a test can have a strong and a fast model and assert which one the + engine picked. + +## Tests + +`tests/test_engine_manager.py` (new, **20 tests**): + +- 3.1: an order becomes tasks with pipeline state; the ≤ 3 cap (extras dropped, the valid + three kept and the user told); the prompt asks for atomic tasks; a chat model call is + billed; a chat without a grant answers with the waiting note. +- 3.2: hire picks a granted model of the tier; hire without that tier is refused; the + employee cap; the last manager cannot be fired; a foreign employee is a 404. +- 3.3: triage assigns the least busy employee; a named employee is matched inside the org; + the manager of org A can never assign org B's employee; `ASK_USER` parks the task and asks + in the thread. +- 3.4: `COMPLETE` → `merge_request`; `REVISE` → back and cycle + 1; `RESOLVE_ERROR` retries + with a hint. +- 3.5: knowledge tags become org memory (read back from `OrgMemoryRow`). +- 3.6: bootstrap creates the team and is idempotent; without grants it creates an inactive + team and the waiting note. + +Plus: `test_engine_runner.py` gains a regression test for the retry cap (a model that repeats +its mistake stops after `1 + engine_max_protocol_retries` calls), and `test_engine_protocol.py` +gains the two protocol-lock tests (`employee=""` is "pick for me"; `REVISE` carries a body). + +## Checks (PLAN-v2 §7.4) + +``` +cd apps/api && .venv/bin/python -m pytest -q → 722 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 158 source files +``` + +No web files were touched; no migration was needed (Phase 1's `a2b3c4d5e6f2` already has +`task_pipeline_state.organization_id` and the engine tables). + +## Deviations & notes + +1. `post_user_message` delegates to `ManagerTurns.chat`; the old `emit_chat_pending` / + `emit_message_created` calls on that path were **not** kept, because the engine's turn is + synchronous with the request and the thread reload shows the reply. Removing them is a + behaviour change to `workforce_service.py` (my file) and is covered by the workforce tests. +2. `ManagerTurns` is built inside `_manager_turns()` (and the bootstrap adapter) rather than + injected: `container.py` is frozen, so the manager must not widen a constructor it does + not own. Session-scoped ports (model grants, memory) are built from the shared session. +3. The manager chat writes no `engine_turns` row (no task/run FK); task turns do, through + `TurnRunner`. Recorded here because the phase-2 report implied all turns leave a row. +4. `ASSIGN employee=""` now means auto-pick; the triage prompt documents it, so prompt and + protocol still agree. +5. `tests/test_e2e_smoke.py` gained a model grant before the manager-chat step: PLAN-v2 D3 + says the engine runs only on granted models, so a chat with no grant now (correctly) + answers nothing but the waiting note. This is a test fix, not a behaviour regression. + +## Open questions / CHANGE REQUESTs + +- None. The org automation policy (`automation_enabled`) is still behind + `ModelGrantPort`, as planned: Phase 4 assumes enabled until the CHANGE REQUEST lands. + +## Next + +Phase 4 — `engine/scheduler.py::tick()`, worktree lifecycle, the gate chain (diff guard, +no-progress, tests, green gate), commit/merge/push, budget pause, and +`tests/test_engine_flow.py`.