diff --git a/README.md b/README.md index 47dddaf..2f7e677 100644 --- a/README.md +++ b/README.md @@ -238,13 +238,35 @@ web/ | قابلیت | env | مقادیر | |---|---|---| | Git | `GIT_PROVIDER` | `mock` \| `gitea` | +| Workspace گیت | `GIT_WORKSPACE` | `fake` (پیش‌فرض، برای تست) \| `local` (git روی host) | | دامنهٔ تست | `DOMAIN_ROUTER` | `mock` \| `caddy` | -| Sandbox | `SANDBOX_PROVIDER` | `mock` \| `daytona` \| `e2b` | +| دامنهٔ سفارشی | `ALLOW_CUSTOM_DOMAINS` | `false` (پیش‌فرض) \| `true` | +| Sandbox | `SANDBOX_PROVIDER` | `mock` \| `docker` \| `daytona` \| `e2b` | +| استقرار روی شبکهٔ خصوصی | `DEPLOY_ALLOW_PRIVATE_HOSTS` | `false` (پیش‌فرض)؛ `true` فقط برای مقصدی در شبکهٔ خصوصی که قابل اعتماد است | +| MCP gateway | `MCP_GATEWAY_ENABLED` | `true` (پیش‌فرض) \| `false` | | پرداخت | `PAYMENT_PROVIDER` | `mock` \| `stripe` | اگر credential واقعی نداریم، **قابلیت حذف نمی‌شود**: مدل دیتابیس، interface، UI و adapter mock کامل ساخته می‌شود و adapter واقعی پشت همین flag آماده می‌ماند. +با پیش‌فرض‌ها (`mock`، `fake` و `false`) تست‌ها بدون docker، git یا شبکه اجرا می‌شوند. + +**حالت واقعی (real mode):** وقتی سامانه باید روی یک host واقعی با docker و caddy اجرا شود، این envها را تنظیم کنید: + +```bash +SANDBOX_PROVIDER=docker GIT_WORKSPACE=local DOMAIN_ROUTER=caddy GIT_PROVIDER=gitea +``` + +نیازمندی‌های host در این حالت: + +- `docker` یا `podman` در `PATH` (با `SANDBOX_DOCKER_BIN` انتخاب می‌شود). کد کاربر فقط داخل کانتینرهای سخت‌شده اجرا می‌شود: `--read-only`، کاربر غیر root، `--cap-drop ALL`، و محدودیت CPU، حافظه و PID. +- `git` نسخهٔ 2.20 یا جدیدتر در `PATH`. git را خود پلتفرم اجرا می‌کند؛ hookها غیرفعال‌اند و اعتبارنامه‌ها هرگز در argv یا URL نمی‌آیند. +- `caddy` در `PATH` و این خط در Caddyfile: `import /etc/caddy/agy-preview/*.caddy`. بدون این خط adapter اجرا نمی‌شود، تا یک موفقیت کاذب گزارش نشود. +- اجازهٔ بدون رمز برای `caddy reload --config ` به کاربر سرویس (روی systemd با sudoers یا polkit، نه با root). +- پوشه‌های قابل‌نوشتن با فضای کافی زیر `SANDBOX_WORKSPACES_ROOT` و `ARTIFACTS_ROOT` (پیش‌فرض `./.data/…`). پاک‌سازی خودکار `runs/`، `artifacts/` و فایل‌های پشتیبان FTP هنوز پیاده نشده است. +- هدف استقرار `ftps` (TLS صریح) است. هدف `ftp` ساده پذیرفته می‌شود، ولی بدون رمزنگاری است. +- تست‌های یکپارچگی docker و استقرار با `SANDBOX_DOCKER_IT=1` و `DEPLOY_IT=1` فعال می‌شوند و بدون آن‌ها رد (skip) می‌شوند. + ### ارائه‌دهندهٔ agy (Google Antigravity) — تنها ارائه‌دهندهٔ مدل agy کلید API ندارد؛ **اکانت**‌ها هویت Google هستند و از طریق OAuth وارد @@ -269,26 +291,24 @@ refresh tokenها فقط encrypted at rest ذخیره می‌شوند و هیچ ## ۹. Workerهای پس‌زمینه -سه worker وجود دارد. همه در `apps/api/app/workers/` پیاده شده‌اند و `services/` -پوستهٔ اجرای آن‌هاست: +پنج process پس‌زمینه وجود دارد: چهار worker و یک MCP gateway. همه در `apps/api/app/workers/` پیاده شده‌اند (MCP gateway در `apps/api/app/interfaces/mcp/`) و `services/` پوستهٔ اجرای آن‌هاست: | worker | کار | دستور | |---|---|---| -| `billing-worker` | جمع‌بندی usage ledger به فاکتور draft (هر دوره یک‌بار، idempotent) | `pnpm worker:billing` | -| `ai-runtime` | برداشتن taskهای در صف و اجرای آن‌ها از مسیر run engine | `pnpm worker:ai-runtime` | -| `sandbox-gateway` | متوقف‌کردن sessionهای بی‌استفاده و ثبت audit | `pnpm worker:sandbox-gateway` | +| `billing-worker` | جمع‌بندی usage ledger به فاکتور draft (هر دوره یک‌بار، idempotent)، و در همان پاس حذف نرم درس‌های حافظه‌ای منقضی‌شده (به‌جز موارد pin شده) | `pnpm worker:billing` | +| `ai-runtime` | برداشتن taskهای در صف و اجرای آن‌ها از مسیر run engine، و یک `Scheduler.tick()` در هر پاس (فاصلهٔ پاس‌ها با `ENGINE_TICK_SECONDS`) | `pnpm worker:ai-runtime` | +| `sandbox-gateway` | متوقف‌کردن sessionهای بی‌استفاده، حذف containerهای یتیم با label `smart-org.*`، و ثبت audit | `pnpm worker:sandbox-gateway` | +| `delivery` | اجرای pipeline runهای در صف (از checkout تا route) و بررسی ریموت‌ها برای pushهای بیرونی هر `DELIVERY_POLL_SECONDS` | `pnpm worker:delivery` | -هر دستور **یک پاس** اجرا می‌کند و خارج می‌شود (شکل مناسب cron). برای اجرای -مستمر `--loop` بدهید: +**MCP gateway** یک سرویس مداوم است که با `pnpm mcp:dev` اجرا می‌شود. عامل‌های بیرونی AI با توکن‌های محدود به یک سازمان به آن وصل می‌شوند. جزئیات در `services/mcp-gateway/README.md`. + +هر دستور worker **یک پاس** اجرا می‌کند و خارج می‌شود (شکل مناسب cron). برای اجرای مستمر `--loop` بدهید: ```bash cd apps/api && .venv/bin/python -m app.workers billing --loop --interval 3600 ``` -**اصل مهم:** workerها هیچ منطق تجاری ندارند. هر سه از -`app/application/container.py` همان سرویس‌هایی را می‌سازند که HTTP API می‌سازد، -پس worker و API هرگز نمی‌توانند دربارهٔ اینکه کدام adapter پشت یک port است اختلاف -داشته باشند. جزئیات هر کدام در `services/*/README.md`. +**اصل مهم:** workerها هیچ منطق تجاری ندارند. `ai-runtime`، `sandbox-gateway` و `delivery` سرویس‌های خود را از همان builderهای `app/application/container*.py` می‌سازند که HTTP API می‌سازد، پس آن‌ها و API هرگز نمی‌توانند دربارهٔ اینکه کدام adapter پشت یک port است اختلاف داشته باشند. `billing-worker` مستقیم از repositoryها استفاده می‌کند. جزئیات هر کدام در `services/*/README.md`. --- diff --git a/apps/api/app/application/container.py b/apps/api/app/application/container.py index bc3c592..7b6274e 100644 --- a/apps/api/app/application/container.py +++ b/apps/api/app/application/container.py @@ -22,8 +22,13 @@ from app.application.container_delivery import ( build_deploy_target_tester, build_repo_workspace, ) -from app.application.container_engine import build_workforce_bootstrap -from app.application.container_org import build_memory, build_model_grants +from app.application.container_engine import build_scheduler, build_workforce_bootstrap +from app.application.container_org import ( + build_memory, + build_memory_service, + build_model_grant_service, + build_model_grants, +) from app.application.services.billing_service import BillingService from app.application.services.knowledge_service import KnowledgeService from app.application.services.project_service import ProjectService @@ -172,8 +177,11 @@ __all__ = [ "build_deploy_service", "build_deploy_target_tester", "build_memory", + "build_memory_service", + "build_model_grant_service", "build_model_grants", "build_repo_workspace", + "build_scheduler", "build_workforce_bootstrap", "build_knowledge_service", "build_project_service", diff --git a/apps/api/app/application/container_engine.py b/apps/api/app/application/container_engine.py index 497cb11..610c52f 100644 --- a/apps/api/app/application/container_engine.py +++ b/apps/api/app/application/container_engine.py @@ -13,7 +13,8 @@ from typing import TYPE_CHECKING from sqlalchemy.ext.asyncio import AsyncSession from app.application.ports.workforce_bootstrap import WorkforceBootstrapPort -from app.domain.sandbox.models import SessionStatus +from app.core.errors import AppError +from app.domain.sandbox.models import SessionStatus, WorkspaceStatus from app.domain.workforce.models import Task if TYPE_CHECKING: @@ -87,11 +88,13 @@ def build_workforce_bootstrap(session: AsyncSession) -> WorkforceBootstrapPort: class _EngineSandboxSessions: """The sandbox session the engine's turns run in. - The engine does not *create* sandboxes: a workspace and a running session - belong to the project and are started through the sandbox flow. The engine - only reuses the active session of the task's project. When there is none it - returns ``""``, and the tests gate treats that as red and sends the work - back — never "merge untested because there was no container". + The engine reuses the active session of the task's project. When there is + none, it starts the project's workspace session itself (see ``_provision``): + a fresh organization cannot start a sandbox through the API, because sandbox + mutations are owner-only, and without a session every tool turn fails. When + the session cannot be started it returns ``""``, and the tests gate treats + that as red and sends the work back — never "merge untested because there + was no container". """ #: A session in one of these states can still run a command. @@ -107,6 +110,7 @@ class _EngineSandboxSessions: self.session = session async def for_task(self, task: Task) -> str: + from app.infrastructure.repositories.projects import ProjectRepository from app.infrastructure.repositories.sandbox import ( SandboxSessionRepository, WorkspaceRepository, @@ -114,9 +118,12 @@ class _EngineSandboxSessions: if not task.project_id: return "" - workspaces = await WorkspaceRepository(self.session).list_for_project( - task.project_id - ) + # A task only ever runs in its own organization's project, whatever the row says. + own = await ProjectRepository(self.session).list_for_org(task.organization_id) + if task.project_id not in {project.id for project in own}: + return "" + found = await WorkspaceRepository(self.session).list_for_project(task.project_id) + workspaces = [w for w in found if w.organization_id == task.organization_id] sessions = SandboxSessionRepository(self.session) for workspace in workspaces: for sandbox_session in await sessions.list_for_workspace(workspace.id): @@ -125,7 +132,66 @@ class _EngineSandboxSessions: and sandbox_session.external_session_id ): return sandbox_session.external_session_id - return "" + return await self._provision(task) + + async def _provision(self, task: Task) -> str: + """Start the project's workspace session and return its id, or ``""``. + + A project keeps one workspace: the newest one that is not destroyed is + reused, and only a project with none gets a new workspace, built from the + platform template for the configured provider. A failure is recorded and + never raised, so the turn runs without a container and the gates say so. + """ + from app.application.container import build_sandbox_service + from app.core.config import settings + from app.infrastructure.error_log import record_error + from app.infrastructure.repositories.sandbox import ( + SandboxTemplateRepository, + WorkspaceRepository, + ) + + project_id = task.project_id + if not project_id: + return "" + metadata = {"task_id": task.id, "project_id": project_id} + service = build_sandbox_service(self.session) + try: + found = await WorkspaceRepository(self.session).list_for_project(project_id) + workspaces = [w for w in found if w.organization_id == task.organization_id] + workspace = next( + (w for w in workspaces if w.status != WorkspaceStatus.DESTROYED), None + ) + if workspace is None: + templates = await SandboxTemplateRepository(self.session).list() + if not templates: + await record_error( + kind="engine.sandbox_provision_failed", + message="no sandbox template is seeded", + organization_id=task.organization_id, + metadata=metadata, + ) + return "" + preferred = [ + t for t in templates if str(t.provider) == settings.sandbox_provider + ] + workspace = await service.create_workspace( + organization_id=task.organization_id, + project_id=project_id, + template_id=(preferred or templates)[0].id, + ) + started = await service.start_session( + organization_id=task.organization_id, workspace_id=workspace.id + ) + except AppError as exc: + await record_error( + kind="engine.sandbox_provision_failed", + message=type(exc).__name__, + exc=exc, + organization_id=task.organization_id, + metadata=metadata, + ) + return "" + return started.external_session_id or "" def _build_manager( diff --git a/apps/api/app/application/container_org.py b/apps/api/app/application/container_org.py index c5b37bc..f1f4552 100644 --- a/apps/api/app/application/container_org.py +++ b/apps/api/app/application/container_org.py @@ -32,6 +32,9 @@ from app.infrastructure.repositories.org_grants import ( def build_memory(session: AsyncSession) -> MemoryPort: """The real memory service: recall, teach, and one reflection per task. + The reflection runs only on a model the organization was granted, so the + owner's grants decide which model learns on the organization's behalf. + ``get_model_provider`` is resolved from ``container``'s namespace rather than from the registry directly: that module is what the composition root re-exports, so a deployment (or the test suite) that swaps the adapter in @@ -39,15 +42,23 @@ def build_memory(session: AsyncSession) -> MemoryPort: """ from app.application.container import get_model_provider - return build_memory_service(session, provider=get_model_provider()) + return build_memory_service( + session, provider=get_model_provider(), grants=build_model_grants(session) + ) def build_memory_service( session: AsyncSession, *, provider=None, + grants: ModelGrantPort | None = None, ) -> MemoryService: - """The same object, typed for the router (list, teach, pin, confirm, delete).""" + """The same object, typed for the router (list, teach, pin, confirm, delete). + + ``grants`` is optional. The router never reflects, and a direct construction + without it keeps the routing-table path; the engine's path (``build_memory``) + always passes it. + """ if provider is None: from app.application.container import get_model_provider @@ -60,6 +71,7 @@ def build_memory_service( rate_cards=RateCardRepository(session), ledger=UsageLedgerRepository(session), provider=provider, + grants=grants, ) diff --git a/apps/api/app/application/engine/manager.py b/apps/api/app/application/engine/manager.py index 5074445..60f3479 100644 --- a/apps/api/app/application/engine/manager.py +++ b/apps/api/app/application/engine/manager.py @@ -302,11 +302,12 @@ class ManagerTurns: raise ValidationError("این نقش قابل استخدام نیست.") if not tier: raise ValidationError("سطح کارمند را انتخاب کنید.") + # The owner's per-org policy may lower the global cap, never raise it. + policy = await self.grants.get_org_policy(organization_id) + cap = min(settings.engine_max_employees, policy.max_employees) count = await self.employees.count_for_org(organization_id) - if count >= settings.engine_max_employees: - raise ConflictError( - f"سقف تعداد کارمندان ({settings.engine_max_employees}) پر شده است." - ) + if count >= cap: + raise ConflictError(f"سقف تعداد کارمندان ({cap}) پر شده است.") granted = await self.grants.granted_models(organization_id=organization_id) # A hire is an explicit request for a tier: no silent downgrade, or the # manager would think it hired a strong employee and got a fast one. @@ -529,12 +530,14 @@ class ManagerTurns: if draft.kind in {k.value for k in TaskKind} else TaskKind.FEATURE ) + # A thread made by the web UI has no project; the org's project is the fallback. + project_id = thread.project_id or await self._project_id_for(organization_id) task = await self.tasks.create( Task( organization_id=organization_id, title=draft.title[:200], description=draft.description, - project_id=thread.project_id, + project_id=project_id, thread_id=thread.id, created_by_user_id=user_id, ) @@ -544,9 +547,11 @@ class ManagerTurns: TaskPipelineState( task_id=task.id, organization_id=organization_id, - project_id=thread.project_id, + project_id=project_id, pipeline=resolve_pipeline( - manager_choice=draft.pipeline or None, task_kind=kind + manager_choice=draft.pipeline or None, + task_kind=kind, + has_active_reviewer=await self._has_active_reviewer(organization_id), ), kind=kind, risk=Risk.MEDIUM, @@ -719,6 +724,7 @@ class ManagerTurns: pipeline = resolve_pipeline( manager_choice=assign.pipeline or None, task_kind=state.kind if state else None, + has_active_reviewer=await self._has_active_reviewer(task.organization_id), ) first_stage = ( Stage.PENDING_ANALYSIS @@ -1067,6 +1073,13 @@ class ManagerTurns: return employee.project_id return None + async def _has_active_reviewer(self, organization_id: str) -> bool: + """A review stage needs an active reviewer to act on it, or the task dead-ends.""" + return any( + e.role_key is RoleKey.REVIEWER and e.is_available + for e in await self.employees.list_for_org(organization_id) + ) + __all__ = [ "BOOTSTRAP_TIERS", diff --git a/apps/api/app/application/engine/scheduler.py b/apps/api/app/application/engine/scheduler.py index b25d5a3..cc53d7f 100644 --- a/apps/api/app/application/engine/scheduler.py +++ b/apps/api/app/application/engine/scheduler.py @@ -31,6 +31,7 @@ from app.application.engine.runner import TurnRequest, TurnRunner from app.application.engine.state import advance, park_dependents from app.application.ports.delivery_hooks import DeliveryHooks, NullDeliveryHooks from app.application.ports.memory import MemoryPort, NullMemory, TaskOutcome +from app.application.ports.model_grants import OrgPolicyView from app.application.ports.repo_workspace import RepoWorkspacePort, WorktreeRef from app.application.ports.sandbox_provider import SandboxProviderPort from app.core.config import settings @@ -165,6 +166,11 @@ class Scheduler: turn_states = await self.states.list_at_stages(TURN_STAGES, limit=limit) merge_states = await self.states.list_at_stages(MERGE_STAGES, limit=limit) + # One policy read per organization for the whole tick, shared by both lists. + policies: dict[str, OrgPolicyView] = {} + turn_states = await self._drop_switched_off(turn_states, policies, report) + merge_states = await self._drop_switched_off(merge_states, policies, report) + selected = self._select(turn_states) report.selected = len(selected) for state in selected: @@ -174,6 +180,30 @@ class Scheduler: await self._guarded(self._merge_step, state, report) return report + async def _drop_switched_off( + self, + states: list[TaskPipelineState], + policies: dict[str, OrgPolicyView], + report: TickReport, + ) -> list[TaskPipelineState]: + """Keep only the states whose organization has automation switched on. + + The owner's switch stops everything up to the merge. A skipped task keeps + its stage, so switching automation back on resumes it where it was. + """ + kept: list[TaskPipelineState] = [] + for state in states: + if state.organization_id not in policies: + policies[state.organization_id] = await self.manager.grants.get_org_policy( + state.organization_id + ) + if not policies[state.organization_id].automation_enabled: + report.skipped += 1 + report.note(f"task {state.task_id}: اتوماسیون این سازمان خاموش است") + continue + kept.append(state) + return kept + async def _guarded(self, step, state: TaskPipelineState, report: TickReport) -> None: """One task must never stop the tick: catch, record, continue.""" try: @@ -590,16 +620,32 @@ class Scheduler: await self._post_note(task, reason) async def _budget_reason(self, task: Task) -> str | None: + """Why this task may not spend a turn now, or ``None`` when it may. + + Usage is post-paid: the ledger feeds invoices and nothing debits a wallet + from usage. A zero balance is therefore the normal starting state of every + org, not "out of money", so only an overdrawn wallet (below zero) pauses + work. The owner's spend lever is the org's monthly credit budget. + """ if self.wallets is not None: wallet = await self.wallets.get_for_org(task.organization_id) # No wallet means no budget was ever set: a fresh org must not be - # frozen out of its first task. - if wallet is not None and wallet.balance <= 0: + # frozen out of its first task. A zero balance is not a pause either. + if wallet is not None and wallet.balance < 0: return BUDGET_PAUSED_MESSAGE if self.spend is not None and settings.engine_max_task_credits > 0: used = await self.spend.spend_for_task(task.id) if used >= settings.engine_max_task_credits: return "سقف اعتبار این کار پر شده است؛ کار متوقف شد." + if self.spend is not None: + policy = await self.manager.grants.get_org_policy(task.organization_id) + budget = policy.monthly_credit_budget + if budget is not None: + # The month is counted in UTC, the clock the ledger stamps rows with. + month_start = utcnow().replace(day=1, hour=0, minute=0, second=0, microsecond=0) + spent = await self.spend.spend_for_org_since(task.organization_id, month_start) + if spent >= budget: + return "سقف اعتبار ماهانهٔ سازمان پر شده است؛ کار متوقف شد." return None async def _retry_or_fail( diff --git a/apps/api/app/application/ports/model_grants.py b/apps/api/app/application/ports/model_grants.py index b0adc69..cc73a1d 100644 --- a/apps/api/app/application/ports/model_grants.py +++ b/apps/api/app/application/ports/model_grants.py @@ -2,6 +2,10 @@ Implemented by the org-platform workstream; used by the engine. FROZEN contract: see docs/plan/PLAN-v2.md §5. ``tier`` is a neutral label; users never see model names. + +The contract was extended at integration: ``get_org_policy`` gives the engine the owner's +per-organization policy (automation switch, employee cap, monthly budget). The port only +reads it; the engine applies its own system ceiling (``settings.engine_max_employees``). """ from __future__ import annotations @@ -27,12 +31,48 @@ def tier_for(*, capabilities: list[str] | tuple[str, ...], output_price_per_1m: return "fast" if output_price_per_1m < 2.0 else "balanced" +class OrgPolicyView(Protocol): + """Read-only view of one organization's owner policy. + + Properties rather than attributes, so both a mutable dataclass (``ModelGrantService``'s + ``PolicyView``) and a frozen one (``DefaultOrgPolicy``) satisfy it. + """ + + @property + def max_employees(self) -> int: ... + + @property + def automation_enabled(self) -> bool: ... + + @property + def mcp_enabled(self) -> bool: ... + + @property + def monthly_credit_budget(self) -> float | None: ... + + +@dataclass(slots=True, frozen=True, kw_only=True) +class DefaultOrgPolicy: + """The policy for an organization the owner has not configured. + + ``max_employees`` is only the port's default. The engine caps it with + ``settings.engine_max_employees`` itself, so this module does not import settings. + """ + + max_employees: int = 20 + automation_enabled: bool = True + mcp_enabled: bool = True + monthly_credit_budget: float | None = None + + @runtime_checkable class ModelGrantPort(Protocol): async def granted_models(self, *, organization_id: str) -> list[GrantedModel]: ... async def is_granted(self, *, organization_id: str, ai_model_id: str) -> bool: ... + async def get_org_policy(self, organization_id: str) -> OrgPolicyView: ... + class AllActiveModelsGrants: """Today's behaviour: every active model is usable by every organization. @@ -48,3 +88,6 @@ class AllActiveModelsGrants: async def is_granted(self, *, organization_id: str, ai_model_id: str) -> bool: return any(m.ai_model_id == ai_model_id for m in await self._list_active()) + + async def get_org_policy(self, organization_id: str) -> OrgPolicyView: + return DefaultOrgPolicy() diff --git a/apps/api/app/application/services/memory_service.py b/apps/api/app/application/services/memory_service.py index c2dd81c..e7b3cd3 100644 --- a/apps/api/app/application/services/memory_service.py +++ b/apps/api/app/application/services/memory_service.py @@ -13,9 +13,10 @@ task — and this service owns three invariants around those calls: whose key) already exists updates that entry instead of adding a duplicate, which is also what lets "the port is 8001" correct "the port is 8000". 3. **A reflection can never break the engine.** ``on_task_finished`` asks one - cheap model for what the task taught. If the call fails, the error is - recorded and the function returns: the task is finished, and the org simply - learned nothing this time. + cheap model the organization is granted for what the task taught. If that + cannot happen (no grant, a failing call), the reason is recorded and the + function returns: the task is finished, and the org simply learned nothing + this time. Nothing here returns a model name, a provider or an infrastructure detail — the engine's contract is a block of text. @@ -27,11 +28,12 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.application.ports import ChatMessage, ModelProviderPort, UsageRecord from app.application.ports.memory import TaskOutcome +from app.application.ports.model_grants import GrantedModel, ModelGrantPort from app.application.ports.model_provider import ModelResponse from app.core.config import settings from app.core.errors import NotFoundError, PermissionDeniedError, ValidationError from app.core.redact import redact_text -from app.domain.ai_control.models import RateUnit, RoutingMode +from app.domain.ai_control.models import AiModel, RateUnit, RoutingMode from app.domain.base import utcnow from app.domain.identity.models import ROLE_RANK, Role from app.domain.memory.models import ( @@ -100,6 +102,7 @@ class MemoryService: rate_cards: RateCardRepository, ledger: UsageLedgerRepository, provider: ModelProviderPort, + grants: ModelGrantPort | None = None, ) -> None: self.session = session self.memories = memories @@ -108,6 +111,7 @@ class MemoryService: self.rate_cards = rate_cards self.ledger = ledger self.provider = provider + self.grants = grants # ------------------------------------------------------------------ # # MemoryPort # @@ -371,16 +375,8 @@ class MemoryService: async def _reflect(self, *, organization_id: str, task: TaskOutcome) -> str: """One cheap completion about the finished task, priced in the ledger.""" - model = await self.models.pick_for_mode(RoutingMode.CHEAP) + model = await self._reflection_model(organization_id=organization_id, task=task) if model is None: - # No catalogue means nothing to route to. Record it once rather than - # letting the engine wonder why nobody learned anything. - await record_error( - kind="memory.reflection_no_model", - message="no active model for the cheap routing mode", - organization_id=organization_id, - metadata={"task_id": task.task_id}, - ) return "" pool = await self.credentials.list_active_for_provider(model.provider_id) response = await self.provider.complete( @@ -406,6 +402,49 @@ class MemoryService: await self._record_usage(organization_id=organization_id, response=response) return response.content or "" + async def _reflection_model(self, *, organization_id: str, task: TaskOutcome) -> AiModel | None: + """The model one reflection runs on, or ``None`` after recording why not. + + With a grant port, an organization reflects only on a model its owner + granted it, cheapest tier first. Without one (a direct construction), the + routing table's cheap mode decides, as it always has. + """ + if self.grants is None: + model = await self.models.pick_for_mode(RoutingMode.CHEAP) + if model is None: + # No catalogue means nothing to route to. Record it once rather than + # letting the engine wonder why nobody learned anything. + await record_error( + kind="memory.reflection_no_model", + message="no active model for the cheap routing mode", + organization_id=organization_id, + metadata={"task_id": task.task_id}, + ) + return model + chosen = _preferred_grant( + await self.grants.granted_models(organization_id=organization_id) + ) + if chosen is None: + # The owner granted this organization no model, so it learns nothing + # from this task. That is a policy outcome, and it is recorded as one. + await record_error( + kind="memory.reflection_no_grant", + message="no model is granted to this organization", + organization_id=organization_id, + metadata={"task_id": task.task_id}, + ) + return None + model = await self.models.get(chosen.ai_model_id) + if model is None or not model.is_active: + await record_error( + kind="memory.reflection_model_unavailable", + message="the granted model is missing or inactive", + organization_id=organization_id, + metadata={"task_id": task.task_id, "ai_model_id": chosen.ai_model_id}, + ) + return None + return model + async def _record_usage(self, *, organization_id: str, response: ModelResponse) -> None: """Price a reflection exactly like any other model call.""" for unit, quantity in ( @@ -464,6 +503,20 @@ def _kind_order(kind: MemoryKind) -> int: return _KIND_ORDER.get(kind, 9) +#: The tiers a reflection prefers, cheapest first. A reflection is one short +#: completion; a strong model would charge more for the same two lessons. +_REFLECTION_TIERS = ("fast", "balanced") + + +def _preferred_grant(granted: list[GrantedModel]) -> GrantedModel | None: + """The first granted model of the cheapest preferred tier, else the first one.""" + for tier in _REFLECTION_TIERS: + for grant in granted: + if grant.tier == tier: + return grant + return granted[0] if granted else None + + def _confidence_for(source: MemorySource, fallback: float) -> float: """Reflections get half confidence; a human's entry keeps its own.""" if source is MemorySource.REFLECTION: diff --git a/apps/api/app/application/services/org_setup_service.py b/apps/api/app/application/services/org_setup_service.py index a355f5c..81cb463 100644 --- a/apps/api/app/application/services/org_setup_service.py +++ b/apps/api/app/application/services/org_setup_service.py @@ -42,6 +42,7 @@ from app.domain.projects.models import ( ) from app.infrastructure.db.models import CreditWalletRow, ProjectRow from app.infrastructure.db.models_org import OrgSetupDraftRow +from app.infrastructure.error_log import record_error from app.infrastructure.repositories.app_profile import EDITABLE_FIELDS from app.infrastructure.repositories.identity import MembershipRepository, OrganizationRepository from app.infrastructure.repositories.projects import ( @@ -251,10 +252,9 @@ class OrgSetupService: # -- commit so hooks see a consistent world -------------------------- await self.session.commit() - # -- post-commit hooks ----------------------------------------------- - await self._call_hooks(organization_id=org.id, project_id=project_row.id) - - return { + # The response is read before the hooks run: a failed hook rolls the session + # back, and that expires the project row this method still reads. + response = { "organization": { "id": org.id, "name": org.name, @@ -268,6 +268,10 @@ class OrgSetupService: "deploy_target_id": deploy_target_id, } + # -- post-commit hooks ----------------------------------------------- + await self._call_hooks(organization_id=org.id, project_id=project_row.id) + return response + async def _create_single_app( self, org_id: str, @@ -387,42 +391,50 @@ class OrgSetupService: exists without its grants would be an organization whose AI staff cannot think. An empty default set is legitimate (the owner granted nothing). """ - from app.application.services.model_grant_service import ModelGrantService - from app.infrastructure.repositories.org_grants import ( - DefaultModelGrantRepository, - OrgModelGrantRepository, - OrgPolicyRepository, - ) + from app.application.container_org import build_model_grant_service - service = ModelGrantService( - session=self.session, - grants=OrgModelGrantRepository(self.session), - defaults=DefaultModelGrantRepository(self.session), - policies=OrgPolicyRepository(self.session), - ) + service = build_model_grant_service(self.session) return await service.copy_defaults_into( organization_id=organization_id, actor_user_id=user_id ) async def _call_hooks(self, *, organization_id: str, project_id: str) -> None: + """Run the post-commit hooks. A failing hook is recorded, never raised. + + A successful hook is committed before the next one runs, so a later failure + cannot roll back earlier work. A failed hook is rolled back, which leaves the + session usable for the next hook and for the response. + """ if self.workforce_bootstrap: try: await self.workforce_bootstrap.bootstrap_team( organization_id=organization_id, project_id=project_id ) - except Exception as exc: # noqa: BLE001 - logger.warning("Workforce bootstrap failed for %s: %s", organization_id, exc) + await self.session.commit() + except Exception as exc: # noqa: BLE001 - a hook must never fail the wizard + await record_error( + kind="wizard.bootstrap_failed", + message=type(exc).__name__, + exc=exc, + organization_id=organization_id, + metadata={"project_id": project_id}, + ) + await self.session.rollback() if self.delivery_hooks: try: await self.delivery_hooks.on_app_created( organization_id=organization_id, project_id=project_id ) - except Exception as exc: # noqa: BLE001 - logger.warning( - "Delivery hook on_app_created failed for %s: %s", - organization_id, - exc, + await self.session.commit() + except Exception as exc: # noqa: BLE001 - a hook must never fail the wizard + await record_error( + kind="wizard.on_app_created_failed", + message=type(exc).__name__, + exc=exc, + organization_id=organization_id, + metadata={"project_id": project_id}, ) + await self.session.rollback() # ------------------------------------------------------------------ # App profile editing (post-wizard) diff --git a/apps/api/app/application/services/workforce_service.py b/apps/api/app/application/services/workforce_service.py index df99f4e..84ab49e 100644 --- a/apps/api/app/application/services/workforce_service.py +++ b/apps/api/app/application/services/workforce_service.py @@ -74,6 +74,7 @@ from app.infrastructure.repositories.ai_control import ( RoutingPolicyRepository, ) from app.infrastructure.repositories.audit import UsageLedgerRepository +from app.infrastructure.repositories.projects import ProjectRepository from app.infrastructure.repositories.workforce import ( AgentTemplateRepository, ArtifactRepository, @@ -321,6 +322,15 @@ class WorkforceService: # Threads & messages # # ------------------------------------------------------------------ # + async def _own_project_or_404(self, *, organization_id: str, project_id: str | None) -> None: + """A project the caller names must belong to the caller's organization.""" + if not project_id: + return + projects = await ProjectRepository(self.threads.session).list_for_org(organization_id) + if project_id not in {project.id for project in projects}: + # Not found rather than forbidden: another tenant's project is not confirmed to exist. + raise NotFoundError("پروژه یافت نشد.") + async def create_thread( self, *, @@ -329,6 +339,7 @@ class WorkforceService: title: str = "", project_id: str | None = None, ) -> Thread: + await self._own_project_or_404(organization_id=organization_id, project_id=project_id) return await self.threads.create( Thread( organization_id=organization_id, @@ -381,15 +392,26 @@ class WorkforceService: 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. + Ports that take a session (model grants, memory, employee profiles, + pipeline states) 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, its + pipeline state and its usage events belong to one transaction. + + The collaborators match ``container_engine._build_manager``. The runner is + left out on purpose: ``chat`` never runs a turn (only triage, final review + and error review do, and the scheduler's manager covers those). Without + ``states`` a manager's CREATE_TASK makes a task the engine never owns, and + the legacy drain runs it instead. """ from app.application.container_org import build_memory, build_model_grants from app.application.engine.manager import ManagerTurns + from app.infrastructure.repositories.engine import ( + EmployeeProfileRepository, + PipelineStateRepository, + ) + session = self.tasks.session return ManagerTurns( provider=self.provider, employees=self.employees, @@ -400,8 +422,10 @@ class WorkforceService: credentials=self.credentials, rate_cards=self.rate_cards, ledger=self.ledger, - grants=build_model_grants(self.tasks.session), - memory=build_memory(self.tasks.session), + grants=build_model_grants(session), + memory=build_memory(session), + profiles=EmployeeProfileRepository(session), + states=PipelineStateRepository(session), ) # ------------------------------------------------------------------ # @@ -421,6 +445,7 @@ class WorkforceService: ) -> Task: if not title.strip(): raise ValidationError("عنوان کار نمی‌تواند خالی باشد.") + await self._own_project_or_404(organization_id=organization_id, project_id=project_id) if assigned_employee_id: # Validates tenant ownership before we ever store the reference. await self.get_employee_or_404( diff --git a/apps/api/app/infrastructure/repositories/engine.py b/apps/api/app/infrastructure/repositories/engine.py index 8ffa959..2102250 100644 --- a/apps/api/app/infrastructure/repositories/engine.py +++ b/apps/api/app/infrastructure/repositories/engine.py @@ -14,6 +14,8 @@ cannot both win — the loser sees zero updated rows and gets a from __future__ import annotations +from datetime import datetime + from sqlalchemy import func, select, update from sqlalchemy.ext.asyncio import AsyncSession @@ -489,10 +491,11 @@ class GateResultRepository: class TaskSpendRepository: - """Per-task spend, read from the usage ledger (never written here). + """Spend, read from the usage ledger (never written here). - The scheduler's budget guard needs "how much has this task cost so far", - which the ledger can answer by joining its events to the task's runs. The + The scheduler's budget guards need "how much has this task cost so far" + (joining the ledger's events to the task's runs) and "how much has this + organization spent since a date" (the owner's monthly credit budget). The ledger stays the source of truth; this only reads it. """ @@ -505,3 +508,11 @@ class TaskSpendRepository: UsageEventRow.run_id.in_(run_ids) ) return float((await self.session.execute(stmt)).scalar_one()) + + async def spend_for_org_since(self, organization_id: str, since: datetime) -> float: + """What the organization's usage has cost since ``since``, over all its tasks.""" + stmt = select(func.coalesce(func.sum(UsageEventRow.total_price), 0.0)).where( + UsageEventRow.organization_id == organization_id, + UsageEventRow.created_at >= since, + ) + return float((await self.session.execute(stmt)).scalar_one()) diff --git a/apps/api/app/infrastructure/repositories/workforce.py b/apps/api/app/infrastructure/repositories/workforce.py index fcb4cd5..5d053bc 100644 --- a/apps/api/app/infrastructure/repositories/workforce.py +++ b/apps/api/app/infrastructure/repositories/workforce.py @@ -7,7 +7,7 @@ tool calls -> artifacts. from __future__ import annotations -from sqlalchemy import func, select +from sqlalchemy import exists, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.domain.workforce.models import ( @@ -42,6 +42,7 @@ from app.infrastructure.db.models import ( ThreadRow, ToolCallRow, ) +from app.infrastructure.db.models_engine import TaskPipelineStateRow # --------------------------------------------------------------------------- # Mappers @@ -200,8 +201,10 @@ class EmployeeRepository: return [_employee(r) for r in (await self.session.execute(stmt)).scalars()] async def count_for_org(self, organization_id: str) -> int: + """The headcount the cap counts: everyone on the team except the fired (archived).""" stmt = select(func.count()).select_from(EmployeeRow).where( - EmployeeRow.organization_id == organization_id + EmployeeRow.organization_id == organization_id, + EmployeeRow.status != str(EmployeeStatus.ARCHIVED), ) return int((await self.session.execute(stmt)).scalar_one()) @@ -447,10 +450,18 @@ class TaskRepository: Cross-tenant by design: the worker is a platform component, not a tenant. Only ``new``/``triaged`` qualify — anything else is either already executing or closed, and re-running it would double-bill. + + A task that owns a ``task_pipeline_state`` row belongs to + ``Scheduler.tick()``. Running it here would double-bill it and mark it + ``done`` behind the engine's back, so those tasks are excluded. """ + has_pipeline_state = exists( + select(TaskPipelineStateRow.task_id).where(TaskPipelineStateRow.task_id == TaskRow.id) + ) stmt = ( select(TaskRow) .where(TaskRow.status.in_([str(TaskStatus.NEW), str(TaskStatus.TRIAGED)])) + .where(~has_pipeline_state) .order_by(TaskRow.created_at.asc()) .limit(min(limit, 100)) ) diff --git a/apps/api/app/interfaces/mcp/gateway.py b/apps/api/app/interfaces/mcp/gateway.py index 639c6d6..9c486f8 100644 --- a/apps/api/app/interfaces/mcp/gateway.py +++ b/apps/api/app/interfaces/mcp/gateway.py @@ -152,7 +152,7 @@ TOOL_SPECS: tuple[ToolSpec, ...] = ( ToolSpec( name="pipeline_status", scope="delivery:read", - description="وضعیت خط تحویل (استقرار و دمو). در این نسخه در دسترس نیست.", + description="وضعیت خط تحویل (استقرار و دمو).", parameters={ "type": "object", "properties": {}, @@ -562,8 +562,52 @@ async def _knowledge_search( async def _pipeline_status( gateway: McpGateway, *, session, identity: TokenIdentity, **_ignored: Any ) -> dict[str, Any]: - """Delivery is another workstream's table; the tool exists and says so.""" - return {"available": False, "message": "خط تحویل در این نسخه در دسترس نیست."} + """The organization's delivery state: demo runtime, last runs, last production deploy. + + Rule 7: the answer says where things stand and nothing about the infrastructure + under it. No port, container id, deploy-target host or log tail is ever returned. + """ + from app.infrastructure.repositories.app_profile import AppProfileRepository + from app.infrastructure.repositories.delivery import ( + AppRuntimeRepository, + PipelineRunRepository, + ProductionDeployRepository, + ) + + org_id = identity.organization_id + profile = await AppProfileRepository(session).get_for_org(org_id) + if profile is None: + return {"available": False, "message": "اپلیکیشنی برای این سازمان یافت نشد."} + project_id = profile.project_id + runtime = await AppRuntimeRepository(session).get_for_org(org_id, project_id) + runs = await PipelineRunRepository(session).list_for_project( + organization_id=org_id, project_id=project_id, limit=5 + ) + deploys = await ProductionDeployRepository(session).list_for_project( + project_id=project_id, limit=1 + ) + return { + "available": True, + "demo_url": runtime.demo_url if runtime else "", + "runtime_status": runtime.status if runtime else None, + "runs": [ + { + "id": run.id, + "status": run.status, + "trigger": run.trigger, + "branch": run.branch, + "sha": run.sha[:12], + "error_summary": run.error_summary, + "created_at": run.created_at.isoformat(), + } + for run in runs + ], + "last_production_deploy": ( + {"status": deploys[0].status, "created_at": deploys[0].created_at.isoformat()} + if deploys + else None + ), + } _HANDLERS: dict[str, Callable[..., Awaitable[dict[str, Any]]]] = { diff --git a/apps/api/app/workers/__main__.py b/apps/api/app/workers/__main__.py index 9247cb7..fde4c17 100644 --- a/apps/api/app/workers/__main__.py +++ b/apps/api/app/workers/__main__.py @@ -64,10 +64,21 @@ def _billing_pass(args: argparse.Namespace): async def one_pass() -> WorkerOutcome: end = utcnow() - return await worker.roll_up( + rolled = await worker.roll_up( period_start=end - timedelta(days=max(1, args.period_days)), period_end=end, ) + # The same pass also expires stale memory lessons (see BillingWorker.expire_memories). + expired = await worker.expire_memories() + merged = WorkerOutcome( + name=rolled.name, + processed=rolled.processed + expired.processed, + skipped=rolled.skipped + expired.skipped, + failed=rolled.failed + expired.failed, + ) + for message in [*rolled.notes, *expired.notes]: + merged.note(message) + return merged return one_pass diff --git a/apps/api/app/workers/ai_runtime.py b/apps/api/app/workers/ai_runtime.py index 173fa6e..5cb0ac1 100644 --- a/apps/api/app/workers/ai_runtime.py +++ b/apps/api/app/workers/ai_runtime.py @@ -52,7 +52,7 @@ class AiRuntimeWorker: async with worker_session() as session: await self._drain_runs(session, outcome, limit=limit) - await self._tick_engine(session, outcome, limit=limit) + await self._tick_engine(session, outcome) return outcome @@ -116,7 +116,7 @@ class AiRuntimeWorker: message=f"کار «{current.title}» در حال اجراست (اجرا {run.id}).", ) - async def _tick_engine(self, session, outcome: WorkerOutcome, *, limit: int) -> None: + async def _tick_engine(self, session, outcome: WorkerOutcome) -> None: """One scheduler tick: move every ready task one step. A tick failure is recorded against the worker and the pass ends — the @@ -125,7 +125,10 @@ class AiRuntimeWorker: """ try: scheduler = build_scheduler(session) - report = await scheduler.tick(limit=max(limit, 10)) + # A wide read: states that cannot run this pass (automation off, or nobody free + # for their stage) must not fill the window and starve the other organizations. + # Scheduler._select still bounds the real work per organization and per pass. + report = await scheduler.tick(limit=500) except Exception as exc: # noqa: BLE001 - a worker must survive a bad tick outcome.failed += 1 outcome.note(f"engine tick: {type(exc).__name__}: {exc}") diff --git a/apps/api/app/workers/billing.py b/apps/api/app/workers/billing.py index 22895cc..3531629 100644 --- a/apps/api/app/workers/billing.py +++ b/apps/api/app/workers/billing.py @@ -9,6 +9,10 @@ background job's. The one hard requirement is **idempotency**. A cron job that runs twice for the same period must not bill twice, so a period that already has an invoice is skipped rather than re-created. + +The same pass also expires memory lessons past their date (``expire_memories``). +That is housekeeping, not billing, but this is the one scheduled job every +deployment runs, and expiry needs no model provider. """ from __future__ import annotations @@ -20,6 +24,7 @@ from app.domain.billing.models import Invoice, InvoiceStatus from app.infrastructure.repositories.audit import UsageLedgerRepository from app.infrastructure.repositories.billing import InvoiceRepository from app.infrastructure.repositories.identity import OrganizationRepository +from app.infrastructure.repositories.memory import MemoryRepository from app.workers.base import WorkerOutcome, worker_session #: A period shorter than this is almost certainly a bug (a stray manual run), @@ -99,6 +104,20 @@ class BillingWorker: return outcome + async def expire_memories(self) -> WorkerOutcome: + """Soft-delete organization lessons whose expiry has passed. + + Pinned entries never expire, and rules and preferences carry no expiry at + all, so only disposable lessons are touched. The sweep covers every + organization at once, which is what a platform-wide schedule wants. + """ + outcome = WorkerOutcome(name=self.name) + async with worker_session() as session: + expired = await MemoryRepository(session).expire(None) + outcome.processed = expired + outcome.note(f"memory: {expired} expired lesson(s) soft-deleted") + return outcome + @staticmethod def _naive_utc(value: datetime) -> datetime: """Normalise a timestamp to naive UTC for comparison. diff --git a/apps/api/tests/engine_fakes.py b/apps/api/tests/engine_fakes.py index c917a92..064631e 100644 --- a/apps/api/tests/engine_fakes.py +++ b/apps/api/tests/engine_fakes.py @@ -15,12 +15,13 @@ from __future__ import annotations from dataclasses import dataclass from app.application.ports import ChatMessage, CommandResult, ModelHealth, ModelResponse -from app.application.ports.model_grants import GrantedModel +from app.application.ports.model_grants import DefaultOrgPolicy, GrantedModel, OrgPolicyView from app.application.ports.sandbox_provider import ( SessionInfo, SnapshotInfo, WorkspaceInfo, ) +from app.core.config import settings @dataclass(slots=True, kw_only=True) @@ -187,9 +188,16 @@ class FakeGrants: ``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. + + ``policy`` is the owner's per-organization policy. ``None`` means the default + policy, built at call time from ``settings.engine_max_employees``. """ - def __init__(self, granted: list[str] | dict[str, str] | None = None) -> None: + def __init__( + self, + granted: list[str] | dict[str, str] | None = None, + policy: OrgPolicyView | None = None, + ) -> None: if granted is None: self._tiers: dict[str, str] = { "model-strong": "strong", @@ -200,6 +208,7 @@ class FakeGrants: else: self._tiers = dict.fromkeys(granted, "balanced") self.checks: list[str] = [] + self.policy = policy @property def granted(self) -> list[str]: @@ -215,6 +224,12 @@ class FakeGrants: self.checks.append(ai_model_id) return ai_model_id in self._tiers + async def get_org_policy(self, organization_id: str) -> OrgPolicyView: + if self.policy is not None: + return self.policy + # Read at call time: a test may monkeypatch ``settings.engine_max_employees``. + return DefaultOrgPolicy(max_employees=settings.engine_max_employees) + def grant(self, model_id: str, tier: str = "balanced") -> None: self._tiers[model_id] = tier diff --git a/apps/api/tests/test_engine_flow.py b/apps/api/tests/test_engine_flow.py index 4b822e2..78e8871 100644 --- a/apps/api/tests/test_engine_flow.py +++ b/apps/api/tests/test_engine_flow.py @@ -14,7 +14,7 @@ git (in-memory ``FakeRepoWorkspace``) are fakes. What it defends: * red tests and a forbidden diff send the work *back* with cycle + 1 instead of merging it * ``analysis_only`` never creates a worktree -* an empty wallet pauses the task before a single turn is spent +* an overdrawn wallet pauses the task before a single turn is spent * a transient provider failure is retried, then handed to the manager """ @@ -85,23 +85,29 @@ class _World: strong_id: str -class _EmptyWallet: - """A wallet with no credit, and no way to top it up in a test.""" +class _OverdrawnWallet: + """A wallet below zero: the org is overdrawn, so the engine stops before a turn.""" + + balance = -1.0 + + +class _ZeroWallet: + """A wallet at exactly zero: the normal starting state of a fresh, post-paid org.""" balance = 0.0 class _Wallets: - def __init__(self, wallet: _EmptyWallet | None) -> None: + def __init__(self, wallet: _OverdrawnWallet | _ZeroWallet | None) -> None: self._wallet = wallet async def get_for_org(self, organization_id: str): return self._wallet -class _EmptyWalletRepo(_Wallets): +class _OverdrawnWalletRepo(_Wallets): def __init__(self) -> None: - super().__init__(_EmptyWallet()) + super().__init__(_OverdrawnWallet()) @pytest_asyncio.fixture @@ -532,7 +538,7 @@ async def test_analysis_only_finishes_without_a_worktree(world, session_factory) assert _calls(workspace, "commit") == [] -async def test_an_empty_wallet_pauses_the_task_before_a_turn_is_spent( +async def test_an_overdrawn_wallet_pauses_the_task_before_a_turn_is_spent( world, session_factory ): scripted = ScriptedModel([ScriptedReply(content="[[ASSIGN/]]")]) @@ -549,7 +555,7 @@ async def test_an_empty_wallet_pauses_the_task_before_a_turn_is_spent( workspace=workspace, memory=RecordingMemory(), delivery=NullDeliveryHooks(), - wallets=_EmptyWalletRepo(), + wallets=_OverdrawnWalletRepo(), ) await _run(scheduler, session, 1) state = await PipelineStateRepository(session).get(row.id) @@ -561,6 +567,38 @@ async def test_an_empty_wallet_pauses_the_task_before_a_turn_is_spent( assert BUDGET_PAUSED_MESSAGE +async def test_a_zero_wallet_does_not_pause_a_fresh_org(world, session_factory): + """A 0.0 balance is every org's starting state, so it must not pause the task. + + This is an intentional behaviour change. The scheduler used to pause on + ``balance <= 0``, which froze every new org before its first turn: usage is + post-paid and nothing debits a wallet. Only an overdrawn wallet (below zero) + pauses now. The old test that encoded the previous rule was renamed and + updated, as PLAN-v2 §7.4 requires for a deliberate behaviour change. + """ + scripted = ScriptedModel([ScriptedReply(content="[[ASSIGN/]]")]) + workspace = FakeRepoWorkspace() + async with session_factory() as session: + await _team(session, world) + row = await _task(session, world, stage=Stage.PENDING_MANAGER_REVIEW) + await session.commit() + scheduler = _scheduler( + session, + world, + scripted=scripted, + sandbox=FakeSandbox(), + workspace=workspace, + memory=RecordingMemory(), + delivery=NullDeliveryHooks(), + wallets=_Wallets(_ZeroWallet()), + ) + await _run(scheduler, session, 1) + state = await PipelineStateRepository(session).get(row.id) + + assert state is not None and state.stage is not Stage.PAUSED + assert scripted.calls == 1 + + async def test_a_task_waits_for_its_dependencies(world, session_factory): scripted = ScriptedModel( [ diff --git a/apps/api/tests/test_integration_hire_cap.py b/apps/api/tests/test_integration_hire_cap.py new file mode 100644 index 0000000..8710597 --- /dev/null +++ b/apps/api/tests/test_integration_hire_cap.py @@ -0,0 +1,70 @@ +"""Integration: hiring respects the owner's per-organization employee cap. + +The team is capped by two numbers: the owner's ``max_employees`` in the org +policy, and the global ``settings.engine_max_employees``. The lower one wins. The +owner can tighten the cap for their organization, but cannot lift it above the +global limit. +""" + +import pytest + +from app.application.ports.model_grants import DefaultOrgPolicy +from app.core.config import settings +from app.core.errors import ConflictError +from app.domain.workforce.models import RoleKey +from app.infrastructure.repositories.workforce import EmployeeRepository +from tests.engine_fakes import FakeGrants, ScriptedModel +from tests.test_engine_manager import _manager +from tests.test_engine_manager import world as manager_world + +#: The manager tests' organization fixture. Binding it under its own name registers it +#: for this module's tests (a bare import would be flagged as unused and shadowed). +world = manager_world + + +async def _hire_tester(manager, world): + return await manager.hire( + organization_id=world.organization_id, role=RoleKey.TESTER, tier="fast" + ) + + +async def test_the_owner_policy_caps_the_team(world, session_factory): + """An owner's ``max_employees`` of 2 refuses the third hire.""" + async with session_factory() as session: + grants = FakeGrants({world.fast_id: "fast"}, policy=DefaultOrgPolicy(max_employees=2)) + manager = await _manager(world, session, scripted=ScriptedModel(), grants=grants) + await _hire_tester(manager, world) + await _hire_tester(manager, world) + with pytest.raises(ConflictError, match=r"\(2\)"): + await _hire_tester(manager, world) + + +async def test_the_global_cap_still_wins_when_it_is_lower(world, session_factory, monkeypatch): + """The owner can tighten the cap, but a policy of 5 does not lift a global cap of 1.""" + monkeypatch.setattr(settings, "engine_max_employees", 1) + async with session_factory() as session: + grants = FakeGrants({world.fast_id: "fast"}, policy=DefaultOrgPolicy(max_employees=5)) + manager = await _manager(world, session, scripted=ScriptedModel(), grants=grants) + await _hire_tester(manager, world) + with pytest.raises(ConflictError, match=r"\(1\)"): + await _hire_tester(manager, world) + + +async def test_a_fired_employee_frees_a_hire_slot(world, session_factory): + """At the cap, firing one employee archives them, and the next hire succeeds.""" + async with session_factory() as session: + grants = FakeGrants({world.fast_id: "fast"}, policy=DefaultOrgPolicy(max_employees=2)) + manager = await _manager(world, session, scripted=ScriptedModel(), grants=grants) + first = await _hire_tester(manager, world) + await _hire_tester(manager, world) + with pytest.raises(ConflictError, match=r"\(2\)"): + await _hire_tester(manager, world) + await manager.fire(organization_id=world.organization_id, employee_id=first.id) + await session.commit() + + replacement = await _hire_tester(manager, world) + await session.commit() + headcount = await EmployeeRepository(session).count_for_org(world.organization_id) + + assert replacement.role_key is RoleKey.TESTER + assert headcount == 2 diff --git a/apps/api/tests/test_integration_order_route.py b/apps/api/tests/test_integration_order_route.py new file mode 100644 index 0000000..6b37b39 --- /dev/null +++ b/apps/api/tests/test_integration_order_route.py @@ -0,0 +1,107 @@ +"""A manager order sent over the HTTP message route becomes a task the engine owns. + +``POST /api/organizations/{org}/threads/{thread}/messages`` is the route the web UI posts +to. ``WorkforceService._manager_turns`` builds the manager behind it, and that manager must +carry the same ``states`` and ``profiles`` as the engine's own builder +(``container_engine._build_manager``). Without them the manager's ``CREATE_TASK`` makes a +task with no ``task_pipeline_state`` row, and ``TaskRepository.list_runnable`` hands that task +to the legacy one-shot drain, which runs it outside the engine. + +The test drives the route the way a member would: the platform owner grants the default +models, a manager signs up and signs in through ``POST /api/auth/login``, completes the +wizard, opens a thread without ``project_id``, and posts the order. Only the model is +scripted. +""" + +from __future__ import annotations + +from sqlalchemy import select + +from app.domain.workforce.models import TaskStatus +from app.domain.workforce.pipeline import Stage +from app.infrastructure.db.models import AiModelRow +from app.infrastructure.repositories.engine import PipelineStateRepository +from app.infrastructure.repositories.workforce import TaskRepository +from tests.conftest import auth_header, signup +from tests.engine_fakes import ScriptedModel, ScriptedReply +from tests.test_org_wizard import _complete + +DEFAULTS = "/api/admin/default-model-grants" +EMAIL = "order-owner@example.com" +PASSWORD = "password123" +TITLE = "صفحهٔ ورود" + +#: The manager's reply to the order: one task tag, so the order becomes one task. +ORDER_REPLY = ( + "باشه، یک کار می‌سازم.\n" + f'[[CREATE_TASK title="{TITLE}" kind="feature"]]' + "ساخت صفحهٔ ورود با اعتبارسنجی[[/CREATE_TASK]]" +) + + +async def test_a_manager_order_over_http_gets_a_pipeline_state( + client, session_factory, seeded, platform_owner_token, monkeypatch, redirect_error_log +): + scripted = ScriptedModel([ScriptedReply(content=ORDER_REPLY)]) + monkeypatch.setattr("app.application.container.get_model_provider", lambda: scripted) + + # The platform owner grants one reasoning model and one cheap model to the default set. + async with session_factory() as session: + models = list( + (await session.execute(select(AiModelRow).order_by(AiModelRow.internal_name))) + .scalars() + .all() + ) + strong = next(m for m in models if "reasoning" in m.capabilities) + cheap = next(m for m in models if "reasoning" not in m.capabilities) + owner = auth_header(platform_owner_token) + for model in (strong, cheap): + granted = await client.put( + f"{DEFAULTS}/{model.id}", json={"is_active": True}, headers=owner + ) + assert granted.status_code == 200, granted.text + + # A manager signs up, signs in over HTTP, and completes the wizard as the org's owner. + await signup(client, EMAIL, password=PASSWORD) + login = await client.post("/api/auth/login", json={"email": EMAIL, "password": PASSWORD}) + assert login.status_code == 200, login.text + headers = auth_header(login.json()["access_token"]) + body = await _complete(client, headers) + org_id = body["organization"]["id"] + project_id = body["project"]["id"] + + # The web UI opens the thread without a project_id. + created = await client.post( + f"/api/organizations/{org_id}/threads", json={"title": "گفتگو"}, headers=headers + ) + assert created.status_code == 201, created.text + assert created.json()["project_id"] is None + thread_id = created.json()["id"] + + # The order goes through the real message route. + posted = await client.post( + f"/api/organizations/{org_id}/threads/{thread_id}/messages", + json={"content": "یک صفحهٔ ورود بساز."}, + headers=headers, + ) + assert posted.status_code == 201, posted.text + assert f"کار ساخته شد: {TITLE}" in posted.json()["assistant_message"]["content"] + + async with session_factory() as session: + tasks = await TaskRepository(session).list_for_org(org_id) + assert len(tasks) == 1 + task = tasks[0] + state = await PipelineStateRepository(session).get(task.id) + runnable = await TaskRepository(session).list_runnable(limit=100) + + assert task.title == TITLE + assert task.project_id == project_id + # The engine owns the task: it has a pipeline state at the manager's review stage. + assert state is not None + assert state.organization_id == org_id + assert state.project_id == project_id + assert state.stage is Stage.PENDING_MANAGER_REVIEW + # The legacy drain picks up new and triaged tasks. This task is new, so only its + # pipeline state keeps the drain away from it. + assert task.status is TaskStatus.NEW + assert task.id not in {t.id for t in runnable} diff --git a/apps/api/tests/test_integration_policy.py b/apps/api/tests/test_integration_policy.py new file mode 100644 index 0000000..0fc99d1 --- /dev/null +++ b/apps/api/tests/test_integration_policy.py @@ -0,0 +1,138 @@ +"""Integration: the owner's org policy is honoured by the real scheduler tick. + +The tick reads the policy through the manager's grants port, once per +organization, and checks two of its switches before a turn is spent: + +* ``automation_enabled`` off stops every turn and every merge. The stage does not + move, so switching automation back on resumes the task where it was (the + owner's human-off switch). +* ``monthly_credit_budget`` is a ceiling on the organization's spend since the + first day of the UTC month. Spend at or above it pauses the task. + +With no policy (the default), the tick behaves as the flow tests pin it. + +The fixtures and the scheduler wiring come from ``tests.test_engine_flow``; this +module adds the owner's policy and the spend guard on top of them. +""" + +from __future__ import annotations + +from app.application.engine.scheduler import Scheduler +from app.application.ports.delivery_hooks import NullDeliveryHooks +from app.application.ports.model_grants import DefaultOrgPolicy, OrgPolicyView +from app.application.ports.repo_workspace import FakeRepoWorkspace +from app.domain.workforce.pipeline import Stage +from app.infrastructure.db.models import UsageEventRow +from app.infrastructure.repositories.engine import PipelineStateRepository, TaskSpendRepository +from tests.engine_fakes import ( + FakeGrants, + FakeSandbox, + RecordingMemory, + ScriptedModel, + ScriptedReply, +) +from tests.test_engine_flow import _run, _scheduler, _task, _team +from tests.test_engine_flow import world as flow_world + +#: The flow tests' organization fixture. Binding it under its own name registers it +#: for this module's tests (a bare import would be flagged as unused and shadowed). +world = flow_world + + +async def _wired( + session, world, *, scripted: ScriptedModel, policy: OrgPolicyView | None +) -> Scheduler: + """The flow test's scheduler, with the owner's policy and the spend guard wired in.""" + scheduler = _scheduler( + session, + world, + scripted=scripted, + sandbox=FakeSandbox(), + workspace=FakeRepoWorkspace(), + memory=RecordingMemory(), + delivery=NullDeliveryHooks(), + ) + # The tick reads the owner's policy through the manager's grants port. + scheduler.manager.grants = FakeGrants({world.strong_id: "strong"}, policy=policy) + scheduler.spend = TaskSpendRepository(session) + return scheduler + + +def _usage(world, total_price: float) -> UsageEventRow: + """A ledger row for this organization, stamped now (so it falls in this month).""" + return UsageEventRow( + organization_id=world.organization_id, + unit="model_input_tokens", + quantity=1000.0, + unit_price=0.0000004, + total_price=total_price, + ) + + +async def test_automation_off_stops_the_turn_and_moves_no_stage(world, session_factory): + """The owner's human-off switch: no turn is spent and the task keeps its stage.""" + scripted = ScriptedModel([ScriptedReply(content="[[ASSIGN/]]")]) + async with session_factory() as session: + await _team(session, world) + row = await _task(session, world, stage=Stage.PENDING_MANAGER_REVIEW) + await session.commit() + scheduler = await _wired( + session, world, scripted=scripted, policy=DefaultOrgPolicy(automation_enabled=False) + ) + report = await scheduler.tick() + state = await PipelineStateRepository(session).get(row.id) + + assert state is not None and state.stage is Stage.PENDING_MANAGER_REVIEW + assert report.skipped == 1 + assert scripted.calls == 0 + + +async def test_a_spent_monthly_budget_pauses_before_a_turn_is_spent(world, session_factory): + """A budget of 0.0 with any spend this month is used up, so the task pauses.""" + scripted = ScriptedModel([ScriptedReply(content="[[ASSIGN/]]")]) + async with session_factory() as session: + await _team(session, world) + row = await _task(session, world, stage=Stage.PENDING_MANAGER_REVIEW) + session.add(_usage(world, total_price=0.25)) + await session.commit() + scheduler = await _wired( + session, world, scripted=scripted, policy=DefaultOrgPolicy(monthly_credit_budget=0.0) + ) + await _run(scheduler, session, 1) + state = await PipelineStateRepository(session).get(row.id) + + assert state is not None and state.stage is Stage.PAUSED + assert scripted.calls == 0 + + +async def test_spend_under_the_monthly_budget_does_not_pause(world, session_factory): + """Spend below the monthly budget leaves the task free to take its turn.""" + scripted = ScriptedModel([ScriptedReply(content="[[ASSIGN/]]")]) + async with session_factory() as session: + await _team(session, world) + row = await _task(session, world, stage=Stage.PENDING_MANAGER_REVIEW) + session.add(_usage(world, total_price=0.25)) + await session.commit() + scheduler = await _wired( + session, world, scripted=scripted, policy=DefaultOrgPolicy(monthly_credit_budget=10.0) + ) + await _run(scheduler, session, 1) + state = await PipelineStateRepository(session).get(row.id) + + assert state is not None and state.stage is not Stage.PAUSED + assert scripted.calls == 1 + + +async def test_no_policy_keeps_the_default_behaviour(world, session_factory): + """With no owner policy the default applies: automation on, no monthly budget.""" + scripted = ScriptedModel([ScriptedReply(content="[[ASSIGN/]]")]) + async with session_factory() as session: + await _team(session, world) + row = await _task(session, world, stage=Stage.PENDING_MANAGER_REVIEW) + await session.commit() + scheduler = await _wired(session, world, scripted=scripted, policy=None) + await _run(scheduler, session, 1) + state = await PipelineStateRepository(session).get(row.id) + + assert state is not None and state.stage is not Stage.PAUSED + assert scripted.calls == 1 diff --git a/apps/api/tests/test_integration_review_fallback.py b/apps/api/tests/test_integration_review_fallback.py new file mode 100644 index 0000000..49bba36 --- /dev/null +++ b/apps/api/tests/test_integration_review_fallback.py @@ -0,0 +1,233 @@ +"""A review pipeline needs an active reviewer, or the task waits forever in a review stage. + +``resolve_pipeline`` already falls back from ``plan_code_review_test`` to ``plan_code_test`` +when it is told there is no active reviewer. The manager used to leave that flag at its +default, so a refactor in an organization without a reviewer went to a stage nobody could +act on. The manager now looks the reviewer up in the organization, for the chat order and +for triage. The bootstrap team still has no reviewer: a fresh organization gets the plain +pipeline, and only a hired, active reviewer gets the review one. +""" + +from __future__ import annotations + +import pytest_asyncio +from sqlalchemy import select + +from app.application.container_org import build_memory +from app.application.engine.manager import ManagerTurns +from app.application.engine.runner import TurnRunner +from app.core.config import settings +from app.domain.base import new_id +from app.domain.workforce.models import Employee, EmployeeStatus, RoleKey +from app.domain.workforce.pipeline import PipelineKey, Stage, TaskKind, TaskPipelineState +from app.infrastructure.db.models import ( + AiModelRow, + OrganizationRow, + RunRow, + TaskRow, + ThreadRow, + UserRow, +) +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 + +#: One refactor, which the task-kind default sends to the review pipeline. +REFACTOR_ORDER = ( + '[[CREATE_TASK title="بازنویسی ورود" kind="refactor"]]بازنویسی ماژول ورود[[/CREATE_TASK]]' +) + + +class _Org: + def __init__(self, **kwargs) -> None: + self.__dict__.update(kwargs) + + +@pytest_asyncio.fixture +async def org(session_factory, seeded): + """An organization and its owner, with one reasoning and one cheap catalog model.""" + async with session_factory() as session: + user = UserRow(email=f"rev-{new_id()[:8]}@example.com", password_hash="x") + session.add(user) + await session.flush() + organization = OrganizationRow( + name="سازمان بازبین", slug=f"rev-{user.id[:6]}", owner_user_id=user.id + ) + session.add(organization) + await session.commit() + models = list( + (await session.execute(select(AiModelRow).order_by(AiModelRow.internal_name))) + .scalars() + .all() + ) + strong = next(m for m in models if "reasoning" in m.capabilities) + fast = next(m for m in models if "reasoning" not in m.capabilities) + yield _Org( + id=organization.id, + user_id=user.id, + strong_id=strong.id, + fast_id=fast.id, + ) + + +def _grants(org: _Org) -> FakeGrants: + return FakeGrants({org.strong_id: "strong", org.fast_id: "fast"}) + + +def _manager(session, *, scripted, grants) -> ManagerTurns: + """A manager service with real repositories and a scripted model.""" + 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 _order_pipeline( + org: _Org, session_factory, *, reviewer_status: EmployeeStatus | None +) -> PipelineKey: + """Bootstrap the team, optionally hire a reviewer, then send one refactor order.""" + scripted = ScriptedModel([ScriptedReply(content=REFACTOR_ORDER)]) + async with session_factory() as session: + manager = _manager(session, scripted=scripted, grants=_grants(org)) + await manager.bootstrap_team(organization_id=org.id, project_id=None) + if reviewer_status is not None: + await EmployeeRepository(session).create( + Employee( + organization_id=org.id, + name="رضا", + role_key=RoleKey.REVIEWER, + status=reviewer_status, + ) + ) + thread = await ThreadRepository(session).create( + ThreadRow(organization_id=org.id, title="گفتگو") + ) + await session.commit() + await manager.chat( + organization_id=org.id, + thread_id=thread.id, + content="این ماژول را بازنویسی کن.", + user_id=org.user_id, + ) + await session.commit() + task = (await TaskRepository(session).list_for_org(org.id))[0] + state = await PipelineStateRepository(session).get(task.id) + assert state is not None + assert state.kind is TaskKind.REFACTOR + return state.pipeline + + +async def test_a_refactor_order_without_a_reviewer_skips_the_review_stage(org, session_factory): + pipeline = await _order_pipeline(org, session_factory, reviewer_status=None) + + assert pipeline is PipelineKey.PLAN_CODE_TEST + + +async def test_a_refactor_order_with_an_active_reviewer_gets_the_review_stage( + org, session_factory +): + pipeline = await _order_pipeline( + org, session_factory, reviewer_status=EmployeeStatus.ACTIVE + ) + + assert pipeline is PipelineKey.PLAN_CODE_REVIEW_TEST + + +async def test_a_paused_reviewer_does_not_count(org, session_factory): + pipeline = await _order_pipeline( + org, session_factory, reviewer_status=EmployeeStatus.PAUSED + ) + + assert pipeline is PipelineKey.PLAN_CODE_TEST + + +async def _refactor_task(session, organization_id: str) -> tuple[str, str]: + task = TaskRow(organization_id=organization_id, title="بازنویسی") + session.add(task) + await session.flush() + run = RunRow(task_id=task.id) + session.add(run) + await session.flush() + # The state starts on the review pipeline, as a refactor order would leave it, so the + # assertion below can only pass if triage moved it back to the plain pipeline. + await PipelineStateRepository(session).create( + TaskPipelineState( + task_id=task.id, + organization_id=organization_id, + pipeline=PipelineKey.PLAN_CODE_REVIEW_TEST, + kind=TaskKind.REFACTOR, + stage=Stage.PENDING_MANAGER_REVIEW, + ) + ) + await session.flush() + return task.id, run.id + + +async def test_triage_of_a_refactor_without_a_reviewer_skips_the_review_stage( + org, session_factory, monkeypatch +): + monkeypatch.setattr(settings, "engine_max_tool_calls_per_turn", 40) + scripted = ScriptedModel([ScriptedReply(content='[[ASSIGN employee=""/]]')]) + grants = _grants(org) + async with session_factory() as session: + manager = _manager(session, scripted=scripted, grants=grants) + await manager.bootstrap_team(organization_id=org.id, project_id=None) + manager_row = next( + e + for e in await EmployeeRepository(session).list_for_org(org.id) + if e.role_key is RoleKey.MANAGER + ) + task_id, run_id = await _refactor_task(session, org.id) + await session.commit() + manager.runner = TurnRunner( + provider=scripted, + models=AiModelRepository(session), + credentials=ModelCredentialRepository(session), + rate_cards=RateCardRepository(session), + ledger=UsageLedgerRepository(session), + grants=grants, + memory=build_memory(session), + turns=None, + run_steps=RunStepRepository(session), + tool_calls=ToolCallRepository(session), + runs=RunRepository(session), + states=PipelineStateRepository(session), + ) + 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 True + assert outcome.pipeline is PipelineKey.PLAN_CODE_TEST + assert state is not None + assert state.pipeline is PipelineKey.PLAN_CODE_TEST diff --git a/apps/api/tests/test_integration_runtime.py b/apps/api/tests/test_integration_runtime.py new file mode 100644 index 0000000..348711f --- /dev/null +++ b/apps/api/tests/test_integration_runtime.py @@ -0,0 +1,134 @@ +"""Integration: the legacy one-shot run drain never picks up an engine task. + +Engine tasks are created with the coarse status ``new``, exactly like a plain +task. ``AiRuntimeWorker._drain_runs`` reads ``TaskRepository.list_runnable`` and +hands every row to the legacy ``start_run``, which marks the task ``done``. If +that query returned an engine task, a manager order would run once through the +drain and never through ``Scheduler.tick()``, and the task would be billed on the +wrong path. A task that owns a ``task_pipeline_state`` row belongs to the engine, +so the drain has to skip it while still picking up the plain task. +""" + +from __future__ import annotations + +from app.application.ports.delivery_hooks import NullDeliveryHooks +from app.application.ports.repo_workspace import FakeRepoWorkspace +from app.domain.base import new_id +from app.domain.workforce.pipeline import PipelineKey, Stage, TaskKind, TaskPipelineState +from app.infrastructure.db.models import OrganizationRow, TaskRow, UserRow +from app.infrastructure.repositories.engine import PipelineStateRepository +from app.infrastructure.repositories.workforce import TaskRepository +from app.workers.ai_runtime import AiRuntimeWorker +from tests.engine_fakes import FakeSandbox, RecordingMemory, ScriptedModel, ScriptedReply +from tests.test_engine_flow import _scheduler, _task, _team +from tests.test_engine_flow import world as flow_world + +#: The flow tests' organization fixture. Binding it under its own name registers it +#: for this module's tests (a bare import would be flagged as unused and shadowed). +world = flow_world + + +async def test_the_drain_skips_a_task_the_engine_owns(session_factory, seeded) -> None: + async with session_factory() as session: + user = UserRow(email=f"drain-{new_id()[:8]}@example.com", password_hash="x") + session.add(user) + await session.flush() + org = OrganizationRow( + name="سازمان صف اجرا", slug=f"drain-{user.id[:6]}", owner_user_id=user.id + ) + session.add(org) + await session.flush() + plain = TaskRow( + organization_id=org.id, title="کار ساده", created_by_user_id=user.id + ) + engine_owned = TaskRow( + organization_id=org.id, title="کار موتور", created_by_user_id=user.id + ) + session.add_all([plain, engine_owned]) + await session.flush() + await PipelineStateRepository(session).create( + TaskPipelineState( + task_id=engine_owned.id, + organization_id=org.id, + pipeline=PipelineKey.PLAN_CODE_TEST, + kind=TaskKind.FEATURE, + stage=Stage.PENDING_MANAGER_REVIEW, + ) + ) + await session.commit() + + runnable = await TaskRepository(session).list_runnable() + + assert [task.id for task in runnable] == [plain.id] + + +async def _stuck_org(session) -> tuple[str, str]: + """An organization with no employees, so none of its tasks can be staffed.""" + user = UserRow(email=f"stuck-{new_id()[:8]}@example.com", password_hash="x") + session.add(user) + await session.flush() + org = OrganizationRow( + name="سازمان بدون کارمند", slug=f"stuck-{user.id[:6]}", owner_user_id=user.id + ) + session.add(org) + await session.flush() + return org.id, user.id + + +async def _stuck_task(session, organization_id: str, user_id: str, title: str) -> None: + row = TaskRow(organization_id=organization_id, title=title, created_by_user_id=user_id) + session.add(row) + await session.flush() + await PipelineStateRepository(session).create( + TaskPipelineState( + task_id=row.id, + organization_id=organization_id, + pipeline=PipelineKey.PLAN_CODE_TEST, + kind=TaskKind.FEATURE, + stage=Stage.PENDING_MANAGER_REVIEW, + ) + ) + await session.flush() + + +async def test_twelve_stuck_tasks_do_not_starve_another_org_in_one_pass( + world, session_factory, redirect_error_log, monkeypatch +) -> None: + """The tick reads the oldest states first, so twelve unstaffed tasks come first. + + With a read window of ten they would fill it, and the staffed organization's task + would never be read. The tick reads a wide window, and the per-organization cap in + ``Scheduler._select`` still lets the staffed task take its turn. + """ + async with session_factory() as session: + stuck_org_id, stuck_user_id = await _stuck_org(session) + for number in range(12): + await _stuck_task(session, stuck_org_id, stuck_user_id, f"کار گیر {number}") + await session.commit() + + scripted = ScriptedModel([ScriptedReply(content='[[ASSIGN employee=""/]]')]) + async with session_factory() as session: + await _team(session, world) + staffed = await _task(session, world, stage=Stage.PENDING_MANAGER_REVIEW) + staffed_id = staffed.id + await session.commit() + + monkeypatch.setattr( + "app.workers.ai_runtime.build_scheduler", + lambda session: _scheduler( + session, + world, + scripted=scripted, + sandbox=FakeSandbox(), + workspace=FakeRepoWorkspace(), + memory=RecordingMemory(), + delivery=NullDeliveryHooks(), + ), + ) + await AiRuntimeWorker().drain() + + async with session_factory() as session: + state = await PipelineStateRepository(session).get(staffed_id) + + assert state is not None and state.stage is Stage.PENDING_ANALYSIS + assert scripted.calls == 1 diff --git a/apps/api/tests/test_integration_sandbox.py b/apps/api/tests/test_integration_sandbox.py new file mode 100644 index 0000000..27d903b --- /dev/null +++ b/apps/api/tests/test_integration_sandbox.py @@ -0,0 +1,134 @@ +"""Integration: the engine starts the project's sandbox session when none is active. + +A fresh organization cannot start a sandbox through the API, because sandbox +mutations are owner-only. Before this, the engine only reused a running session, +so every tool turn of a new app failed and the tests gate stayed red. The engine +now starts the project's single workspace session on demand. The tests use the +default mock provider and the seeded sandbox templates. +""" + +from __future__ import annotations + +from sqlalchemy import select + +from app.application.container_engine import _EngineSandboxSessions +from app.core.errors import ProviderUnavailableError +from app.domain.base import new_id +from app.domain.sandbox.models import SessionStatus +from app.infrastructure.db.models import ( + ErrorEventRow, + OrganizationRow, + ProjectRow, + TaskRow, + UserRow, +) +from app.infrastructure.repositories.sandbox import ( + SandboxSessionRepository, + WorkspaceRepository, +) +from app.infrastructure.repositories.workforce import TaskRepository + + +async def _seed_task(session_factory) -> str: + """One organization with one project and one queued task. Returns the task id.""" + async with session_factory() as session: + user = UserRow(email=f"sbx-{new_id()[:8]}@example.com", password_hash="x") + session.add(user) + await session.flush() + org = OrganizationRow( + name="سازمان سندباکس", slug=f"sbx-{user.id[:6]}", owner_user_id=user.id + ) + session.add(org) + await session.flush() + project = ProjectRow( + id=new_id(), organization_id=org.id, name="app", slug="app", test_command="pytest -q" + ) + session.add(project) + await session.flush() + task = TaskRow( + organization_id=org.id, + title="کار سندباکس", + project_id=project.id, + created_by_user_id=user.id, + ) + session.add(task) + await session.commit() + return task.id + + +async def _sandbox_for(session, task_id: str) -> str: + task = await TaskRepository(session).get(task_id) + assert task is not None + return await _EngineSandboxSessions(session).for_task(task) + + +async def test_a_project_without_a_workspace_gets_one_running_session(session_factory, seeded): + task_id = await _seed_task(session_factory) + async with session_factory() as session: + sandbox_id = await _sandbox_for(session, task_id) + await session.commit() + task = await TaskRepository(session).get(task_id) + assert task is not None and task.project_id is not None + workspaces = await WorkspaceRepository(session).list_for_project(task.project_id) + sessions = await SandboxSessionRepository(session).list_for_workspace(workspaces[0].id) + + assert sandbox_id + assert len(workspaces) == 1 + assert [s.status for s in sessions] == [SessionStatus.RUNNING] + assert sessions[0].external_session_id == sandbox_id + + +async def test_a_second_call_reuses_the_running_session(session_factory, seeded): + task_id = await _seed_task(session_factory) + async with session_factory() as session: + first = await _sandbox_for(session, task_id) + second = await _sandbox_for(session, task_id) + await session.commit() + task = await TaskRepository(session).get(task_id) + assert task is not None and task.project_id is not None + workspaces = await WorkspaceRepository(session).list_for_project(task.project_id) + sessions = await SandboxSessionRepository(session).list_for_workspace(workspaces[0].id) + + assert first + assert second == first + assert len(workspaces) == 1 + assert len(sessions) == 1 + + +async def test_without_a_template_nothing_starts_and_the_reason_is_recorded( + session_factory, redirect_error_log +): + """No seeded sandbox templates (this test does not use ``seeded``).""" + task_id = await _seed_task(session_factory) + async with session_factory() as session: + sandbox_id = await _sandbox_for(session, task_id) + await session.commit() + + async with session_factory() as session: + rows = (await session.execute(select(ErrorEventRow))).scalars().all() + + assert sandbox_id == "" + assert any(row.kind == "engine.sandbox_provision_failed" for row in rows) + + +async def test_a_provider_failure_is_recorded_not_raised( + session_factory, seeded, redirect_error_log, monkeypatch +): + class _UnavailableService: + async def create_workspace(self, **_kwargs): + raise ProviderUnavailableError("ساخت محیط اجرای ایزوله ناموفق بود.") + + monkeypatch.setattr( + "app.application.container.build_sandbox_service", + lambda _session: _UnavailableService(), + ) + task_id = await _seed_task(session_factory) + async with session_factory() as session: + sandbox_id = await _sandbox_for(session, task_id) + await session.commit() + + async with session_factory() as session: + rows = (await session.execute(select(ErrorEventRow))).scalars().all() + + assert sandbox_id == "" + assert any(row.kind == "engine.sandbox_provision_failed" for row in rows) diff --git a/apps/api/tests/test_integration_tenant_project.py b/apps/api/tests/test_integration_tenant_project.py new file mode 100644 index 0000000..ff5a6d1 --- /dev/null +++ b/apps/api/tests/test_integration_tenant_project.py @@ -0,0 +1,164 @@ +"""Integration: a project id from another organization never reaches a thread, a task or a sandbox. + +The caller names the project of a thread or a task. The service answers 404 for a project +that is not the caller's organization's, so one tenant cannot put work into another +tenant's project. The engine applies the same rule before it reuses or starts a sandbox, +so a task row that names a foreign project never gets that project's session, and a +workspace of another organization on the same project is ignored. +""" + +from __future__ import annotations + +from app.application.container_engine import _EngineSandboxSessions +from app.domain.base import new_id +from app.domain.sandbox.models import SandboxSession, SessionStatus, Workspace, WorkspaceStatus +from app.infrastructure.db.models import OrganizationRow, ProjectRow, TaskRow, UserRow +from app.infrastructure.repositories.sandbox import ( + SandboxSessionRepository, + SandboxTemplateRepository, + WorkspaceRepository, +) +from app.infrastructure.repositories.workforce import TaskRepository +from tests.conftest import auth_header, signup +from tests.test_org_wizard import _complete + + +async def _org_with_project(session, *, label: str) -> tuple[str, str, str]: + """One organization with its owner and one project. Returns (org, owner, project) ids.""" + user = UserRow(email=f"{label}-{new_id()[:8]}@example.com", password_hash="x") + session.add(user) + await session.flush() + org = OrganizationRow( + name=f"سازمان {label}", slug=f"{label}-{user.id[:6]}", owner_user_id=user.id + ) + session.add(org) + await session.flush() + project = ProjectRow( + id=new_id(), organization_id=org.id, name="app", slug="app", test_command="pytest -q" + ) + session.add(project) + await session.flush() + return org.id, user.id, project.id + + +async def test_a_thread_or_task_cannot_name_another_orgs_project(client, seeded): + headers_a = auth_header((await signup(client, "tenant-a@example.com"))["access_token"]) + headers_b = auth_header((await signup(client, "tenant-b@example.com"))["access_token"]) + body_a = await _complete(client, headers_a, {"org_name": "سازمان الف", "slug": "tenant-alpha"}) + body_b = await _complete(client, headers_b, {"org_name": "سازمان ب", "slug": "tenant-bravo"}) + org_a = body_a["organization"]["id"] + own_project = body_a["project"]["id"] + foreign_project = body_b["project"]["id"] + + thread = await client.post( + f"/api/organizations/{org_a}/threads", + json={"title": "گفتگو", "project_id": foreign_project}, + headers=headers_a, + ) + task = await client.post( + f"/api/organizations/{org_a}/tasks", + json={"title": "کار مشترک", "project_id": foreign_project}, + headers=headers_a, + ) + own = await client.post( + f"/api/organizations/{org_a}/tasks", + json={"title": "کار خودی", "project_id": own_project}, + headers=headers_a, + ) + listed = await client.get(f"/api/organizations/{org_a}/tasks", headers=headers_a) + + assert thread.status_code == 404, thread.text + assert task.status_code == 404, task.text + assert own.status_code == 201, own.text + # The refused task was never stored: the only task of org A is the one in its own project. + assert [row["project_id"] for row in listed.json()] == [own_project] + + +async def test_the_engine_never_runs_a_task_in_another_orgs_project(session_factory, seeded): + """Org B has a running session. A task of org A that names B's project gets none of it.""" + async with session_factory() as session: + org_b, _, project_b = await _org_with_project(session, label="sbx-b") + org_a, user_a, _ = await _org_with_project(session, label="sbx-a") + template = (await SandboxTemplateRepository(session).list())[0] + workspace_b = await WorkspaceRepository(session).create( + Workspace( + organization_id=org_b, + project_id=project_b, + template_id=template.id, + status=WorkspaceStatus.READY, + ) + ) + await SandboxSessionRepository(session).create( + SandboxSession( + workspace_id=workspace_b.id, + external_session_id="sess-b", + status=SessionStatus.RUNNING, + ) + ) + task = TaskRow( + organization_id=org_a, + title="کار غریبه", + project_id=project_b, + created_by_user_id=user_a, + ) + session.add(task) + await session.commit() + task_id = task.id + + async with session_factory() as session: + row = await TaskRepository(session).get(task_id) + assert row is not None + sandbox_id = await _EngineSandboxSessions(session).for_task(row) + await session.commit() + workspaces = await WorkspaceRepository(session).list_for_project(project_b) + + assert sandbox_id == "" + # Nothing was provisioned under org A's name in org B's project. + assert [w.organization_id for w in workspaces] == [org_b] + + +async def test_a_workspace_of_another_org_on_the_same_project_is_ignored(session_factory, seeded): + """A foreign workspace linked to org A's own project is never the one the task uses.""" + async with session_factory() as session: + org_a, user_a, project_a = await _org_with_project(session, label="sbx-own") + org_b, _, _ = await _org_with_project(session, label="sbx-other") + template = (await SandboxTemplateRepository(session).list())[0] + foreign = await WorkspaceRepository(session).create( + Workspace( + organization_id=org_b, + project_id=project_a, + template_id=template.id, + status=WorkspaceStatus.READY, + ) + ) + await SandboxSessionRepository(session).create( + SandboxSession( + workspace_id=foreign.id, + external_session_id="sess-foreign", + status=SessionStatus.RUNNING, + ) + ) + task = TaskRow( + organization_id=org_a, + title="کار خودی", + project_id=project_a, + created_by_user_id=user_a, + ) + session.add(task) + await session.commit() + task_id = task.id + + async with session_factory() as session: + row = await TaskRepository(session).get(task_id) + assert row is not None + sandbox_id = await _EngineSandboxSessions(session).for_task(row) + await session.commit() + own = [ + w + for w in await WorkspaceRepository(session).list_for_project(project_a) + if w.organization_id == org_a + ] + + assert sandbox_id + assert sandbox_id != "sess-foreign" + assert len(own) == 1 diff --git a/apps/api/tests/test_integration_thread_project.py b/apps/api/tests/test_integration_thread_project.py new file mode 100644 index 0000000..3dd0f03 --- /dev/null +++ b/apps/api/tests/test_integration_thread_project.py @@ -0,0 +1,88 @@ +"""A manager order from a thread without a project still gets the organization's project. + +The web UI creates the manager thread without ``project_id`` (``manager/page.tsx``). Before +this fix such an order made a task with no project, so the engine gave it no worktree and +no sandbox. ``ManagerTurns`` now falls back to the organization's project, for the task and +for its pipeline state. This test drives the path: the owner grants the default models, the +manager completes the wizard over HTTP, a thread is created over HTTP without ``project_id``, +and the order goes to the engine's manager, as in ``test_engine_manager.py``. +""" + +from __future__ import annotations + +from sqlalchemy import select + +from app.application.container_engine import build_scheduler +from app.infrastructure.db.models import AiModelRow, OrganizationRow +from app.infrastructure.repositories.engine import PipelineStateRepository +from app.infrastructure.repositories.workforce import TaskRepository +from tests.conftest import auth_header +from tests.engine_fakes import ScriptedModel, ScriptedReply +from tests.test_org_wizard import _complete, _manager + +DEFAULTS = "/api/admin/default-model-grants" + +#: The manager's chat reply: one task tag, so the order becomes one task. +ORDER_REPLY = ( + "باشه، یک کار می‌سازم.\n" + '[[CREATE_TASK title="صفحهٔ ورود" kind="feature"]]' + "ساخت صفحهٔ ورود با اعتبارسنجی[[/CREATE_TASK]]" +) + + +async def test_an_order_in_a_thread_without_a_project_gets_the_orgs_project( + client, session_factory, seeded, platform_owner_token, monkeypatch +): + scripted = ScriptedModel([ScriptedReply(content=ORDER_REPLY)]) + monkeypatch.setattr("app.application.container.get_model_provider", lambda: scripted) + + # The platform owner grants one reasoning model and one cheap model to the default set. + async with session_factory() as session: + models = list( + (await session.execute(select(AiModelRow).order_by(AiModelRow.internal_name))) + .scalars() + .all() + ) + strong = next(m for m in models if "reasoning" in m.capabilities) + cheap = next(m for m in models if "reasoning" not in m.capabilities) + owner = auth_header(platform_owner_token) + for model in (strong, cheap): + granted = await client.put( + f"{DEFAULTS}/{model.id}", json={"is_active": True}, headers=owner + ) + assert granted.status_code == 200, granted.text + + # The manager completes the wizard: org, project, team and grants. + _, headers = await _manager(client) + body = await _complete(client, headers) + org_id = body["organization"]["id"] + project_id = body["project"]["id"] + + # The web UI's way: a thread with no project_id. + created = await client.post( + f"/api/organizations/{org_id}/threads", json={"title": "گفتگو"}, headers=headers + ) + assert created.status_code == 201, created.text + assert created.json()["project_id"] is None + thread_id = created.json()["id"] + + # The order goes to the engine's manager. The HTTP message route builds its manager + # without pipeline states (workforce_service._manager_turns), so it cannot show the state. + async with session_factory() as session: + owner_user_id = (await session.get(OrganizationRow, org_id)).owner_user_id + _, assistant = await build_scheduler(session).manager.chat( + organization_id=org_id, + thread_id=thread_id, + content="یک صفحهٔ ورود بساز.", + user_id=owner_user_id, + ) + await session.commit() + tasks = await TaskRepository(session).list_for_org(org_id) + states = await PipelineStateRepository(session).list_for_org(org_id) + + assert "کار ساخته شد: صفحهٔ ورود" in assistant.content + + assert len(tasks) == 1 + assert tasks[0].project_id == project_id + assert len(states) == 1 + assert states[0].project_id == project_id diff --git a/apps/api/tests/test_mcp.py b/apps/api/tests/test_mcp.py index 6e47818..48ef946 100644 --- a/apps/api/tests/test_mcp.py +++ b/apps/api/tests/test_mcp.py @@ -20,6 +20,8 @@ rules are ours. from __future__ import annotations +import json + import pytest from httpx import AsyncClient from sqlalchemy import select @@ -38,8 +40,14 @@ from app.infrastructure.db.models import ( AuditEventRow, MembershipRow, OrganizationRow, + ProjectRow, UserRow, ) +from app.infrastructure.db.models_delivery import ( + AppRuntimeRow, + PipelineRunRow, + ProductionDeployRow, +) from app.infrastructure.db.models_org import McpTokenRow from app.interfaces.mcp.gateway import McpGateway from tests.conftest import auth_header, signup @@ -273,12 +281,70 @@ async def test_knowledge_and_pipeline_tools(session_factory, seeded): assert empty.ok is True assert empty.payload["results"] == [] - # Delivery is another workstream's table; the tool exists and says so. + # No app yet: the honest answer is "not available", never an empty delivery state. status = await _call(gateway, token, "pipeline_status") assert status.ok is True assert status.payload["available"] is False +async def test_pipeline_status_reports_the_delivery_state_and_no_host_details( + session_factory, seeded +): + org_id, user_id = await _org(session_factory, "pipeline-org") + project_id = new_id() + async with session_factory() as session: + session.add( + ProjectRow( + id=project_id, organization_id=org_id, name="اپ", slug="app", + created_by_user_id=user_id, + ) + ) + await session.flush() + session.add( + AppRuntimeRow( + project_id=project_id, organization_id=org_id, + demo_host="pipeline-org.preview.example", host_port=18080, + demo_container_id="ctr-secret-42", status="live", + ) + ) + session.add( + PipelineRunRow( + id=new_id(), organization_id=org_id, project_id=project_id, + trigger="engine", branch="dev", sha="0123456789abcdef", status="succeeded", + ) + ) + session.add( + ProductionDeployRow( + id=new_id(), organization_id=org_id, project_id=project_id, + kind="deploy", host="deploy-host.internal", status="succeeded", + log_tail="token=super-secret-log-line", + ) + ) + await session.commit() + gateway = await _gateway(session_factory) + token = await _token( + session_factory, gateway, org_id, scopes=["delivery:read"], user_id=user_id + ) + + status = await _call(gateway, token, "pipeline_status") + + assert status.ok is True + payload = status.payload + assert payload["available"] is True + assert payload["demo_url"] == "https://pipeline-org.preview.example" + assert payload["runtime_status"] == "live" + assert [run["status"] for run in payload["runs"]] == ["succeeded"] + assert payload["runs"][0]["sha"] == "0123456789ab" + assert set(payload["last_production_deploy"]) == {"status", "created_at"} + assert payload["last_production_deploy"]["status"] == "succeeded" + # Rule 7: no port, container id, deploy-target host or log tail leaves the gateway. + text = json.dumps(payload, ensure_ascii=False) + for hidden in ( + "host_port", "18080", "ctr-secret-42", "deploy-host.internal", "super-secret-log-line", + ): + assert hidden not in text + + async def test_a_task_created_through_a_token_is_attributed_to_its_minter( session_factory, seeded ): diff --git a/apps/api/tests/test_memory_expiry_worker.py b/apps/api/tests/test_memory_expiry_worker.py new file mode 100644 index 0000000..09f3d79 --- /dev/null +++ b/apps/api/tests/test_memory_expiry_worker.py @@ -0,0 +1,119 @@ +"""The memory-expiry step of the billing pass removes expired lessons, and nothing else. + +The invariant under test: an expired, unpinned lesson is soft-deleted by the worker, +while a pinned lesson whose expiry is also in the past survives. The step needs no +model provider, so the provider used in the setup fails loudly if anything asks it +for a call. +""" + +from __future__ import annotations + +from argparse import Namespace + +import pytest_asyncio +from sqlalchemy import select + +import app.infrastructure.error_log as error_log +from app.application.container_org import build_memory_service +from app.domain.base import new_id, utcnow +from app.domain.identity.models import PlatformRole, Role +from app.infrastructure.db.models import MembershipRow, OrganizationRow, UserRow +from app.infrastructure.db.models_org import OrgMemoryRow +from app.workers import base as worker_base +from app.workers.billing import BillingWorker + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +@pytest_asyncio.fixture(autouse=True) +async def _workers_use_the_test_database(monkeypatch, session_factory): + """Point the worker session and the error recorder at the in-memory database.""" + monkeypatch.setattr(worker_base, "SessionFactory", session_factory) + monkeypatch.setattr(error_log, "SessionFactory", session_factory) + + +class _NoProviderAllowed: + """Recording a lesson never calls a model; the expiry step must not either.""" + + async def complete(self, **_kwargs): + raise AssertionError("memory expiry must not call a model provider") + + +async def _org(session_factory, slug: str) -> str: + """An organization with one owner, written directly to the database.""" + async with session_factory() as session: + user = UserRow( + id=new_id(), + email=f"{slug}@example.com", + password_hash="x", + name=slug, + platform_role=PlatformRole.MANAGER.value, + ) + session.add(user) + await session.flush() + org = OrganizationRow(id=new_id(), name=slug, slug=slug, owner_user_id=user.id) + session.add(org) + await session.flush() + session.add( + MembershipRow(organization_id=org.id, user_id=user.id, role=Role.OWNER.value) + ) + await session.commit() + return org.id + + +# --------------------------------------------------------------------------- +# Expiry +# --------------------------------------------------------------------------- + + +async def test_the_expiry_step_removes_only_expired_unpinned_lessons(session_factory): + org_id = await _org(session_factory, "expiry-worker") + async with session_factory() as session: + service = build_memory_service(session, provider=_NoProviderAllowed()) + expired = await service.record( + organization_id=org_id, kind="lesson", content="درس منقضی", source="reflection" + ) + pinned = await service.record( + organization_id=org_id, kind="lesson", content="درس سنجاق‌شده", source="reflection" + ) + await service.set_pinned(organization_id=org_id, memory_id=pinned, pinned=True) + # Both lessons are past their date; only the unpinned one may be removed. + for memory_id in (expired, pinned): + row = await session.get(OrgMemoryRow, memory_id) + row.expires_at = utcnow().replace(year=2020) + await session.commit() + + outcome = await BillingWorker().expire_memories() + + assert outcome.processed == 1 + assert outcome.failed == 0 + async with session_factory() as session: + rows = {row.id: row for row in (await session.execute(select(OrgMemoryRow))).scalars()} + assert rows[expired].deleted_at is not None + assert rows[pinned].deleted_at is None + + +async def test_the_billing_pass_runs_the_expiry_step_after_the_rollup(session_factory): + from app.workers.__main__ import _billing_pass + + org_id = await _org(session_factory, "expiry-pass") + async with session_factory() as session: + service = build_memory_service(session, provider=_NoProviderAllowed()) + memory_id = await service.record( + organization_id=org_id, kind="lesson", content="درس قدیمی", source="reflection" + ) + row = await session.get(OrgMemoryRow, memory_id) + row.expires_at = utcnow().replace(year=2020) + await session.commit() + + # No org has billable usage here, so the rollup contributes nothing and the + # merged outcome is exactly the expiry step's. + outcome = await _billing_pass(Namespace(period_days=30))() + + assert outcome.processed == 1 + assert any("expired lesson" in note for note in outcome.notes) + async with session_factory() as session: + row = await session.get(OrgMemoryRow, memory_id) + assert row.deleted_at is not None diff --git a/apps/api/tests/test_memory_grants.py b/apps/api/tests/test_memory_grants.py new file mode 100644 index 0000000..8c9b5a2 --- /dev/null +++ b/apps/api/tests/test_memory_grants.py @@ -0,0 +1,152 @@ +"""Org memory reflection runs only on a model the owner granted the organization. + +The invariants under test: + +* an organization with no granted model learns nothing from a finished task: no + provider call, and the refusal is recorded as ``memory.reflection_no_grant``; +* a granted model is the one that answers, and the cheapest tier wins: the + ``fast`` model is chosen over the ``strong`` one, whatever the grant order. +""" + +from __future__ import annotations + +from contextlib import asynccontextmanager + +from sqlalchemy import select + +from app.application.container_org import ( + build_memory_service, + build_model_grant_service, + build_model_grants, +) +from app.application.ports import ModelResponse +from app.application.ports.memory import TaskOutcome +from app.domain.base import new_id +from app.domain.identity.models import PlatformRole, Role +from app.domain.memory.models import MemoryKind +from app.infrastructure.db.models import ( + AiModelRow, + ErrorEventRow, + MembershipRow, + OrganizationRow, + TaskRow, + UserRow, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +async def _org(session_factory, slug: str) -> str: + """An organization with one owner, written directly to the database.""" + async with session_factory() as session: + user = UserRow( + id=new_id(), + email=f"{slug}@example.com", + password_hash="x", + name=slug, + platform_role=PlatformRole.MANAGER.value, + ) + session.add(user) + await session.flush() + org = OrganizationRow(id=new_id(), name=slug, slug=slug, owner_user_id=user.id) + session.add(org) + await session.flush() + session.add( + MembershipRow(organization_id=org.id, user_id=user.id, role=Role.OWNER.value) + ) + await session.commit() + return org.id + + +async def _task(session_factory, org_id: str, task_id: str) -> None: + """A real task row: a memory's evidence is a reference, not a free string.""" + async with session_factory() as session: + session.add(TaskRow(id=task_id, organization_id=org_id, title=f"کار {task_id}")) + await session.commit() + + +async def _grant(session_factory, org_id: str, *internal_names: str) -> None: + """Grant the named seeded models to the organization, in the order given.""" + async with session_factory() as session: + service = build_model_grant_service(session) + for name in internal_names: + row = ( + await session.execute(select(AiModelRow).where(AiModelRow.internal_name == name)) + ).scalar_one() + await service.grant_model(organization_id=org_id, ai_model_id=row.id) + await session.commit() + + +class _RecordingProvider: + """Answers every reflection with one lesson, and remembers which model was asked.""" + + def __init__(self) -> None: + self.asked: list[str] = [] + + async def complete(self, **kwargs): + model = kwargs["model"] + self.asked.append(model.internal_name) + return ModelResponse( + content="درس ۱: تست‌ها را قبل از ادغام اجرا کن.", + model_internal_name=model.internal_name, + input_tokens=20, + output_tokens=8, + ) + + +@asynccontextmanager +async def _service(session_factory, provider): + """A memory service wired the way the engine wires it, committed on exit.""" + session = session_factory() + try: + service = build_memory_service( + session, provider=provider, grants=build_model_grants(session) + ) + yield service + await session.commit() + finally: + await session.close() + + +def _outcome(task_id: str) -> TaskOutcome: + return TaskOutcome(task_id=task_id, title="ساخت صفحه", status="completed") + + +# --------------------------------------------------------------------------- +# Reflection and grants +# --------------------------------------------------------------------------- + + +async def test_an_organization_without_a_grant_learns_nothing( + session_factory, redirect_error_log, seeded +): + org_id = await _org(session_factory, "no-grant") + provider = _RecordingProvider() + async with _service(session_factory, provider) as service: + # The catalogue has models, but the owner granted this organization none. + await service.on_task_finished(organization_id=org_id, task=_outcome("task-no-grant")) + assert await service.list(org_id) == [] + + assert provider.asked == [] + async with session_factory() as session: + kinds = {row.kind for row in (await session.execute(select(ErrorEventRow))).scalars()} + assert "memory.reflection_no_grant" in kinds + + +async def test_a_granted_model_answers_and_the_cheapest_tier_wins( + session_factory, redirect_error_log, seeded +): + org_id = await _org(session_factory, "granted") + await _task(session_factory, org_id, "task-granted") + # Both models are granted; the reflection must pick the fast one whatever the order. + await _grant(session_factory, org_id, "claude-sonnet-4-5-thinking", "gemini-3-flash") + provider = _RecordingProvider() + async with _service(session_factory, provider) as service: + await service.on_task_finished(organization_id=org_id, task=_outcome("task-granted")) + memories = await service.list(org_id) + + assert provider.asked == ["gemini-3-flash"] + assert [m.kind for m in memories] == [MemoryKind.LESSON] + assert all(m.evidence_task_id == "task-granted" for m in memories) diff --git a/apps/api/tests/test_v2_e2e_smoke.py b/apps/api/tests/test_v2_e2e_smoke.py new file mode 100644 index 0000000..239cd8b --- /dev/null +++ b/apps/api/tests/test_v2_e2e_smoke.py @@ -0,0 +1,185 @@ +"""End-to-end smoke for the v2 seams: wizard, team, manager order, engine task, push, run. + +Only the model is fake: ``tests.engine_fakes.ScriptedModel`` answers every call, reached +through ``app.application.container.get_model_provider``. Git is the default fake +workspace (``GIT_WORKSPACE=fake``, the ``container_delivery._fake_workspace`` singleton), +and the sandbox and domain router are the mock defaults. Everything else is the real +code path: the owner grants the default models over HTTP, the manager completes the +wizard over HTTP (which runs the real bootstrap and delivery hooks), the manager opens a +thread over HTTP without a project as the web UI does, the order goes through the real +``ManagerTurns.chat``, and every pass is one real ``AiRuntimeWorker.drain()``. + +What it proves is that the seams line up. A fresh organization gets its team, its demo +host and its granted models. A thread made without a project still yields a task in the +organization's project, with a pipeline state. The engine starts the project's sandbox +session on its own. The task walks to ``COMPLETED``, and the merge queues a ``dev`` +pipeline run that names the pushed commit. No network, docker or real git. +""" + +from __future__ import annotations + +from sqlalchemy import select + +from app.application.container_delivery import _fake_workspace +from app.application.container_engine import build_scheduler +from app.application.ports.repo_workspace import DiffResult +from app.core.config import settings +from app.domain.workforce.models import RoleKey, TaskStatus +from app.domain.workforce.pipeline import Stage +from app.infrastructure.db.models import ( + AiModelRow, + OrganizationRow, + OrgModelGrantRow, + SandboxSessionRow, + WorkspaceRow, +) +from app.infrastructure.db.models_delivery import AppRuntimeRow +from app.infrastructure.repositories.delivery import PipelineRunRepository +from app.infrastructure.repositories.engine import PipelineStateRepository +from app.infrastructure.repositories.workforce import EmployeeRepository, TaskRepository +from app.workers.ai_runtime import AiRuntimeWorker +from tests.conftest import auth_header +from tests.engine_fakes import ScriptedModel, ScriptedReply +from tests.test_org_wizard import _complete, _manager + +DEFAULTS = "/api/admin/default-model-grants" + +#: The manager's chat reply: one task tag, so the order becomes one task. +ORDER_REPLY = ( + "باشه، یک کار می‌سازم.\n" + '[[CREATE_TASK title="صفحهٔ ورود" kind="feature"]]' + "ساخت صفحهٔ ورود با اعتبارسنجی[[/CREATE_TASK]]" +) + +#: The replies that drive one task from triage to merge, as in test_engine_flow. The +#: pipeline is ``plan_code_test``: the default for a feature, and the one the ASSIGN reply +#: names. It has no review stage, so no review replies are needed. +PIPELINE_REPLIES = [ + '[[ASSIGN employee="" pipeline="plan_code_test"/]]', + "[[PLAN_SPEC]]پلان کار[[/PLAN_SPEC]]", + "[[WORK_REPORT]]پیاده‌سازی شد[[/WORK_REPORT]]", + "[[TEST_PASS]]سبز است[[/TEST_PASS]]", + "[[COMPLETE/]]", +] + + +async def _sandboxes_of(session_factory, organization_id: str, project_id: str) -> list: + """The sandbox sessions of one organization's project.""" + async with session_factory() as session: + stmt = ( + select(SandboxSessionRow) + .join(WorkspaceRow, SandboxSessionRow.workspace_id == WorkspaceRow.id) + .where( + WorkspaceRow.organization_id == organization_id, + WorkspaceRow.project_id == project_id, + ) + ) + return list((await session.execute(stmt)).scalars().all()) + + +async def test_a_manager_order_ships_through_the_engine( + client, session_factory, seeded, platform_owner_token, monkeypatch, redirect_error_log +): + scripted = ScriptedModel( + [ScriptedReply(content=ORDER_REPLY), *(ScriptedReply(content=r) for r in PIPELINE_REPLIES)] + ) + monkeypatch.setattr("app.application.container.get_model_provider", lambda: scripted) + + # The platform owner grants one reasoning model and one cheap model to the default set. + async with session_factory() as session: + models = list( + (await session.execute(select(AiModelRow).order_by(AiModelRow.internal_name))) + .scalars() + .all() + ) + strong = next(m for m in models if "reasoning" in m.capabilities) + cheap = next(m for m in models if "reasoning" not in m.capabilities) + owner = auth_header(platform_owner_token) + for model in (strong, cheap): + granted = await client.put( + f"{DEFAULTS}/{model.id}", json={"is_active": True}, headers=owner + ) + assert granted.status_code == 200, granted.text + + # The manager completes the wizard over HTTP: org, project, team, runtime, grants. + _, headers = await _manager(client) + body = await _complete(client, headers) + org_id = body["organization"]["id"] + slug = body["organization"]["slug"] + project_id = body["project"]["id"] + + async with session_factory() as session: + roles = {e.role_key for e in await EmployeeRepository(session).list_for_org(org_id)} + runtime = await session.get(AppRuntimeRow, project_id) + grants = ( + await session.execute( + select(OrgModelGrantRow).where(OrgModelGrantRow.organization_id == org_id) + ) + ).scalars().all() + owner_user_id = (await session.get(OrganizationRow, org_id)).owner_user_id + + assert {RoleKey.MANAGER, RoleKey.PLANNER, RoleKey.DEVELOPER, RoleKey.TESTER} <= roles + assert runtime is not None + assert runtime.demo_host == f"{slug}.{settings.preview_root_domain}" + assert grants + + # The manager opens a thread the way the web UI does: no project_id. + created = await client.post( + f"/api/organizations/{org_id}/threads", json={"title": "گفتگو"}, headers=headers + ) + assert created.status_code == 201, created.text + assert created.json()["project_id"] is None + thread_id = created.json()["id"] + + # A manager order in that thread becomes one task with a pipeline state. The order goes to + # the engine's manager directly: the HTTP message route builds its manager without pipeline + # states (see test_integration_thread_project.py). + async with session_factory() as session: + await build_scheduler(session).manager.chat( + organization_id=org_id, + thread_id=thread_id, + content="یک صفحهٔ ورود بساز.", + user_id=owner_user_id, + ) + await session.commit() + tasks = await TaskRepository(session).list_for_org(org_id) + states = await PipelineStateRepository(session).list_for_org(org_id) + + assert len(tasks) == 1 + assert len(states) == 1 + assert tasks[0].project_id == project_id + assert states[0].project_id == project_id + task_id = tasks[0].id + + # Nothing has started the project's sandbox yet; the engine must do it. + assert await _sandboxes_of(session_factory, org_id, project_id) == [] + + # The engine walks the task through its stages. Each pass is one real worker drain (the + # legacy drain, then one Scheduler.tick()); the merge pushes to dev. + _fake_workspace.diffs[task_id] = DiffResult( + changed_files=["app/main.py"], text="--- a\n+++ b\n@@\n+x\n" + ) + state = None + for _ in range(10): + await AiRuntimeWorker().drain() + async with session_factory() as session: + state = await PipelineStateRepository(session).get(task_id) + if state is not None and state.stage is Stage.COMPLETED: + break + + async with session_factory() as session: + task = await TaskRepository(session).get(task_id) + + assert state is not None and state.stage is Stage.COMPLETED + assert task is not None and task.status is TaskStatus.DONE + # The engine started the project's sandbox session on its own; no API call did. + assert await _sandboxes_of(session_factory, org_id, project_id) + + pushed_sha = _fake_workspace.heads[(project_id, "dev")] + async with session_factory() as session: + runs = await PipelineRunRepository(session).list_for_project( + organization_id=org_id, project_id=project_id + ) + assert any( + run.trigger == "engine" and run.branch == "dev" and run.sha == pushed_sha for run in runs + ) diff --git a/apps/web/src/app/(app)/app/page.tsx b/apps/web/src/app/(app)/app/page.tsx index 157a15d..d13d019 100644 --- a/apps/web/src/app/(app)/app/page.tsx +++ b/apps/web/src/app/(app)/app/page.tsx @@ -95,12 +95,14 @@ function payload(event: { data: Record }): T { } export default function ApplicationPage() { - const { activeOrganization, user } = useSession(); + const { activeOrganization } = useSession(); const orgId = activeOrganization?.id ?? null; const orgSlug = activeOrganization?.slug ?? ""; - // Org membership is enough to *see* everything here; the server is the real - // boundary for the actions, which is why a 403 still surfaces as a message. - const canManage = Boolean(user); + // Org membership is enough to *see* everything here. The actions (re-run, + // rollback, production) need the org's admin or owner role, the same rule the + // server applies; the server stays the real boundary, which is why a 403 still + // surfaces as a message. + const canManage = activeOrganization?.role === "owner" || activeOrganization?.role === "admin"; const [tab, setTab] = React.useState("pipeline"); const [project, setProject] = React.useState(null); diff --git a/apps/web/src/app/(app)/memory/page.tsx b/apps/web/src/app/(app)/memory/page.tsx index 9bc1e54..d7271bf 100644 --- a/apps/web/src/app/(app)/memory/page.tsx +++ b/apps/web/src/app/(app)/memory/page.tsx @@ -136,7 +136,7 @@ export default function MemoryPage() { // Live: another member (or a reflection) changed memory while this page was open. useLiveChannel(orgId ? [`org:${orgId}`] : [], (event) => { - if ((event.type as string) === MEMORY_CHANGED_EVENT) void load(); + if (event.type === MEMORY_CHANGED_EVENT) void load(); }); const visible = React.useMemo( diff --git a/apps/web/src/app/(app)/settings/_components/sharing-section.tsx b/apps/web/src/app/(app)/settings/_components/sharing-section.tsx new file mode 100644 index 0000000..fda9228 --- /dev/null +++ b/apps/web/src/app/(app)/settings/_components/sharing-section.tsx @@ -0,0 +1,339 @@ +"use client"; + +/** + * Sharing controls on the settings "members" tab (v2 TODO 1.10): the invitations + * nobody has accepted yet, handing the organization to an active admin, and + * leaving it. + * + * The server is the boundary for every rule: who may list or revoke invitations, + * that only the owner transfers, that the target must be an active admin, that the + * typed slug must match, and that the last owner cannot leave. The UI only hides + * the controls a role can never use, and shows the server's message when a request + * is still refused. + */ + +import * as React from "react"; +import { useRouter } from "next/navigation"; +import type { Member, OrganizationWithRole } from "@agy/contracts"; +import { + Badge, + Button, + Card, + CardHint, + CardTitle, + ConfirmDialog, + EmptyState, + ErrorState, + Field, + LoadingState, + Modal, + Select, +} from "@agy/ui"; +import { ApiError } from "@/lib/api"; +import { orgApi } from "@/lib/api-org"; +import { useSession } from "@/lib/session"; + +const ROLE_LABELS: Record = { + owner: "مالک", + admin: "مدیر", + manager: "سرپرست", + member: "عضو", + viewer: "بیننده", +}; + +interface PendingInvite { + id: string; + email: string; + role: string; + expires_at: string; + accepted_at: string | null; +} + +function errorText(err: unknown, fallback: string): string { + return err instanceof ApiError ? err.message : fallback; +} + +export interface SharingSectionProps { + org: OrganizationWithRole; + members: Member[]; + /** Pending invitations are shown to members who can manage members. */ + canManageMembers: boolean; + /** Called after a change that makes the parent's member list stale. */ + onChanged: () => void; +} + +export function SharingSection({ + org, + members, + canManageMembers, + onChanged, +}: SharingSectionProps) { + const router = useRouter(); + const { organizations, refresh } = useSession(); + const isOwner = org.role === "owner"; + + // --- pending invitations --------------------------------------------------- + const [invites, setInvites] = React.useState([]); + const [invitesLoading, setInvitesLoading] = React.useState(false); + const [invitesError, setInvitesError] = React.useState(null); + const [pendingRevoke, setPendingRevoke] = React.useState(null); + const [revoking, setRevoking] = React.useState(false); + + // --- transfer ownership (owner only) --------------------------------------- + const [targetUserId, setTargetUserId] = React.useState(""); + const [transferOpen, setTransferOpen] = React.useState(false); + const [slugInput, setSlugInput] = React.useState(""); + const [transferring, setTransferring] = React.useState(false); + const [transferError, setTransferError] = React.useState(null); + + // --- leave ------------------------------------------------------------------- + const [leaveOpen, setLeaveOpen] = React.useState(false); + const [leaving, setLeaving] = React.useState(false); + const [leaveError, setLeaveError] = React.useState(null); + + const admins = members.filter((m) => m.role === "admin" && m.status === "active"); + const target = admins.find((m) => m.user_id === targetUserId) ?? null; + // Same comparison the server makes (trimmed, case-insensitive), so the button + // enables exactly when the request would be accepted. + const slugMatches = slugInput.trim().toLowerCase() === org.slug.trim().toLowerCase(); + + React.useEffect(() => { + if (!canManageMembers) return; + let cancelled = false; + setInvitesLoading(true); + setInvitesError(null); + orgApi + .listOrgInvites(org.id) + .then((rows) => { + if (!cancelled) setInvites(rows.filter((row) => row.accepted_at == null)); + }) + .catch((err: unknown) => { + if (!cancelled) setInvitesError(errorText(err, "دریافت دعوت‌های در انتظار ناموفق بود.")); + }) + .finally(() => { + if (!cancelled) setInvitesLoading(false); + }); + return () => { + cancelled = true; + }; + }, [canManageMembers, org.id]); + + const onRevoke = async () => { + const invite = pendingRevoke; + if (!invite) return; + setRevoking(true); + setInvitesError(null); + try { + await orgApi.revokeOrgInvite(org.id, invite.id); + setInvites((current) => current.filter((row) => row.id !== invite.id)); + } catch (err) { + setInvitesError(errorText(err, "لغو دعوت ناموفق بود.")); + } finally { + setPendingRevoke(null); + setRevoking(false); + } + }; + + const onTransfer = async () => { + if (!target || !slugMatches) return; + setTransferring(true); + setTransferError(null); + try { + await orgApi.transferOrgOwnership(org.id, { + new_owner_user_id: target.user_id, + confirm: slugInput.trim(), + }); + setTransferOpen(false); + setSlugInput(""); + setTargetUserId(""); + // Our own role just changed (owner to admin), and the session carries it. + await refresh(); + onChanged(); + } catch (err) { + setTransferError(errorText(err, "انتقال مالکیت ناموفق بود.")); + } finally { + setTransferring(false); + } + }; + + const onLeave = async () => { + setLeaving(true); + setLeaveError(null); + try { + await orgApi.leaveOrganization(org.id); + setLeaveOpen(false); + const remaining = organizations.filter((o) => o.id !== org.id); + await refresh(); + // Nothing left to work in: the creation wizard is the only useful next screen. + if (remaining.length === 0) router.replace("/onboarding"); + } catch (err) { + setLeaveOpen(false); + setLeaveError(errorText(err, "ترک سازمان ناموفق بود.")); + } finally { + setLeaving(false); + } + }; + + return ( + <> + {canManageMembers && ( + + دعوت‌های در انتظار پذیرش + + لینک‌هایی که هنوز استفاده نشده‌اند. اگر لینکی به نشانی اشتباهی رفته، آن را لغو کنید. + +
+ {invitesError ? ( + + ) : invitesLoading ? ( + + ) : invites.length === 0 ? ( + + ) : ( +
    + {invites.map((invite) => ( +
  • +
    + {invite.email} + + معتبر تا {new Date(invite.expires_at).toLocaleDateString("fa-IR")} + +
    +
    + {ROLE_LABELS[invite.role] ?? invite.role} + +
    +
    • + ))} +
+ )} +
+ + )} + + {isOwner && ( + + انتقال مالکیت + + مالکیت را به یکی از مدیران فعال سازمان بسپارید. پس از انتقال، شما به مدیر سازمان + تبدیل می‌شوید. + + {admins.length === 0 ? ( +

+ هنوز مدیر فعالی در سازمان نیست. ابتدا یکی از اعضا را به نقش «مدیر» ببرید. +

+ ) : ( +
+
+