نسخهٔ وب مستقل — پلتفرم نیروی کار هوش مصنوعی

مخزن اختصاصی برنامهٔ وب؛ محتوای پوشهٔ web در ریشهٔ مخزن قرار گرفت.
تاریخچهٔ پروژهٔ ربات (agy-bot) در این مخزن وجود ندارد و هیچ وابستگی runtime به آن نیست.

apps/api/app/application/services/audit_service.py

@@ -0,0 +1,37 @@
+"""Audit helper used by the HTTP layer.
+
+Every sensitive operation calls this. It is deliberately cheap (one INSERT) so
+there is no excuse to skip it.
+"""
+
+from __future__ import annotations
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.domain.audit.models import AuditEvent
+from app.infrastructure.repositories.audit import AuditRepository
+
+
+async def record_audit(
+    session: AsyncSession,
+    *,
+    action: str,
+    organization_id: str | None = None,
+    actor_user_id: str | None = None,
+    target_type: str = "",
+    target_id: str = "",
+    metadata: dict | None = None,
+    ip: str = "",
+    user_agent: str = "",
+) -> AuditEvent:
+    event = AuditEvent(
+        action=action,
+        organization_id=organization_id,
+        actor_user_id=actor_user_id,
+        target_type=target_type,
+        target_id=target_id,
+        metadata_json=metadata or {},
+        ip_address=ip,
+        user_agent=user_agent,
+    )
+    return await AuditRepository(session).record(event=event)

apps/api/app/application/services/billing_service.py

@@ -0,0 +1,184 @@
+"""Billing use cases.
+
+Design rule, stated once and enforced here: **the internal usage ledger is the
+source of truth.** A payment provider only moves money; it never decides what a
+tenant owes. That is why ``summary`` computes totals from ``UsageLedgerRepository``
+and not from any external invoice.
+
+Credit is spent, not checked-then-spent: the debit is a single adjustment, so two
+concurrent runs cannot both pass a stale affordability check.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from app.application.ports import PaymentProviderPort
+from app.core.errors import NotFoundError, PermissionDeniedError, ValidationError
+from app.domain.billing.models import (
+    CreditWallet,
+    Invoice,
+    Payment,
+    PaymentStatus,
+    Subscription,
+)
+from app.infrastructure.repositories.audit import UsageLedgerRepository
+from app.infrastructure.repositories.billing import (
+    CreditWalletRepository,
+    InvoiceRepository,
+    PaymentRepository,
+    PlanRepository,
+    SubscriptionRepository,
+)
+
+#: A tenant may not buy an arbitrary amount of credit in one action.
+MIN_TOPUP = 1.0
+MAX_TOPUP = 10_000.0
+
+
+class BillingService:
+    def __init__(
+        self,
+        *,
+        wallets: CreditWalletRepository,
+        invoices: InvoiceRepository,
+        payments: PaymentRepository,
+        plans: PlanRepository,
+        subscriptions: SubscriptionRepository,
+        ledger: UsageLedgerRepository,
+        payment_provider: PaymentProviderPort,
+    ) -> None:
+        self.wallets = wallets
+        self.invoices = invoices
+        self.payments = payments
+        self.plans = plans
+        self.subscriptions = subscriptions
+        self.ledger = ledger
+        self.payment_provider = payment_provider
+
+    # -- reads ---------------------------------------------------------------
+
+    async def wallet_for(self, organization_id: str) -> CreditWallet:
+        return await self.wallets.ensure(organization_id)
+
+    async def summary(
+        self,
+        *,
+        organization_id: str,
+        since: datetime | None = None,
+        until: datetime | None = None,
+    ) -> dict:
+        wallet = await self.wallets.ensure(organization_id)
+        breakdown = await self.ledger.breakdown(
+            organization_id=organization_id, since=since, until=until
+        )
+        return {
+            "balance": wallet.balance,
+            "currency": wallet.currency,
+            "low_balance": wallet.is_low,
+            "low_balance_threshold": wallet.low_balance_threshold,
+            "total_spend": breakdown.total_price,
+            "by_unit": breakdown.by_unit,
+            "by_employee": breakdown.by_employee,
+            "by_project": breakdown.by_project,
+        }
+
+    async def list_usage_events(
+        self, *, organization_id: str, limit: int = 100, offset: int = 0
+    ) -> list:
+        return await self.ledger.list_events(
+            organization_id=organization_id, limit=limit, offset=offset
+        )
+
+    async def list_invoices(self, organization_id: str) -> list[Invoice]:
+        return await self.invoices.list_for_org(organization_id)
+
+    async def list_payments(self, organization_id: str) -> list[Payment]:
+        return await self.payments.list_for_org(organization_id)
+
+    async def current_subscription(self, organization_id: str) -> Subscription | None:
+        return await self.subscriptions.get_for_org(organization_id)
+
+    # -- writes --------------------------------------------------------------
+
+    async def add_credit(
+        self, *, organization_id: str, amount: float, allow_negative: bool = False
+    ) -> CreditWallet:
+        """Grant (or, for a platform owner, claw back) credit.
+
+        ``allow_negative`` is reserved for the platform owner console; an org
+        admin may only ever add credit.
+        """
+        if amount == 0:
+            raise ValidationError("مقدار اعتبار نمی‌تواند صفر باشد.")
+        if amount < 0 and not allow_negative:
+            raise PermissionDeniedError("کسر اعتبار فقط توسط مالک سامانه انجام می‌شود.")
+        if not allow_negative and amount > MAX_TOPUP:
+            raise ValidationError(f"حداکثر افزایش اعتبار در هر عملیات {MAX_TOPUP} است.")
+
+        wallet = await self.wallets.adjust(organization_id=organization_id, delta=amount)
+        if wallet.balance < 0:
+            # Refuse to leave a wallet in debt; undo and surface the conflict.
+            await self.wallets.adjust(organization_id=organization_id, delta=-amount)
+            raise ValidationError("اعتبار کافی برای این کسر وجود ندارد.")
+        return wallet
+
+    async def start_checkout(
+        self,
+        *,
+        organization_id: str,
+        amount: float,
+        success_url: str,
+        cancel_url: str,
+    ) -> tuple[Payment, str]:
+        """Create a pending payment plus the provider's hosted checkout URL."""
+        if amount < MIN_TOPUP:
+            raise ValidationError(f"حداقل مبلغ شارژ {MIN_TOPUP} است.")
+        if amount > MAX_TOPUP:
+            raise ValidationError(f"حداکثر مبلغ شارژ {MAX_TOPUP} است.")
+
+        wallet = await self.wallets.ensure(organization_id)
+        payment = await self.payments.create(
+            Payment(
+                organization_id=organization_id,
+                amount=amount,
+                status=PaymentStatus.PENDING,
+                provider="mock",
+            )
+        )
+        session = await self.payment_provider.create_checkout(
+            organization_id=organization_id,
+            amount=amount,
+            currency=wallet.currency,
+            success_url=success_url,
+            cancel_url=cancel_url,
+        )
+        updated = await self.payments.set_status(
+            payment.id, PaymentStatus.PENDING, external_id=session.external_id
+        )
+        return (updated or payment), session.url
+
+    async def settle_payment(
+        self, *, organization_id: str, payment_id: str, succeeded: bool
+    ) -> Payment:
+        """Mark a payment final and credit the wallet when it succeeded.
+
+        This is the only path that turns money into credit, so it is idempotent
+        with respect to status: a payment that is already final is returned
+        untouched rather than credited twice.
+        """
+        payment = next(
+            (p for p in await self.payments.list_for_org(organization_id) if p.id == payment_id),
+            None,
+        )
+        if payment is None:
+            raise NotFoundError("پرداخت یافت نشد.")
+        if payment.status != PaymentStatus.PENDING:
+            return payment
+
+        if succeeded:
+            await self.wallets.adjust(organization_id=organization_id, delta=payment.amount)
+            final = await self.payments.set_status(payment_id, PaymentStatus.SUCCEEDED)
+        else:
+            final = await self.payments.set_status(payment_id, PaymentStatus.FAILED)
+        return final or payment

apps/api/app/application/services/knowledge_service.py

@@ -0,0 +1,202 @@
+"""Knowledge Base use cases: ingest text, chunk it, and retrieve it safely.
+
+Two things matter here:
+
+* **Chunking is deterministic.** The same text always produces the same chunks,
+  which makes ingestion idempotent-ish and the tests meaningful.
+* **Retrieval is permission-aware.** Search is always scoped to one organization,
+  and optionally narrowed to one project. An org-wide source is visible to every
+  project; a project-scoped source is not visible from another project.
+
+Embeddings are computed through ``VectorStorePort`` when a real adapter is
+configured. With the mock adapter no vectors are produced and retrieval falls
+back to keyword matching — the same fallback the plan calls for at MVP.
+"""
+
+from __future__ import annotations
+
+from app.application.ports import VectorStorePort
+from app.core.errors import NotFoundError, ValidationError
+from app.domain.knowledge.models import (
+    Document,
+    DocumentChunk,
+    EmbeddingJob,
+    EmbeddingJobStatus,
+    KnowledgeSource,
+    RetrievalPolicy,
+    SourceKind,
+)
+from app.infrastructure.repositories.knowledge import (
+    DocumentChunkRepository,
+    DocumentRepository,
+    EmbeddingJobRepository,
+    KnowledgeSourceRepository,
+    RetrievalPolicyRepository,
+)
+
+#: Chunk size in words. Small enough to keep a retrieved passage focused.
+CHUNK_WORDS = 180
+CHUNK_OVERLAP_WORDS = 30
+MAX_INGEST_CHARS = 400_000
+
+
+def chunk_text(
+    text: str, *, size: int = CHUNK_WORDS, overlap: int = CHUNK_OVERLAP_WORDS
+) -> list[str]:
+    """Split text into overlapping word windows.
+
+    Pure and deterministic — no regex surprises, no dependence on punctuation.
+    """
+    words = text.split()
+    if not words:
+        return []
+    if size <= 0:
+        raise ValueError("chunk size must be positive")
+    step = max(1, size - max(0, overlap))
+
+    chunks: list[str] = []
+    for start in range(0, len(words), step):
+        window = words[start : start + size]
+        if not window:
+            break
+        chunks.append(" ".join(window))
+        if start + size >= len(words):
+            break
+    return chunks
+
+
+class KnowledgeService:
+    def __init__(
+        self,
+        *,
+        sources: KnowledgeSourceRepository,
+        documents: DocumentRepository,
+        chunks: DocumentChunkRepository,
+        jobs: EmbeddingJobRepository,
+        policies: RetrievalPolicyRepository,
+        vector_store: VectorStorePort,
+    ) -> None:
+        self.sources = sources
+        self.documents = documents
+        self.chunks = chunks
+        self.jobs = jobs
+        self.policies = policies
+        self.vector_store = vector_store
+
+    # -- ingestion -----------------------------------------------------------
+
+    async def create_text_source(
+        self,
+        *,
+        organization_id: str,
+        name: str,
+        content: str,
+        project_id: str | None = None,
+        title: str = "",
+    ) -> tuple[KnowledgeSource, int]:
+        if not name.strip():
+            raise ValidationError("نام منبع دانش نمی‌تواند خالی باشد.")
+        if not content.strip():
+            raise ValidationError("محتوای منبع دانش نمی‌تواند خالی باشد.")
+        if len(content) > MAX_INGEST_CHARS:
+            raise ValidationError("حجم محتوا بیش از حد مجاز است.")
+
+        source = await self.sources.create(
+            KnowledgeSource(
+                organization_id=organization_id,
+                name=name.strip(),
+                kind=SourceKind.TEXT,
+                project_id=project_id,
+            )
+        )
+        document = await self.documents.create(
+            Document(source_id=source.id, title=title.strip() or source.name, content=content)
+        )
+
+        pieces = chunk_text(content)
+        chunk_rows = [
+            DocumentChunk(document_id=document.id, index=index, content=piece)
+            for index, piece in enumerate(pieces)
+        ]
+        created = await self.chunks.create_many(chunk_rows)
+
+        job = await self.jobs.create(EmbeddingJob(source_id=source.id))
+        try:
+            # The mock vector store is a no-op that accepts and discards; a real
+            # adapter embeds here. Keyword search below works either way.
+            await self.vector_store.upsert(
+                organization_id=organization_id,
+                chunks=[(chunk.id, chunk.content, []) for chunk in chunk_rows],
+                metadata={"source_id": source.id, "project_id": project_id},
+            )
+            await self.jobs.set_status(job.id, EmbeddingJobStatus.DONE)
+        except Exception as exc:  # noqa: BLE001 - embedding is best-effort at MVP
+            # Ingestion still succeeded; only the vector index is degraded, and
+            # keyword search keeps working. Record why rather than failing hard.
+            await self.jobs.set_status(job.id, EmbeddingJobStatus.FAILED, error=str(exc))
+
+        return source, created
+
+    # -- reads ---------------------------------------------------------------
+
+    async def list_sources(
+        self, organization_id: str, *, limit: int = 100, offset: int = 0
+    ) -> list[KnowledgeSource]:
+        return await self.sources.list_for_org(organization_id, limit=limit, offset=offset)
+
+    async def get_source_or_404(
+        self, *, organization_id: str, source_id: str
+    ) -> KnowledgeSource:
+        source = await self.sources.get_for_org(organization_id, source_id)
+        if source is None:
+            raise NotFoundError("منبع دانش یافت نشد.")
+        return source
+
+    async def source_detail(self, *, source_id: str) -> dict:
+        documents = await self.documents.list_for_source(source_id)
+        return {
+            "documents": documents,
+            "chunk_count": await self.chunks.count_for_source(source_id),
+            "job": await self.jobs.latest_for_source(source_id),
+        }
+
+    async def chunk_counts(self, source_ids: list[str]) -> dict[str, int]:
+        """Chunk counts for many sources at once — see ``count_for_sources``."""
+        return await self.chunks.count_for_sources(source_ids)
+
+    async def search(
+        self,
+        *,
+        organization_id: str,
+        query: str,
+        project_id: str | None = None,
+        limit: int = 5,
+    ) -> list[dict]:
+        return await self.chunks.keyword_search(
+            organization_id=organization_id,
+            query=query,
+            project_id=project_id,
+            limit=limit,
+        )
+
+    async def delete_source(self, *, organization_id: str, source_id: str) -> None:
+        source = await self.get_source_or_404(
+            organization_id=organization_id, source_id=source_id
+        )
+        await self.sources.delete(source.id)
+
+    # -- retrieval policies --------------------------------------------------
+
+    async def list_policies(self, organization_id: str) -> list[RetrievalPolicy]:
+        return await self.policies.list_for_org(organization_id)
+
+    async def create_policy(
+        self, *, organization_id: str, name: str, scope: dict | None = None
+    ) -> RetrievalPolicy:
+        if not name.strip():
+            raise ValidationError("نام سیاست بازیابی نمی‌تواند خالی باشد.")
+        return await self.policies.create(
+            RetrievalPolicy(
+                organization_id=organization_id, name=name.strip(), scope_json=dict(scope or {})
+            )
+        )

apps/api/app/application/services/project_service.py

@@ -0,0 +1,243 @@
+"""Project and preview-domain use cases.
+
+This is the application layer: it orchestrates domain rules, repositories and the
+driven ports (Git, domain router). It knows nothing about HTTP.
+
+Security properties enforced here (not in the router, not only in the adapter):
+
+* a project repository is always created **private**
+* the repository name is collision-free across tenants
+* the per-project Git password is encrypted before it is stored and never
+  returned to a caller
+* a preview host must be a single label under the configured root domain
+* a preview host is unique platform-wide
+* the proxy target port must be allocatable and must not be a protected
+  platform port
+* a project may hold at most ``max_subdomains_per_project`` hosts
+"""
+
+from __future__ import annotations
+
+import hashlib
+
+from app.application.ports import (
+    DomainRouterPort,
+    GitCredentials,
+    GitProviderPort,
+    RepoInfo,
+    RepoStatus,
+)
+from app.core.config import settings
+from app.core.crypto import decrypt, encrypt
+from app.core.errors import ConflictError, NotFoundError, PermissionDeniedError, ValidationError
+from app.domain.base import ensure_unique_slug, slugify
+from app.domain.platform.rules import (
+    assert_not_protected_port,
+    assert_port_allocatable,
+    build_host,
+    validate_subdomain_label,
+)
+from app.domain.projects.models import (
+    Project,
+    ProjectGitCredential,
+    ProjectGitRepo,
+)
+from app.infrastructure.repositories.projects import (
+    PreviewDomainRepository,
+    ProjectGitCredentialRepository,
+    ProjectGitRepoRepository,
+    ProjectRepository,
+)
+
+
+class ProjectService:
+    def __init__(
+        self,
+        *,
+        projects: ProjectRepository,
+        repos: ProjectGitRepoRepository,
+        credentials: ProjectGitCredentialRepository,
+        domains: PreviewDomainRepository,
+        git: GitProviderPort,
+        router: DomainRouterPort,
+    ) -> None:
+        self.projects = projects
+        self.repos = repos
+        self.credentials = credentials
+        self.domains = domains
+        self.git = git
+        self.router = router
+
+    # -- projects ------------------------------------------------------------
+
+    async def create_project(
+        self,
+        *,
+        organization_id: str,
+        organization_slug: str,
+        created_by_user_id: str,
+        name: str,
+        description: str = "",
+    ) -> Project:
+        slug = ensure_unique_slug(slugify(name), await self.projects.list_slugs(organization_id))
+        return await self.projects.create(
+            Project(
+                organization_id=organization_id,
+                name=name.strip(),
+                slug=slug,
+                description=description,
+                created_by_user_id=created_by_user_id,
+            )
+        )
+
+    async def get_or_404(self, *, organization_id: str, project_id: str) -> Project:
+        project = await self.projects.get(project_id)
+        # Returning 404 (not 403) keeps the existence of other tenants' projects
+        # out of the response.
+        if project is None or project.organization_id != organization_id:
+            raise NotFoundError("پروژه یافت نشد.")
+        return project
+
+    @staticmethod
+    def repo_name_for(*, organization_slug: str, project: Project) -> str:
+        """``orgslug-projectslug-shortid`` — stable and collision-free."""
+        digest = hashlib.sha1(f"{organization_slug}:{project.id}".encode()).hexdigest()[:6]
+        raw = f"{organization_slug}-{project.slug}-{digest}"
+        return raw[:100].strip("-")
+
+    # -- git -----------------------------------------------------------------
+
+    async def ensure_repository(
+        self, *, organization_slug: str, project: Project
+    ) -> ProjectGitRepo:
+        repo_name = self.repo_name_for(organization_slug=organization_slug, project=project)
+        info: RepoInfo = await self.git.ensure_project_repo(
+            project_id=project.id,
+            repo_name=repo_name,
+            description=f"مخزن پروژهٔ {project.name}",
+        )
+
+        credentials: GitCredentials = await self.git.ensure_credentials(
+            project_id=project.id, owner=info.owner, repo_name=info.repo_name
+        )
+        await self.credentials.upsert(
+            ProjectGitCredential(
+                project_id=project.id,
+                username=credentials.username,
+                # Encrypted at rest. The plaintext only exists inside this method.
+                encrypted_password=encrypt(credentials.password),
+            )
+        )
+
+        return await self.repos.upsert(
+            ProjectGitRepo(
+                project_id=project.id,
+                provider=settings.git_provider,
+                owner=info.owner,
+                repo_name=info.repo_name,
+                clone_url=info.clone_url,
+                web_url=info.web_url,
+                default_branch=info.default_branch,
+            )
+        )
+
+    async def repo_status(self, *, project: Project) -> RepoStatus:
+        record = await self.repos.get_for_project(project.id)
+        if record is None:
+            return RepoStatus(exists=False)
+        # Identity comes from our own record so the adapter stays stateless.
+        return await self.git.get_repo_status(
+            project_id=project.id, owner=record.owner, repo_name=record.repo_name
+        )
+
+    async def clone_credentials(self, *, project: Project) -> tuple[str, str]:
+        """``(username, password)`` for a one-off clone.
+
+        Only the application layer may call this, and the router must never
+        serialise the result.
+        """
+        record = await self.credentials.get_for_project(project.id)
+        if record is None:
+            raise NotFoundError("اطلاعات دسترسی مخزن ثبت نشده است.")
+        return record.username, decrypt(record.encrypted_password)
+
+    # -- preview domains -----------------------------------------------------
+
+    async def allocate_port(self, *, project: Project, service_name: str = "web") -> int:
+        existing = await self.domains.get_port(project.id, service_name)
+        if existing is not None:
+            return existing
+
+        start, end = settings.preview_port_start, settings.preview_port_end
+        used = await self.domains.list_used_ports(start=start, end=end)
+        for port in range(start, end + 1):
+            if port in used:
+                continue
+            try:
+                assert_port_allocatable(port, start=start, end=end)
+            except ValueError:
+                continue
+            await self.domains.add_port(
+                organization_id=project.organization_id,
+                project_id=project.id,
+                service_name=service_name,
+                port=port,
+            )
+            return port
+        raise ConflictError("پورت آزادی در بازهٔ تعیین‌شده وجود ندارد.")
+
+    async def create_preview_domain(
+        self,
+        *,
+        project: Project,
+        requested_subdomain: str,
+        created_by_user_id: str,
+        port: int | None = None,
+    ) -> dict:
+        label = validate_subdomain_label(requested_subdomain)
+        host = build_host(label, settings.preview_root_domain)
+
+        if await self.domains.get_by_host(host) is not None:
+            raise ConflictError(f"دامنهٔ {host} قبلاً ثبت شده است.")
+
+        count = await self.domains.count_for_project(project.id)
+        if count >= settings.max_subdomains_per_project:
+            raise ConflictError(
+                f"هر پروژه حداکثر {settings.max_subdomains_per_project} دامنه می‌تواند داشته باشد."
+            )
+
+        target_port = port if port is not None else await self.allocate_port(project=project)
+        # A tenant must never be able to expose an internal platform service.
+        assert_not_protected_port(target_port)
+
+        result = await self.router.configure_reverse_proxy(host=host, port=target_port)
+        if not result.ok:
+            raise ValidationError(result.message or "پیکربندی دامنه ناموفق بود.")
+
+        record = await self.domains.create(
+            organization_id=project.organization_id,
+            project_id=project.id,
+            host=host,
+            port=target_port,
+            created_by_user_id=created_by_user_id,
+        )
+        record["url"] = result.url or f"https://{host}"
+        return record
+
+    async def remove_preview_domain(self, *, project: Project, domain_id: str) -> None:
+        record = await self.domains.get(domain_id)
+        if record is None or record["project_id"] != project.id:
+            raise NotFoundError("دامنه یافت نشد.")
+        if record["protected"]:
+            raise PermissionDeniedError(
+                "این دامنه محافظت‌شده است و فقط مالک سامانه می‌تواند آن را تغییر دهد."
+            )
+
+        await self.router.remove_preview_domain(domain_id=record["host"])
+        await self.domains.delete(domain_id)
+
+    async def list_preview_domains(self, *, project: Project) -> list[dict]:
+        records = await self.domains.list_for_project(project.id)
+        for record in records:
+            record["url"] = f"https://{record['host']}"
+        return records

apps/api/app/application/services/sandbox_service.py

@@ -0,0 +1,454 @@
+"""Sandbox Gateway use cases.
+
+The security posture is the whole point of this module:
+
+* **No host execution path.** The only way to run anything is through
+  ``SandboxProviderPort``. There is no fallback that shells out locally.
+* **Limits come from policy, never from the caller.** A request cannot ask for a
+  longer timeout, more memory, or a bigger output cap than its scope allows.
+* **Egress is deny-by-default.** The stored ``network_policy`` defaults to
+  ``deny_all``; nothing here opens it implicitly.
+* **Every command is metered.** CPU seconds are written to the usage ledger, so
+  sandbox time is billable and auditable.
+"""
+
+from __future__ import annotations
+
+import contextlib
+from datetime import datetime
+
+from app.application.ports import CommandResult, SandboxProviderPort, UsageRecord
+from app.core.config import settings
+from app.core.errors import ConflictError, NotFoundError, ProviderUnavailableError, ValidationError
+from app.domain.ai_control.models import RateUnit
+from app.domain.base import utcnow
+from app.domain.sandbox.models import (
+    CommandRun,
+    NetworkPolicy,
+    PreviewUrl,
+    SandboxProviderKind,
+    SandboxResourceLimit,
+    SandboxSession,
+    SessionStatus,
+    Snapshot,
+    Workspace,
+    WorkspaceStatus,
+)
+from app.infrastructure.repositories.ai_control import RateCardRepository
+from app.infrastructure.repositories.audit import UsageLedgerRepository
+from app.infrastructure.repositories.projects import ProjectRepository
+from app.infrastructure.repositories.sandbox import (
+    CommandRunRepository,
+    PreviewUrlRepository,
+    ResourceLimitRepository,
+    SandboxSessionRepository,
+    SandboxTemplateRepository,
+    SnapshotRepository,
+    WorkspaceRepository,
+)
+
+#: Hard ceiling, independent of policy. A policy may be lower, never higher.
+ABSOLUTE_MAX_TIMEOUT_SECONDS = 900
+ABSOLUTE_MAX_OUTPUT_BYTES = 2_000_000
+MAX_COMMAND_LENGTH = 4_000
+
+
+class SandboxService:
+    def __init__(
+        self,
+        *,
+        templates: SandboxTemplateRepository,
+        workspaces: WorkspaceRepository,
+        sessions: SandboxSessionRepository,
+        commands: CommandRunRepository,
+        snapshots: SnapshotRepository,
+        previews: PreviewUrlRepository,
+        limits: ResourceLimitRepository,
+        rate_cards: RateCardRepository,
+        ledger: UsageLedgerRepository,
+        projects: ProjectRepository,
+        provider: SandboxProviderPort,
+    ) -> None:
+        self.templates = templates
+        self.workspaces = workspaces
+        self.sessions = sessions
+        self.commands = commands
+        self.snapshots = snapshots
+        self.previews = previews
+        self.limits = limits
+        self.rate_cards = rate_cards
+        self.ledger = ledger
+        self.projects = projects
+        self.provider = provider
+
+    # -- limits --------------------------------------------------------------
+
+    async def effective_limits(
+        self, *, organization_id: str, project_id: str | None = None
+    ) -> SandboxResourceLimit:
+        """Resolve the strictest applicable policy.
+
+        Project scope wins over organization scope; the absolute ceiling always
+        wins over both.
+        """
+        if project_id:
+            scoped = await self.limits.get_for_scope("project", project_id)
+            if scoped is not None:
+                return self._clamp(scoped)
+
+        org_scoped = await self.limits.get_for_scope("organization", organization_id)
+        if org_scoped is not None:
+            return self._clamp(org_scoped)
+
+        return self._clamp(
+            SandboxResourceLimit(
+                scope="platform",
+                scope_id=None,
+                timeout_seconds=settings.sandbox_command_timeout_seconds,
+                max_output_bytes=settings.sandbox_max_output_bytes,
+            )
+        )
+
+    @staticmethod
+    def _clamp(limit: SandboxResourceLimit) -> SandboxResourceLimit:
+        limit.timeout_seconds = max(
+            1, min(int(limit.timeout_seconds), ABSOLUTE_MAX_TIMEOUT_SECONDS)
+        )
+        limit.max_output_bytes = max(
+            1024, min(int(limit.max_output_bytes), ABSOLUTE_MAX_OUTPUT_BYTES)
+        )
+        limit.memory_mb = max(128, int(limit.memory_mb))
+        limit.disk_mb = max(256, int(limit.disk_mb))
+        return limit
+
+    # -- templates -----------------------------------------------------------
+
+    async def list_templates(self):
+        return await self.templates.list()
+
+    # -- workspaces ----------------------------------------------------------
+
+    async def create_workspace(
+        self, *, organization_id: str, project_id: str, template_id: str
+    ) -> Workspace:
+        template = await self.templates.get(template_id)
+        if template is None:
+            raise NotFoundError("قالب سندباکس یافت نشد.")
+
+        provider_kind = str(template.provider) or SandboxProviderKind.MOCK
+        try:
+            info = await self.provider.create_workspace(
+                organization_id=organization_id,
+                project_id=project_id,
+                template_id=template_id,
+                image=template.image,
+            )
+        except Exception as exc:  # noqa: BLE001 - surface as a provider outage
+            raise ProviderUnavailableError(
+                "ساخت محیط اجرای ایزوله ناموفق بود."
+            ) from exc
+
+        workspace = await self.workspaces.create(
+            Workspace(
+                organization_id=organization_id,
+                project_id=project_id,
+                template_id=template_id,
+                provider=provider_kind,
+                external_workspace_id=info.external_id,
+                status=WorkspaceStatus.READY,
+            )
+        )
+        return workspace
+
+    async def list_workspaces(
+        self,
+        *,
+        organization_id: str,
+        project_id: str,
+        limit: int = 100,
+        offset: int = 0,
+    ) -> list[Workspace]:
+        """Workspaces for a project the caller's tenant actually owns."""
+        project = await self.projects.get(project_id)
+        # 404 rather than an empty list, so a foreign project id is
+        # indistinguishable from a non-existent one.
+        if project is None or project.organization_id != organization_id:
+            raise NotFoundError("پروژه یافت نشد.")
+        return await self.workspaces.list_for_project(
+            project_id, limit=limit, offset=offset
+        )
+
+    async def get_workspace_or_404(
+        self, *, organization_id: str, workspace_id: str
+    ) -> Workspace:
+        workspace = await self.workspaces.get_for_org(organization_id, workspace_id)
+        if workspace is None:
+            raise NotFoundError("محیط اجرا یافت نشد.")
+        return workspace
+
+    # -- sessions ------------------------------------------------------------
+
+    async def start_session(
+        self, *, organization_id: str, workspace_id: str
+    ) -> SandboxSession:
+        workspace = await self.get_workspace_or_404(
+            organization_id=organization_id, workspace_id=workspace_id
+        )
+        if workspace.status == WorkspaceStatus.DESTROYED:
+            raise ConflictError("این محیط اجرا از بین رفته است.")
+        if not workspace.external_workspace_id:
+            raise ProviderUnavailableError("محیط اجرا در ارائه‌دهنده ثبت نشده است.")
+
+        info = await self.provider.start_session(
+            workspace_external_id=workspace.external_workspace_id
+        )
+        session = await self.sessions.create(
+            SandboxSession(
+                workspace_id=workspace.id,
+                external_session_id=info.external_id,
+                status=SessionStatus.RUNNING,
+            )
+        )
+        await self.workspaces.set_status(workspace.id, WorkspaceStatus.READY)
+        return session
+
+    async def get_session_or_404(
+        self, *, organization_id: str, session_id: str
+    ) -> SandboxSession:
+        session = await self.sessions.get_for_org(organization_id, session_id)
+        if session is None:
+            raise NotFoundError("نشست اجرا یافت نشد.")
+        return session
+
+    async def stop_session(self, *, organization_id: str, session_id: str) -> SandboxSession:
+        session = await self.get_session_or_404(
+            organization_id=organization_id, session_id=session_id
+        )
+        if not session.external_session_id:
+            raise ProviderUnavailableError("نشست در ارائه‌دهنده ثبت نشده است.")
+        await self.provider.destroy(session_external_id=session.external_session_id)
+        await self.sessions.set_status(session.id, SessionStatus.STOPPED, stopped=True)
+        refreshed = await self.sessions.get(session.id)
+        return refreshed or session
+
+    # -- commands ------------------------------------------------------------
+
+    async def run_command(
+        self,
+        *,
+        organization_id: str,
+        session_id: str,
+        command: str,
+        user_id: str | None = None,
+    ) -> tuple[CommandRun, CommandResult]:
+        session = await self.get_session_or_404(
+            organization_id=organization_id, session_id=session_id
+        )
+        if not session.external_session_id:
+            raise ProviderUnavailableError("نشست در ارائه‌دهنده ثبت نشده است.")
+        if session.status != SessionStatus.RUNNING:
+            raise ConflictError("نشست در وضعیت اجرا نیست.")
+
+        cleaned = command.strip()
+        if not cleaned:
+            raise ValidationError("دستور نمی‌تواند خالی باشد.")
+        if len(cleaned) > MAX_COMMAND_LENGTH:
+            raise ValidationError("دستور بیش از حد طولانی است.")
+
+        workspace = await self.workspaces.get_for_org(
+            organization_id, session.workspace_id
+        )
+        limits = await self.effective_limits(
+            organization_id=organization_id,
+            project_id=workspace.project_id if workspace else None,
+        )
+
+        result = await self.provider.run_command(
+            session_external_id=session.external_session_id,
+            command=cleaned,
+            # Timeout and output cap are decided here, not by the caller.
+            timeout_seconds=limits.timeout_seconds,
+            max_output_bytes=limits.max_output_bytes,
+            env={},
+        )
+
+        run = await self.commands.create(
+            CommandRun(
+                session_id=session.id,
+                command=cleaned,
+                exit_code=result.exit_code,
+                stdout_ref=result.stdout[: limits.max_output_bytes],
+                stderr_ref=result.stderr[: limits.max_output_bytes],
+                duration_ms=result.duration_ms,
+                truncated=result.truncated,
+            )
+        )
+        await self.sessions.set_status(session.id, SessionStatus.RUNNING)
+
+        # CPU seconds are billed. 1 second of wall time ~= 1 CPU second at MVP.
+        cpu_seconds = max(0.0, result.duration_ms / 1000.0)
+        if cpu_seconds > 0:
+            card = await self.rate_cards.get_by_unit(RateUnit.SANDBOX_CPU_SECOND)
+            unit_price = card.customer_price if card else 0.0
+            internal_cost = card.internal_cost_estimate if card else 0.0
+            await self.ledger.record(
+                usage=UsageRecord(
+                    organization_id=organization_id,
+                    project_id=workspace.project_id if workspace else None,
+                    user_id=user_id,
+                    unit=RateUnit.SANDBOX_CPU_SECOND,
+                    quantity=cpu_seconds,
+                    unit_price=unit_price,
+                    internal_cost_estimate=cpu_seconds * internal_cost,
+                    total_price=cpu_seconds * unit_price,
+                    metadata_json={"session_id": session.id, "exit_code": result.exit_code},
+                )
+            )
+
+        return run, result
+
+    async def list_commands(self, *, session_id: str) -> list[CommandRun]:
+        return await self.commands.list_for_session(session_id)
+
+    # -- files ---------------------------------------------------------------
+
+    async def read_file(
+        self, *, organization_id: str, session_id: str, path: str
+    ) -> str:
+        session = await self.get_session_or_404(
+            organization_id=organization_id, session_id=session_id
+        )
+        if not session.external_session_id:
+            raise ProviderUnavailableError("نشست در ارائه‌دهنده ثبت نشده است.")
+        return await self.provider.read_file(
+            session_external_id=session.external_session_id, path=path
+        )
+
+    async def write_file(
+        self, *, organization_id: str, session_id: str, path: str, content: str
+    ) -> None:
+        session = await self.get_session_or_404(
+            organization_id=organization_id, session_id=session_id
+        )
+        if not session.external_session_id:
+            raise ProviderUnavailableError("نشست در ارائه‌دهنده ثبت نشده است.")
+        await self.provider.write_file(
+            session_external_id=session.external_session_id, path=path, content=content
+        )
+
+    # -- snapshots & previews ------------------------------------------------
+
+    async def snapshot(
+        self, *, organization_id: str, session_id: str, name: str
+    ) -> Snapshot:
+        session = await self.get_session_or_404(
+            organization_id=organization_id, session_id=session_id
+        )
+        if not session.external_session_id:
+            raise ProviderUnavailableError("نشست در ارائه‌دهنده ثبت نشده است.")
+        info = await self.provider.snapshot(
+            session_external_id=session.external_session_id, name=name
+        )
+        return await self.snapshots.create(
+            Snapshot(
+                workspace_id=session.workspace_id, name=name,
+                external_snapshot_id=info.external_id,
+            )
+        )
+
+    async def get_preview_url(
+        self, *, organization_id: str, session_id: str, port: int
+    ) -> PreviewUrl:
+        session = await self.get_session_or_404(
+            organization_id=organization_id, session_id=session_id
+        )
+        if not session.external_session_id:
+            raise ProviderUnavailableError("نشست در ارائه‌دهنده ثبت نشده است.")
+        url = await self.provider.get_preview_url(
+            session_external_id=session.external_session_id, port=port
+        )
+        return await self.previews.create(
+            PreviewUrl(workspace_id=session.workspace_id, port=port, url=url)
+        )
+
+    # -- policy administration ----------------------------------------------
+
+    async def set_limits(
+        self,
+        *,
+        scope: str,
+        scope_id: str | None,
+        cpu: float,
+        memory_mb: int,
+        disk_mb: int,
+        timeout_seconds: int,
+        max_output_bytes: int,
+        network_policy: NetworkPolicy,
+    ) -> SandboxResourceLimit:
+        if scope not in ("organization", "project", "platform"):
+            raise ValidationError("دامنهٔ محدودیت نامعتبر است.")
+        return await self.limits.upsert(
+            self._clamp(
+                SandboxResourceLimit(
+                    scope=scope, scope_id=scope_id, cpu=cpu, memory_mb=memory_mb,
+                    disk_mb=disk_mb, timeout_seconds=timeout_seconds,
+                    max_output_bytes=max_output_bytes, network_policy=network_policy,
+                )
+            )
+        )
+
+    async def destroy_workspace(
+        self, *, organization_id: str, workspace_id: str
+    ) -> Workspace:
+        workspace = await self.get_workspace_or_404(
+            organization_id=organization_id, workspace_id=workspace_id
+        )
+        for session in await self.sessions.list_for_workspace(workspace.id):
+            if session.external_session_id:
+                # Teardown is best-effort: a remote sandbox that is already gone
+                # must not block us from marking the local row stopped.
+                with contextlib.suppress(Exception):
+                    await self.provider.destroy(
+                        session_external_id=session.external_session_id
+                    )
+                await self.sessions.set_status(
+                    session.id, SessionStatus.STOPPED, stopped=True
+                )
+        await self.workspaces.set_status(workspace.id, WorkspaceStatus.DESTROYED)
+        refreshed = await self.workspaces.get_for_org(organization_id, workspace_id)
+        return refreshed or workspace
+
+    @staticmethod
+    def now():
+        return utcnow()
+
+    # -- platform maintenance ------------------------------------------------
+
+    async def reap_idle_sessions(
+        self, *, idle_before: datetime, limit: int = 200
+    ) -> list[SandboxSession]:
+        """Stop sandbox sessions untouched since ``idle_before``.
+
+        **Deliberately not tenant-scoped.** This is the gateway's own housekeeping
+        and spans every organization, so it is never reachable from a
+        tenant-scoped route — only from the sandbox-gateway worker. Callers are
+        responsible for the audit trail.
+
+        A provider that has already forgotten the session must not keep it
+        "active" in our own record forever, so the remote teardown is
+        best-effort and the local row is always marked stopped.
+        """
+        stopped: list[SandboxSession] = []
+        stale = await self.sessions.list_stale_active(
+            idle_before=idle_before, limit=limit
+        )
+        for session in stale:
+            if session.external_session_id:
+                with contextlib.suppress(Exception):
+                    await self.provider.destroy(
+                        session_external_id=session.external_session_id
+                    )
+            await self.sessions.set_status(session.id, SessionStatus.STOPPED, stopped=True)
+            refreshed = await self.sessions.get(session.id)
+            stopped.append(refreshed or session)
+        return stopped

apps/api/app/application/services/workforce_service.py

@@ -0,0 +1,794 @@
+"""AI Workforce use cases: employees, Manager Chat, tasks and the run engine.
+
+This module is the application layer. It owns three invariants that the rest of
+the system depends on:
+
+1. **A tenant never sees model identity.** The service resolves a routing policy
+   to a concrete ``AiModel`` internally and passes it to the provider port. No
+   router may serialise that name back to a customer.
+2. **Every model call writes a usage event.** Billing reads the internal ledger,
+   never the provider's invoice.
+3. **Every run leaves a timeline.** ``run_steps`` is the observable record of
+   what happened, and it is what the UI renders.
+
+The MVP "engine" is deliberately synchronous and deterministic: it drives a run
+through its real state machine (``RUN_TRANSITIONS``) and records each transition
+as a step. Swapping in a durable worker only changes *where* the transitions are
+executed, not the transitions themselves.
+"""
+
+from __future__ import annotations
+
+from app.application.ports import (
+    ChatMessage,
+    ModelProviderPort,
+    UsageRecord,
+)
+from app.core.errors import (
+    ConflictError,
+    NotFoundError,
+    PermissionDeniedError,
+    ProviderUnavailableError,
+    ValidationError,
+)
+from app.domain.ai_control.models import RateUnit, RoutingMode
+from app.domain.workforce.models import (
+    TERMINAL_RUN_STATUSES,
+    AgentTemplate,
+    Artifact,
+    Employee,
+    EmployeeStatus,
+    EmployeeToolPermission,
+    Message,
+    MessageRole,
+    PromptVersion,
+    RoleKey,
+    Run,
+    RunStatus,
+    RunStep,
+    Task,
+    TaskStatus,
+    Thread,
+    ToolCall,
+    ToolCallStatus,
+    can_transition_run,
+)
+from app.domain.workforce.tools import (
+    TOOL_CATALOG,
+    ToolSpec,
+    is_known_tool,
+    tool_requires_approval,
+)
+from app.infrastructure.repositories.ai_control import (
+    AiModelRepository,
+    ModelCredentialRepository,
+    RateCardRepository,
+    RoutingPolicyRepository,
+)
+from app.infrastructure.repositories.audit import UsageLedgerRepository
+from app.infrastructure.repositories.workforce import (
+    AgentTemplateRepository,
+    ArtifactRepository,
+    EmployeeRepository,
+    EmployeeToolPermissionRepository,
+    MessageRepository,
+    PromptVersionRepository,
+    RunRepository,
+    RunStepRepository,
+    TaskRepository,
+    ThreadRepository,
+    ToolCallRepository,
+)
+
+#: Fallback price when the platform owner has not configured a rate card yet.
+#: Charging zero silently would hide a misconfiguration, so we use a tiny
+#: non-zero placeholder and mark it in the usage metadata.
+_UNPRICED = 0.0000001
+
+_SYSTEM_PROMPT = (
+    "تو دستیار هوش مصنوعی سازمان هستی. پاسخ‌ها را کوتاه، دقیق و عملی بده. "
+    "هیچ‌گاه نام مدل، ارائه‌دهنده یا جزئیات زیرساخت را فاش نکن."
+)
+
+
+class WorkforceService:
+    def __init__(
+        self,
+        *,
+        employees: EmployeeRepository,
+        templates: AgentTemplateRepository,
+        prompts: PromptVersionRepository,
+        permissions: EmployeeToolPermissionRepository,
+        threads: ThreadRepository,
+        messages: MessageRepository,
+        tasks: TaskRepository,
+        runs: RunRepository,
+        run_steps: RunStepRepository,
+        tool_calls: ToolCallRepository,
+        artifacts: ArtifactRepository,
+        models: AiModelRepository,
+        credentials: ModelCredentialRepository,
+        routing_policies: RoutingPolicyRepository,
+        rate_cards: RateCardRepository,
+        ledger: UsageLedgerRepository,
+        provider: ModelProviderPort,
+    ) -> None:
+        self.employees = employees
+        self.templates = templates
+        self.prompts = prompts
+        self.permissions = permissions
+        self.threads = threads
+        self.messages = messages
+        self.tasks = tasks
+        self.runs = runs
+        self.run_steps = run_steps
+        self.tool_calls = tool_calls
+        self.artifacts = artifacts
+        self.models = models
+        self.credentials = credentials
+        self.routing_policies = routing_policies
+        self.rate_cards = rate_cards
+        self.ledger = ledger
+        self.provider = provider
+
+    # ------------------------------------------------------------------ #
+    # Templates                                                           #
+    # ------------------------------------------------------------------ #
+
+    async def list_templates(self) -> list[AgentTemplate]:
+        return await self.templates.list()
+
+    async def list_routing_choices(self) -> list:
+        """Active routing policies, for customer-facing selection.
+
+        Callers must render only ``name``/``description`` — ``mode`` is a routing
+        hint that should not reach a tenant.
+        """
+        return [p for p in await self.routing_policies.list() if p.is_active]
+
+    # ------------------------------------------------------------------ #
+    # Employees                                                           #
+    # ------------------------------------------------------------------ #
+
+    async def create_employee(
+        self,
+        *,
+        organization_id: str,
+        created_by_user_id: str,
+        name: str,
+        role_key: RoleKey,
+        template_id: str | None = None,
+        project_id: str | None = None,
+        routing_policy_id: str | None = None,
+    ) -> Employee:
+        template: AgentTemplate | None = None
+        if template_id:
+            template = await self.templates.get(template_id)
+            if template is None:
+                raise NotFoundError("قالب کارمند یافت نشد.")
+
+        employee = await self.employees.create(
+            Employee(
+                organization_id=organization_id,
+                name=name.strip(),
+                role_key=role_key,
+                template_id=template_id,
+                project_id=project_id,
+                routing_policy_id=routing_policy_id
+                or (template.default_routing_policy_id if template else None),
+            )
+        )
+
+        # Seed version 1 of the prompt from the template, so the employee is
+        # usable immediately and every later edit is a tracked revision.
+        if template is not None and template.default_prompt:
+            await self.prompts.create(
+                PromptVersion(
+                    employee_id=employee.id,
+                    version=1,
+                    content=template.default_prompt,
+                    created_by_user_id=created_by_user_id,
+                )
+            )
+
+        if template is not None:
+            for tool_key in template.default_tools_json:
+                await self.permissions.set_permission(
+                    employee_id=employee.id, tool_key=tool_key, allowed=True
+                )
+
+        return employee
+
+    async def list_employees(
+        self, organization_id: str, *, limit: int = 100, offset: int = 0
+    ) -> list[Employee]:
+        return await self.employees.list_for_org(organization_id, limit=limit, offset=offset)
+
+    async def get_employee_or_404(
+        self, *, organization_id: str, employee_id: str
+    ) -> Employee:
+        employee = await self.employees.get_for_org(organization_id, employee_id)
+        # 404 rather than 403: another tenant's id must be indistinguishable
+        # from a non-existent one.
+        if employee is None:
+            raise NotFoundError("کارمند یافت نشد.")
+        return employee
+
+    async def update_employee(
+        self,
+        *,
+        organization_id: str,
+        employee_id: str,
+        name: str | None = None,
+        status: EmployeeStatus | None = None,
+        project_id: str | None = None,
+        routing_policy_id: str | None = None,
+    ) -> Employee:
+        employee = await self.get_employee_or_404(
+            organization_id=organization_id, employee_id=employee_id
+        )
+        if name is not None:
+            employee.name = name.strip()
+        if status is not None:
+            employee.status = status
+        if project_id is not None:
+            employee.project_id = project_id or None
+        if routing_policy_id is not None:
+            employee.routing_policy_id = routing_policy_id or None
+        return await self.employees.update(employee)
+
+    async def list_prompt_versions(self, employee_id: str) -> list[PromptVersion]:
+        return await self.prompts.list_for_employee(employee_id)
+
+    async def add_prompt_version(
+        self, *, employee_id: str, content: str, created_by_user_id: str
+    ) -> PromptVersion:
+        if not content.strip():
+            raise ValidationError("متن پرامپت نمی‌تواند خالی باشد.")
+        return await self.prompts.create(
+            PromptVersion(
+                employee_id=employee_id,
+                version=await self.prompts.next_version_number(employee_id),
+                content=content,
+                created_by_user_id=created_by_user_id,
+            )
+        )
+
+    async def list_tool_permissions(self, employee_id: str) -> list[EmployeeToolPermission]:
+        return await self.permissions.list_for_employee(employee_id)
+
+    # ------------------------------------------------------------------ #
+    # Threads & messages                                                  #
+    # ------------------------------------------------------------------ #
+
+    async def create_thread(
+        self,
+        *,
+        organization_id: str,
+        created_by_user_id: str,
+        title: str = "",
+        project_id: str | None = None,
+    ) -> Thread:
+        return await self.threads.create(
+            Thread(
+                organization_id=organization_id,
+                title=title.strip(),
+                project_id=project_id,
+                created_by_user_id=created_by_user_id,
+            )
+        )
+
+    async def list_threads(
+        self, organization_id: str, *, limit: int = 100, offset: int = 0
+    ) -> list[Thread]:
+        return await self.threads.list_for_org(organization_id, limit=limit, offset=offset)
+
+    async def get_thread_or_404(self, *, organization_id: str, thread_id: str) -> Thread:
+        thread = await self.threads.get_for_org(organization_id, thread_id)
+        if thread is None:
+            raise NotFoundError("گفتگو یافت نشد.")
+        return thread
+
+    async def list_messages(self, thread_id: str) -> list[Message]:
+        return await self.messages.list_for_thread(thread_id)
+
+    async def post_user_message(
+        self,
+        *,
+        organization_id: str,
+        thread_id: str,
+        content: str,
+        user_id: str,
+        employee_id: str | None = None,
+    ) -> tuple[Message, Message]:
+        """Append the user's message, then produce the assistant reply.
+
+        Returns ``(user_message, assistant_message)`` so the caller can render
+        both without a second round-trip.
+        """
+        thread = await self.get_thread_or_404(
+            organization_id=organization_id, thread_id=thread_id
+        )
+        if not content.strip():
+            raise ValidationError("متن پیام نمی‌تواند خالی باشد.")
+
+        user_message = await self.messages.create(
+            Message(thread_id=thread.id, role=MessageRole.USER, content=content)
+        )
+
+        employee: Employee | None = None
+        if employee_id:
+            employee = await self.get_employee_or_404(
+                organization_id=organization_id, employee_id=employee_id
+            )
+        else:
+            employee = await self._default_manager(organization_id)
+
+        history = await self.messages.list_for_thread(thread.id)
+        system_prompt = await self._system_prompt_for(employee)
+        reply = await self._complete(
+            organization_id=organization_id,
+            employee=employee,
+            project_id=thread.project_id,
+            user_id=user_id,
+            system=system_prompt,
+            messages=[
+                ChatMessage(role=m.role.value, content=m.content)
+                for m in history
+                if m.role in (MessageRole.USER, MessageRole.ASSISTANT)
+            ],
+        )
+
+        assistant_message = await self.messages.create(
+            Message(
+                thread_id=thread.id,
+                role=MessageRole.ASSISTANT,
+                content=reply,
+                employee_id=employee.id if employee else None,
+            )
+        )
+        return user_message, assistant_message
+
+    # ------------------------------------------------------------------ #
+    # Tasks                                                               #
+    # ------------------------------------------------------------------ #
+
+    async def create_task(
+        self,
+        *,
+        organization_id: str,
+        created_by_user_id: str,
+        title: str,
+        description: str = "",
+        project_id: str | None = None,
+        thread_id: str | None = None,
+        assigned_employee_id: str | None = None,
+    ) -> Task:
+        if not title.strip():
+            raise ValidationError("عنوان کار نمی‌تواند خالی باشد.")
+        if assigned_employee_id:
+            # Validates tenant ownership before we ever store the reference.
+            await self.get_employee_or_404(
+                organization_id=organization_id, employee_id=assigned_employee_id
+            )
+        return await self.tasks.create(
+            Task(
+                organization_id=organization_id,
+                title=title.strip(),
+                description=description,
+                project_id=project_id,
+                thread_id=thread_id,
+                assigned_employee_id=assigned_employee_id,
+                created_by_user_id=created_by_user_id,
+            )
+        )
+
+    async def list_tasks(
+        self,
+        organization_id: str,
+        *,
+        status: TaskStatus | None = None,
+        limit: int = 100,
+        offset: int = 0,
+    ) -> list[Task]:
+        return await self.tasks.list_for_org(
+            organization_id, status=status, limit=limit, offset=offset
+        )
+
+    async def get_task_or_404(self, *, organization_id: str, task_id: str) -> Task:
+        task = await self.tasks.get_for_org(organization_id, task_id)
+        if task is None:
+            raise NotFoundError("کار یافت نشد.")
+        return task
+
+    async def list_task_runs(self, task_id: str) -> list[Run]:
+        return await self.runs.list_for_task(task_id)
+
+    # ------------------------------------------------------------------ #
+    # Runs                                                                #
+    # ------------------------------------------------------------------ #
+
+    async def start_run(
+        self,
+        *,
+        organization_id: str,
+        task_id: str,
+        user_id: str,
+        employee_id: str | None = None,
+    ) -> Run:
+        task = await self.get_task_or_404(organization_id=organization_id, task_id=task_id)
+        if task.status in (TaskStatus.DONE, TaskStatus.CANCELLED):
+            raise ConflictError("این کار قبلاً بسته شده است.")
+
+        employee: Employee | None = None
+        if employee_id or task.assigned_employee_id:
+            employee = await self.get_employee_or_404(
+                organization_id=organization_id,
+                employee_id=employee_id or task.assigned_employee_id or "",
+            )
+            if not employee.is_available:
+                raise ConflictError("این کارمند در حالت فعال نیست.")
+        else:
+            employee = await self._default_manager(organization_id)
+
+        run = await self.runs.create(
+            Run(task_id=task.id, employee_id=employee.id if employee else None)
+        )
+        await self.tasks.set_status(task.id, TaskStatus.RUNNING)
+
+        await self.run_steps.append(
+            run_id=run.id, kind="queued", title="کار در صف اجرا قرار گرفت.",
+            payload={"task_id": task.id},
+        )
+
+        try:
+            run = await self._advance(run, RunStatus.PLANNING)
+            await self.run_steps.append(
+                run_id=run.id, kind="planning", title="برنامه‌ریزی مرحلهٔ اجرا."
+            )
+
+            run = await self._advance(run, RunStatus.RUNNING)
+            await self.run_steps.append(
+                run_id=run.id, kind="running", title="اجرای کار آغاز شد."
+            )
+
+            reply = await self._complete(
+                organization_id=organization_id,
+                employee=employee,
+                project_id=task.project_id,
+                user_id=user_id,
+                run_id=run.id,
+                system=await self._system_prompt_for(employee),
+                messages=[
+                    ChatMessage(role="user", content=self._task_brief(task)),
+                ],
+            )
+            await self.run_steps.append(
+                run_id=run.id, kind="output", title="خروجی آماده شد.",
+                payload={"chars": len(reply)},
+            )
+
+            artifact = await self.artifacts.create(
+                Artifact(
+                    organization_id=organization_id,
+                    project_id=task.project_id,
+                    run_id=run.id,
+                    name=f"خروجی {task.title}"[:200],
+                    kind="text",
+                    content_ref=reply[:8000],
+                    size_bytes=len(reply.encode("utf-8")),
+                )
+            )
+            await self.run_steps.append(
+                run_id=run.id, kind="artifact", title="سند خروجی ساخته شد.",
+                payload={"artifact_id": artifact.id},
+            )
+
+            run = await self._finish(run, RunStatus.COMPLETED)
+            await self.tasks.set_status(task.id, TaskStatus.DONE)
+        except ProviderUnavailableError as exc:
+            run = await self._finish(run, RunStatus.FAILED, error=str(exc))
+            await self.tasks.set_status(task.id, TaskStatus.FAILED)
+            await self.run_steps.append(
+                run_id=run.id, kind="error", title="اجرا ناموفق بود.",
+                payload={"error": str(exc)},
+            )
+        except Exception as exc:  # noqa: BLE001 - the run must always terminate
+            run = await self._finish(run, RunStatus.FAILED, error=str(exc))
+            await self.tasks.set_status(task.id, TaskStatus.FAILED)
+            await self.run_steps.append(
+                run_id=run.id, kind="error", title="خطای غیرمنتظره در اجرا.",
+                payload={"error": str(exc)},
+            )
+
+        return run
+
+    async def get_run_or_404(
+        self, *, organization_id: str, run_id: str
+    ) -> tuple[Run, list[RunStep], list[ToolCall], list[Artifact]]:
+        run = await self.runs.get(run_id)
+        if run is None:
+            raise NotFoundError("اجرا یافت نشد.")
+        task = await self.tasks.get(run.task_id)
+        # Re-scope through the task, which is the object that carries the tenant.
+        if task is None or task.organization_id != organization_id:
+            raise NotFoundError("اجرا یافت نشد.")
+        return (
+            run,
+            await self.run_steps.list_for_run(run.id),
+            await self.tool_calls.list_for_run(run.id),
+            await self.artifacts.list_for_run(run.id),
+        )
+
+    async def cancel_run(self, *, organization_id: str, run_id: str) -> Run:
+        run, _, _, _ = await self.get_run_or_404(
+            organization_id=organization_id, run_id=run_id
+        )
+        if run.status in TERMINAL_RUN_STATUSES:
+            raise ConflictError("این اجرا در وضعیت پایانی است و لغو نمی‌شود.")
+        if not can_transition_run(run.status, RunStatus.CANCELLED):
+            raise ConflictError("از وضعیت فعلی امکان لغو وجود ندارد.")
+
+        run = await self._finish(run, RunStatus.CANCELLED)
+        await self.run_steps.append(
+            run_id=run.id, kind="cancelled", title="اجرا توسط کاربر لغو شد."
+        )
+        task = await self.tasks.get(run.task_id)
+        if task is not None and task.status not in (TaskStatus.DONE, TaskStatus.CANCELLED):
+            await self.tasks.set_status(task.id, TaskStatus.CANCELLED)
+        return run
+
+    async def list_pending_approvals(self, organization_id: str) -> list[ToolCall]:
+        return await self.tool_calls.list_pending_approvals_for_org(organization_id)
+
+    async def list_artifacts(self, organization_id: str) -> list[Artifact]:
+        return await self.artifacts.list_for_org(organization_id)
+
+    # ------------------------------------------------------------------ #
+    # Tools & approvals                                                   #
+    # ------------------------------------------------------------------ #
+
+    @staticmethod
+    def tool_catalog() -> tuple[ToolSpec, ...]:
+        return TOOL_CATALOG
+
+    async def set_tool_permission(
+        self,
+        *,
+        organization_id: str,
+        employee_id: str,
+        tool_key: str,
+        allowed: bool,
+    ) -> EmployeeToolPermission:
+        """Grant or revoke one tool for one employee.
+
+        An unknown tool key is rejected rather than stored: a permission row for
+        a tool that does not exist is a silent no-op that hides a typo.
+        """
+        await self.get_employee_or_404(
+            organization_id=organization_id, employee_id=employee_id
+        )
+        if not is_known_tool(tool_key):
+            raise ValidationError(f"ابزار ناشناخته: {tool_key}")
+        return await self.permissions.set_permission(
+            employee_id=employee_id, tool_key=tool_key, allowed=allowed
+        )
+
+    async def request_tool_call(
+        self,
+        *,
+        organization_id: str,
+        run_id: str,
+        employee_id: str | None,
+        tool_key: str,
+        arguments: dict | None = None,
+    ) -> ToolCall:
+        """Record an intended tool call, gating it on permission and approval.
+
+        The call is persisted in every case so the audit trail shows what was
+        attempted, not only what succeeded.
+        """
+        run, _, _, _ = await self.get_run_or_404(
+            organization_id=organization_id, run_id=run_id
+        )
+
+        if not is_known_tool(tool_key):
+            raise ValidationError(f"ابزار ناشناخته: {tool_key}")
+
+        # Deny by default. An employee with no permission row cannot call a tool.
+        if employee_id is None or not await self.permissions.is_allowed(
+            employee_id, tool_key
+        ):
+            raise PermissionDeniedError(
+                f"کارمند مجاز به استفاده از «{tool_key}» نیست."
+            )
+
+        needs_approval = tool_requires_approval(tool_key)
+        call = await self.tool_calls.create(
+            ToolCall(
+                run_id=run.id,
+                tool_key=tool_key,
+                arguments_json=dict(arguments or {}),
+                status=(
+                    ToolCallStatus.AWAITING_APPROVAL
+                    if needs_approval
+                    else ToolCallStatus.PENDING
+                ),
+                requires_approval=needs_approval,
+            )
+        )
+        await self.run_steps.append(
+            run_id=run.id,
+            kind="tool_call",
+            title=(
+                f"درخواست تأیید برای «{tool_key}»"
+                if needs_approval
+                else f"فراخوانی «{tool_key}»"
+            ),
+            payload={"tool_call_id": call.id, "requires_approval": needs_approval},
+        )
+        return call
+
+    async def decide_approval(
+        self,
+        *,
+        organization_id: str,
+        tool_call_id: str,
+        approve: bool,
+        decided_by_user_id: str,
+        note: str = "",
+    ) -> ToolCall:
+        """Approve or reject a pending tool call. Only a human may call this."""
+        call = await self.tool_calls.get_for_org(organization_id, tool_call_id)
+        if call is None:
+            raise NotFoundError("درخواست تأیید یافت نشد.")
+        if call.status != ToolCallStatus.AWAITING_APPROVAL:
+            raise ConflictError("این درخواست در وضعیت انتظار تأیید نیست.")
+
+        result = dict(call.result_json)
+        if note:
+            result["note"] = note
+
+        updated = await self.tool_calls.set_status(
+            call.id,
+            ToolCallStatus.APPROVED if approve else ToolCallStatus.REJECTED,
+            result=result,
+            approved_by_user_id=decided_by_user_id,
+        )
+        await self.run_steps.append(
+            run_id=call.run_id,
+            kind="approval",
+            title=("درخواست تأیید شد." if approve else "درخواست رد شد."),
+            payload={"tool_call_id": call.id, "tool_key": call.tool_key},
+        )
+        return updated or call
+
+    # ------------------------------------------------------------------ #
+    # Internals                                                           #
+    # ------------------------------------------------------------------ #
+
+    async def _default_manager(self, organization_id: str) -> Employee | None:
+        """The employee that answers when the caller did not pick one."""
+        for employee in await self.employees.list_for_org(organization_id):
+            if employee.role_key == RoleKey.MANAGER and employee.is_available:
+                return employee
+        return None
+
+    async def _system_prompt_for(self, employee: Employee | None) -> str:
+        if employee is None:
+            return _SYSTEM_PROMPT
+        latest = await self.prompts.latest(employee.id)
+        if latest is None:
+            return _SYSTEM_PROMPT
+        return f"{_SYSTEM_PROMPT}\n\n{latest.content}"
+
+    async def _resolve_model(self, employee: Employee | None):
+        """Map the employee's routing policy to a concrete model. Never exposed."""
+        mode = RoutingMode.BALANCED
+        policy_id = employee.routing_policy_id if employee else None
+        policy = (
+            await self.routing_policies.get(policy_id)
+            if policy_id
+            else await self.routing_policies.get_default()
+        )
+        if policy is not None:
+            mode = policy.mode
+
+        model = await self.models.pick_for_mode(mode)
+        if model is None:
+            raise ProviderUnavailableError(
+                "هیچ مدل فعالی برای مسیریابی پیکربندی نشده است."
+            )
+        return model
+
+    async def _complete(
+        self,
+        *,
+        organization_id: str,
+        employee: Employee | None,
+        project_id: str | None,
+        user_id: str | None,
+        system: str,
+        messages: list[ChatMessage],
+        run_id: str | None = None,
+    ) -> str:
+        """Run one completion and record its usage. The single model entry point."""
+        model = await self._resolve_model(employee)
+        credential = await self.credentials.get_active_for_provider(model.provider_id)
+
+        response = await self.provider.complete(
+            model=model, credential=credential, messages=messages, system=system
+        )
+
+        await self._record_usage(
+            organization_id=organization_id,
+            project_id=project_id,
+            user_id=user_id,
+            employee_id=employee.id if employee else None,
+            run_id=run_id,
+            unit=RateUnit.MODEL_INPUT_TOKENS,
+            quantity=float(response.input_tokens),
+        )
+        await self._record_usage(
+            organization_id=organization_id,
+            project_id=project_id,
+            user_id=user_id,
+            employee_id=employee.id if employee else None,
+            run_id=run_id,
+            unit=RateUnit.MODEL_OUTPUT_TOKENS,
+            quantity=float(response.output_tokens),
+        )
+        return response.content
+
+    async def _record_usage(
+        self,
+        *,
+        organization_id: str,
+        unit: RateUnit,
+        quantity: float,
+        project_id: str | None = None,
+        user_id: str | None = None,
+        employee_id: str | None = None,
+        run_id: str | None = None,
+    ) -> None:
+        card = await self.rate_cards.get_by_unit(unit)
+        unit_price = card.customer_price if card else _UNPRICED
+        internal_cost = card.internal_cost_estimate if card else 0.0
+        await self.ledger.record(
+            usage=UsageRecord(
+                organization_id=organization_id,
+                project_id=project_id,
+                user_id=user_id,
+                employee_id=employee_id,
+                run_id=run_id,
+                unit=unit,
+                quantity=quantity,
+                unit_price=unit_price,
+                internal_cost_estimate=quantity * internal_cost,
+                total_price=quantity * unit_price,
+                metadata_json={"priced": card is not None},
+            )
+        )
+
+    async def _advance(self, run: Run, target: RunStatus) -> Run:
+        if not can_transition_run(run.status, target):
+            raise ConflictError(f"گذار نامعتبر از {run.status} به {target}.")
+        return await self.runs.set_status(run.id, target) or run
+
+    async def _finish(self, run: Run, target: RunStatus, *, error: str | None = None) -> Run:
+        from app.domain.base import utcnow
+
+        if not can_transition_run(run.status, target):
+            # Already terminal — return the current row untouched.
+            return await self.runs.get(run.id) or run
+        updated = await self.runs.set_status(
+            run.id, target, error=error, finished_at=utcnow()
+        )
+        return updated or run
+
+    @staticmethod
+    def _task_brief(task: Task) -> str:
+        parts = [f"عنوان کار: {task.title}"]
+        if task.description:
+            parts.append(f"توضیحات: {task.description}")
+        return "\n".join(parts)