feat(integration): prepaid billing, engine UI, credits console, direct tasks, v2 final

- Prepaid credit system: org wallet, debit on usage, chat gate, admin credit console
- Engine UI: /manager, /tasks, /tasks/[id], /workforce, /dashboard, /traces pages
- Direct task creation (HTTP POST /tasks, MCP tasks_create) at pending_manager_review
- Pause/resume/cancel/answer actions in engine API and UI
- Employee cap (7 per org default), review pipeline guard (active reviewer required)
- Scheduler: wide window prevents org starvation; archived employees excluded from cap
- Tenant isolation: project-id cross-org 404, engine never uses another tenant's sandbox
- Order route fix: manager built with pipeline states + employee profiles
- Contract extension: ModelGrantPort.get_org_policy with DefaultOrgPolicy
- Engine event types in @agy/contracts (task.step, task.gate, run.log, run.tool, team.changed)
- api-engine.ts typed client; api-credit.ts client
- Alembic migration c4d5e6f7a8b9: prepaid credit + engine UI tables
- 859 tests passed, 1 skipped; ruff/mypy/tsc/eslint/vitest all green

Refs: PLAN-v2.md §9 (T1–T43), todo-v2-final.md

apps/api/app/application/container_engine.py

@@ -209,6 +209,7 @@ def _build_manager(
         RateCardRepository,
     )
     from app.infrastructure.repositories.audit import UsageLedgerRepository
+    from app.infrastructure.repositories.billing import CreditWalletRepository
     from app.infrastructure.repositories.engine import (
         EmployeeProfileRepository,
         PipelineStateRepository,
@@ -235,6 +236,7 @@ def _build_manager(
         profiles=EmployeeProfileRepository(session),
         states=PipelineStateRepository(session),
         runner=runner,
+        wallets=CreditWalletRepository(session),
     )
 
 

apps/api/app/application/engine/events.py

@@ -28,6 +28,7 @@ from app.application.ports.realtime import (
     task_channel,
 )
 from app.core.config import settings
+from app.domain.workforce.pipeline import PipelineKey, Stage, coarse_status
 
 
 @dataclass(slots=True, kw_only=True)
@@ -204,4 +205,33 @@ async def emit_team_changed(
     )
 
 
-__all__ = ["RunFeed", "emit_team_changed"]
+async def emit_task_step(
+    *,
+    task_id: str,
+    organization_id: str,
+    from_stage: Stage,
+    to_stage: Stage,
+    pipeline: PipelineKey,
+) -> None:
+    """A task moved one stage: the task page's stepper follows it live.
+
+    The payload carries stage names and the coarse status only, never a model
+    or an employee name.
+    """
+    await publish(
+        RealtimeEvent(
+            type="task.step",
+            channels=(task_channel(task_id), org_channel(organization_id)),
+            organization_id=organization_id,
+            data={
+                "task_id": task_id,
+                "from": str(from_stage),
+                "to": str(to_stage),
+                "pipeline": str(pipeline),
+                "status": coarse_status(to_stage).value,
+            },
+        )
+    )
+
+
+__all__ = ["RunFeed", "emit_task_step", "emit_team_changed"]

apps/api/app/application/engine/manager.py

@@ -88,6 +88,7 @@ from app.infrastructure.repositories.ai_control import (
     RateCardRepository,
 )
 from app.infrastructure.repositories.audit import UsageLedgerRepository
+from app.infrastructure.repositories.billing import CreditWalletRepository
 from app.infrastructure.repositories.engine import (
     EmployeeProfileRepository,
     PipelineStateRepository,
@@ -109,6 +110,11 @@ CHAT_MAX_ATTEMPTS = 2
 #: The message a new organization sees when the owner has granted nothing yet.
 NO_MODEL_GRANTED_MESSAGE = "منتظر تخصیص مدل توسط مالک سامانه"
 
+#: The reply a zero-credit organization gets instead of a model call (prepaid, D-B).
+CHAT_NO_CREDIT_MESSAGE = (
+    "اعتبار سازمان صفر است؛ مدیر پس از افزودن اعتبار توسط مالک سامانه پاسخ می‌دهد."
+)
+
 #: Bootstrap tiers per role, in the plan's order (strongest / balanced / fast).
 BOOTSTRAP_TIERS: dict[RoleKey, str] = {
     RoleKey.MANAGER: "strong",
@@ -223,6 +229,7 @@ class ManagerTurns:
         states: PipelineStateRepository | None = None,
         runner: TurnRunner | None = None,
         feed=None,
+        wallets: CreditWalletRepository | None = None,
     ) -> None:
         self.provider = provider
         self.employees = employees
@@ -239,6 +246,7 @@ class ManagerTurns:
         self.states = states
         self.runner = runner
         self.feed = feed
+        self.wallets = wallets
 
     # ------------------------------------------------------------------ #
     # 3.6 bootstrap                                                       #
@@ -404,6 +412,17 @@ class ManagerTurns:
         user_message = await self.messages.create(
             Message(thread_id=thread.id, role=MessageRole.USER, content=content)
         )
+        # Prepaid: a reply costs a model call, so an org with no credit gets the
+        # note instead. No model is called and no tag is executed.
+        if await self._out_of_credit(organization_id):
+            note = await self.messages.create(
+                Message(
+                    thread_id=thread.id,
+                    role=MessageRole.ASSISTANT,
+                    content=CHAT_NO_CREDIT_MESSAGE,
+                )
+            )
+            return user_message, note
         employee = await self._manager_of(organization_id, employee_id)
         extra = "\n\n".join(
             part
@@ -492,6 +511,13 @@ class ManagerTurns:
         )
         return user_message, assistant_message
 
+    async def _out_of_credit(self, organization_id: str) -> bool:
+        """True when a wired wallet shows no credit left (a missing wallet counts as none)."""
+        if self.wallets is None:
+            return False
+        wallet = await self.wallets.get_for_org(organization_id)
+        return wallet is None or wallet.balance <= 0
+
     def _chat_context(
         self,
         *,
@@ -1083,6 +1109,7 @@ class ManagerTurns:
 
 __all__ = [
     "BOOTSTRAP_TIERS",
+    "CHAT_NO_CREDIT_MESSAGE",
     "CHAT_TURN_TIMEOUT_SECONDS",
     "HIRABLE_ROLES",
     "NO_MODEL_GRANTED_MESSAGE",

apps/api/app/application/engine/scheduler.py

@@ -23,6 +23,7 @@ from __future__ import annotations
 
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
+from typing import Literal
 
 from app.application.engine.gates import GateOutcome, GateRunner
 from app.application.engine.manager import ManagerOutcome, ManagerTurns, _pick_granted_model
@@ -32,10 +33,11 @@ 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.realtime import emit_task_status
 from app.application.ports.repo_workspace import RepoWorkspacePort, WorktreeRef
 from app.application.ports.sandbox_provider import SandboxProviderPort
 from app.core.config import settings
-from app.core.errors import ConflictError
+from app.core.errors import ConflictError, NotFoundError, ValidationError
 from app.domain.base import utcnow
 from app.domain.workforce.engine import (
     MAX_TRANSIENT_RETRIES,
@@ -43,8 +45,18 @@ from app.domain.workforce.engine import (
     GateName,
     is_transient,
 )
-from app.domain.workforce.models import Employee, RoleKey, Run, RunStatus, Task, TaskStatus
+from app.domain.workforce.models import (
+    Employee,
+    Message,
+    MessageRole,
+    RoleKey,
+    Run,
+    RunStatus,
+    Task,
+    TaskStatus,
+)
 from app.domain.workforce.pipeline import (
+    TERMINAL_STAGES,
     Outcome,
     Stage,
     TaskPipelineState,
@@ -85,8 +97,10 @@ MERGE_STAGES: tuple[Stage, ...] = (Stage.MERGE_REQUEST,)
 #: address; the *name* is the employee's, which is what a reviewer reads.
 ENGINE_COMMIT_EMAIL = "platform@smart-org.local"
 
-#: Persian reason shown when work pauses because the money ran out.
-BUDGET_PAUSED_MESSAGE = "اعتبار سازمان برای ادامهٔ کار کافی نیست؛ کار موقتاً متوقف شد."
+#: Persian reason shown when work pauses because the credit ran out (prepaid, D-B).
+BUDGET_PAUSED_MESSAGE = (
+    "اعتبار سازمان تمام شده است؛ کار تا افزودن اعتبار توسط مالک سامانه متوقف شد."
+)
 
 #: The stage a turn has to be at for each manager mode.
 _MANAGER_STAGES: frozenset[Stage] = frozenset(
@@ -426,7 +440,14 @@ class Scheduler:
             same_count=state.same_diff_count,
             diff_text=diff.text,
         )
-        await self._patch(state, last_diff_hash=new_hash, same_diff_count=same_count)
+        # One write for the hash, the counter and the diff the task page shows
+        # (capped at 20,000 chars). Only a diff that passed the guard is stored.
+        await self._patch(
+            state,
+            last_diff_hash=new_hash,
+            same_diff_count=same_count,
+            last_diff=diff.text[:20000],
+        )
         if not progress.ok:
             await self._send_back(state, task, progress.detail)
             report.failed += 1
@@ -530,6 +551,7 @@ class Scheduler:
             return
 
         final_sha = pushed.sha or merged.sha or sha
+        await self._patch(state, merged_sha=final_sha)
         await self._move(state, task, Stage.COMPLETED)
         await self.repo_workspace.remove_worktree(worktree=worktree)
         if task.project_id:
@@ -622,16 +644,14 @@ class Scheduler:
     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.
+        Billing is prepaid: the platform owner credits the wallet, and every usage
+        row debits it. Zero means "no credit left", so a zero, negative or missing
+        wallet pauses work before a turn is spent. The per-task cap and the org's
+        monthly budget still apply on top of that.
         """
         if self.wallets is not None:
             wallet = await self.wallets.get_for_org(task.organization_id)
-            # No wallet means no budget was ever set: a fresh org must not be
-            # frozen out of its first task. A zero balance is not a pause either.
-            if wallet is not None and wallet.balance < 0:
+            if wallet is None or 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)
@@ -648,6 +668,74 @@ class Scheduler:
                     return "سقف اعتبار ماهانهٔ سازمان پر شده است؛ کار متوقف شد."
         return None
 
+    # ------------------------------------------------------------------ #
+    # the person's controls                                              #
+    # ------------------------------------------------------------------ #
+
+    async def user_action(
+        self,
+        *,
+        organization_id: str,
+        task_id: str,
+        action: Literal["pause", "resume", "cancel", "answer"],
+        user_id: str,
+        answer: str = "",
+    ) -> TaskPipelineState:
+        """Apply one control a person pressed on a task, and return its new state.
+
+        The task is looked up inside the caller's organization, so a foreign id is
+        a 404. A task outside the engine, or one already closed, is a conflict.
+        Resume is refused while the budget is spent, so a task paused for credit
+        stays paused until credit is added and resume is pressed again. ``user_id``
+        names the person who pressed the control; the engine keeps no actor column,
+        so the route writes the audit row.
+        """
+        task = await self.tasks.get_for_org(organization_id, task_id)
+        if task is None:
+            raise NotFoundError("کار یافت نشد.")
+        state = await self.states.get(task_id)
+        if state is None:
+            raise ConflictError("این کار در موتور خودکار نیست.")
+        if state.stage in TERMINAL_STAGES:
+            raise ConflictError("این کار بسته شده است.")
+        if action == "pause":
+            await self._move(state, task, Stage.PAUSED)
+        elif action == "resume":
+            if state.stage is not Stage.PAUSED:
+                raise ConflictError("فقط کار متوقف‌شده را می‌توان ادامه داد.")
+            reason = await self._budget_reason(task)
+            if reason is not None:
+                raise ConflictError(reason)
+            await self._move(state, task, Stage.PENDING_MANAGER_REVIEW)
+        elif action == "cancel":
+            await self._move(state, task, Stage.CANCELLED)
+        else:
+            await self._answer(state, task, answer)
+        updated = await self.states.get(task_id)
+        if updated is None:  # pragma: no cover - the row was read above
+            raise NotFoundError("این کار وضعیت موتور ندارد.")
+        await emit_task_status(
+            task_id=task_id,
+            organization_id=organization_id,
+            status=coarse_status(updated.stage).value,
+        )
+        return updated
+
+    async def _answer(self, state: TaskPipelineState, task: Task, answer: str) -> None:
+        """A person answers the decision a task waits on; triage reads it next."""
+        content = answer.strip()
+        if not 1 <= len(content) <= 4000:
+            raise ValidationError("پاسخ باید بین ۱ تا ۴۰۰۰ نویسه باشد.")
+        if state.stage is not Stage.PENDING_USER_DECISION:
+            raise ConflictError("این کار منتظر پاسخ شما نیست.")
+        # The move comes first, so a lost race raises before anything is written.
+        await self._move(state, task, Stage.PENDING_MANAGER_REVIEW)
+        await self.tasks.append_description(task.id, "\n\nپاسخ کاربر: " + content)
+        if task.thread_id:
+            await self.manager.messages.create(
+                Message(thread_id=task.thread_id, role=MessageRole.USER, content=content)
+            )
+
     async def _retry_or_fail(
         self, state: TaskPipelineState, task: Task, message: str, kind: str
     ) -> None:

apps/api/app/application/engine/state.py

@@ -19,6 +19,7 @@ truth, the status column is the summary existing APIs keep reading.
 
 from __future__ import annotations
 
+from app.application.engine.events import emit_task_step
 from app.core.errors import ConflictError, NotFoundError
 from app.domain.workforce.pipeline import (
     SIDE_STAGES,
@@ -46,14 +47,14 @@ async def advance(
 
     Raises ``ConflictError`` when the transition is illegal or when another
     worker moved the task first, and ``NotFoundError`` when the task has no
-    engine state row at all.
+    engine state row at all. A move that lands publishes one ``task.step``.
     """
     if not can_transition_stage(expected_from, to):
         raise ConflictError(
             f"گذار نابجا از {expected_from.value} به {to.value}."
         )
     try:
-        return await states.advance(task_id=task_id, expected_from=expected_from, to=to)
+        moved = await states.advance(task_id=task_id, expected_from=expected_from, to=to)
     except UnknownTaskError as exc:
         raise NotFoundError("این کار وضعیت موتور ندارد.") from exc
     except InvalidTransitionError as exc:
@@ -62,6 +63,15 @@ async def advance(
         raise ConflictError(
             "این کار هم‌اکنون توسط موتور دیگری در حرکت است؛ دوباره تلاش کنید."
         ) from exc
+    # The publisher never raises, so a broken bus cannot undo the move.
+    await emit_task_step(
+        task_id=moved.task_id,
+        organization_id=moved.organization_id,
+        from_stage=expected_from,
+        to_stage=to,
+        pipeline=moved.pipeline,
+    )
+    return moved
 
 
 async def park_dependents(

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

@@ -17,6 +17,7 @@ from app.application.ports import PaymentProviderPort
 from app.application.ports.realtime import emit_wallet_changed
 from app.core.errors import NotFoundError, PermissionDeniedError, ValidationError
 from app.domain.billing.models import (
+    CreditKind,
     CreditWallet,
     Invoice,
     Payment,
@@ -31,6 +32,7 @@ from app.infrastructure.repositories.billing import (
     PlanRepository,
     SubscriptionRepository,
 )
+from app.infrastructure.repositories.credit import CreditLedgerRepository
 
 #: A tenant may not buy an arbitrary amount of credit in one action.
 MIN_TOPUP = 1.0
@@ -56,6 +58,8 @@ class BillingService:
         self.subscriptions = subscriptions
         self.ledger = ledger
         self.payment_provider = payment_provider
+        # Settled payments are credited through the credit ledger, in the wallets' session.
+        self.credit_ledger = CreditLedgerRepository(wallets.session)
 
     # -- reads ---------------------------------------------------------------
 
@@ -107,6 +111,8 @@ class BillingService:
     ) -> CreditWallet:
         """Grant (or, for a platform owner, claw back) credit.
 
+        Superseded by the owner-only ``CreditService``, which writes the credit ledger.
+
         ``allow_negative`` is reserved for the platform owner console; an org
         admin may only ever add credit.
         """
@@ -171,7 +177,9 @@ class BillingService:
 
         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.
+        untouched rather than credited twice. The credit is one ``adjustment`` row in
+        the credit ledger, written in this session, so it shows in the owner's credit
+        history and updates ``last_credit_at``.
         """
         payment = next(
             (p for p in await self.payments.list_for_org(organization_id) if p.id == payment_id),
@@ -183,12 +191,15 @@ class BillingService:
             return payment
 
         if succeeded:
-            wallet = await self.wallets.adjust(
-                organization_id=organization_id, delta=payment.amount
+            credit = await self.credit_ledger.apply(
+                organization_id=organization_id,
+                amount=payment.amount,
+                kind=CreditKind.ADJUSTMENT,
+                reason="billing/settle-payment",
             )
             final = await self.payments.set_status(payment_id, PaymentStatus.SUCCEEDED)
             await emit_wallet_changed(
-                organization_id=organization_id, balance=wallet.balance
+                organization_id=organization_id, balance=credit.balance_after
             )
         else:
             final = await self.payments.set_status(payment_id, PaymentStatus.FAILED)

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

@@ -0,0 +1,136 @@
+"""Prepaid credit: the platform owner's top-ups, the history and the balances.
+
+An organization's wallet is only ever moved through ``CreditLedgerRepository``, so every
+change has a history row. This service is the owner's side of that ledger: it credits one
+organization by hand, reads its manual history, lists every organization's balance for the
+owner's credit page, and grants the optional starting credit to a new organization. Usage
+debits are not written here; ``UsageLedgerRepository.record`` writes them.
+
+Only the platform owner may call the write methods and the cross-organization ``balances``.
+The routers enforce that; this service does not look at who is calling.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.application.ports.realtime import emit_wallet_changed
+from app.core.config import settings
+from app.core.errors import NotFoundError, ValidationError
+from app.domain.billing.models import CreditKind, CreditTransaction
+from app.domain.workforce.pipeline import Stage
+from app.infrastructure.repositories.billing import CreditWalletRepository
+from app.infrastructure.repositories.credit import CreditLedgerRepository
+from app.infrastructure.repositories.engine import PipelineStateRepository
+from app.infrastructure.repositories.identity import OrganizationRepository
+
+#: The most one owner action may credit at once. A bigger top-up is made in steps.
+MAX_OWNER_CREDIT = 100_000.0
+
+#: The reason is required and stored with the history row (at most 500 characters).
+_REASON_MIN = 3
+_REASON_MAX = 500
+
+#: The entries a history page shows: what someone did to the wallet, not usage debits.
+_MANUAL_KINDS = (CreditKind.OWNER_CREDIT, CreditKind.STARTING_CREDIT, CreditKind.ADJUSTMENT)
+
+
+@dataclass(slots=True, kw_only=True)
+class OrgBalance:
+    """One organization's row on the platform owner's credit page."""
+
+    organization_id: str
+    name: str
+    slug: str
+    balance: float
+    currency: str
+    paused_tasks: int
+    last_credit_at: datetime | None
+
+
+class CreditService:
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+        self.ledger = CreditLedgerRepository(session)
+        self.wallets = CreditWalletRepository(session)
+        self.orgs = OrganizationRepository(session)
+        self.states = PipelineStateRepository(session)
+
+    async def owner_credit(
+        self,
+        *,
+        organization_id: str,
+        amount: float,
+        reason: str,
+        actor_user_id: str,
+    ) -> CreditTransaction:
+        """Credit one organization's wallet on the platform owner's word."""
+        if await self.orgs.get(organization_id) is None:
+            raise NotFoundError("سازمان یافت نشد.")
+        if not 0 < amount <= MAX_OWNER_CREDIT:
+            raise ValidationError(
+                f"مقدار اعتبار باید بیشتر از صفر و حداکثر {MAX_OWNER_CREDIT:g} باشد."
+            )
+        clean_reason = reason.strip()
+        if not _REASON_MIN <= len(clean_reason) <= _REASON_MAX:
+            raise ValidationError("دلیل افزودن اعتبار لازم است.")
+
+        transaction = await self.ledger.apply(
+            organization_id=organization_id,
+            amount=amount,
+            kind=CreditKind.OWNER_CREDIT,
+            reason=clean_reason,
+            actor_user_id=actor_user_id,
+        )
+        # Live push: the sidebar credit badge follows the wallet as it moves.
+        await emit_wallet_changed(
+            organization_id=organization_id, balance=transaction.balance_after
+        )
+        return transaction
+
+    async def history(self, organization_id: str, *, limit: int = 50) -> list[CreditTransaction]:
+        """One organization's manual credit history, newest first. Usage debits are left out."""
+        return await self.ledger.history(organization_id, kinds=_MANUAL_KINDS, limit=limit)
+
+    async def balances(self) -> list[OrgBalance]:
+        """Every organization's balance, paused task count and last credit date.
+
+        Cross-organization on purpose: only the platform owner's credit page may call it.
+        A missing wallet is shown as zero.
+        """
+        paused = await self.states.count_at_by_org(Stage.PAUSED)
+        last_credit = await self.ledger.last_credit_at_by_org()
+        rows: list[OrgBalance] = []
+        for org in await self.orgs.list_all():
+            wallet = await self.wallets.get_for_org(org.id)
+            rows.append(
+                OrgBalance(
+                    organization_id=org.id,
+                    name=org.name,
+                    slug=org.slug,
+                    balance=wallet.balance if wallet else 0.0,
+                    currency=wallet.currency if wallet else "USD",
+                    paused_tasks=paused.get(org.id, 0),
+                    last_credit_at=last_credit.get(org.id),
+                )
+            )
+        return rows
+
+    async def grant_starting_credit(self, organization_id: str) -> CreditTransaction | None:
+        """Give a new organization the configured starting credit, if one is set.
+
+        ``BILLING_STARTING_CREDIT`` defaults to 0, so by default nothing is granted and the
+        owner credits new organizations by hand.
+        """
+        amount = float(getattr(settings, "billing_starting_credit", 0.0))
+        if amount <= 0:
+            return None
+        return await self.ledger.apply(
+            organization_id=organization_id,
+            amount=amount,
+            kind=CreditKind.STARTING_CREDIT,
+            reason="اعتبار آغازین",
+        )

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

@@ -45,8 +45,9 @@ TIER_LABELS: dict[str, str] = {
 
 #: Policy defaults for an organization that has never been configured. Kept in
 #: code (not only in the row default) so ``get_policy`` can answer for any org
-#: without writing a row first.
-DEFAULT_MAX_EMPLOYEES = 5
+#: without writing a row first. D-A: the default cap is 7 employees per
+#: organization; ``settings.engine_max_employees`` stays the upper bound.
+DEFAULT_MAX_EMPLOYEES = 7
 
 
 @dataclass(slots=True)

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

@@ -221,6 +221,11 @@ class OrgSetupService:
             )
         )
         self.session.add(CreditWalletRow(organization_id=org.id, balance=0.0, currency="USD"))
+        # Prepaid: the optional starting credit (0 by default) goes through the ledger.
+        await self.session.flush()
+        from app.application.services.credit_service import CreditService
+
+        await CreditService(self.session).grant_starting_credit(org.id)
 
         # -- create the single app -------------------------------------------
         project_row = await self._create_single_app(org.id, user.id, payload, catalog_defaults)

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

@@ -32,6 +32,7 @@ from app.application.ports.realtime import (
     emit_task_created,
     emit_task_status,
     emit_usage_recorded,
+    emit_wallet_changed,
 )
 from app.core.errors import (
     ConflictError,
@@ -61,6 +62,13 @@ from app.domain.workforce.models import (
     ToolCallStatus,
     can_transition_run,
 )
+from app.domain.workforce.pipeline import (
+    Risk,
+    Stage,
+    TaskKind,
+    TaskPipelineState,
+    resolve_pipeline,
+)
 from app.domain.workforce.tools import (
     TOOL_CATALOG,
     ToolSpec,
@@ -74,6 +82,7 @@ from app.infrastructure.repositories.ai_control import (
     RoutingPolicyRepository,
 )
 from app.infrastructure.repositories.audit import UsageLedgerRepository
+from app.infrastructure.repositories.engine import PipelineStateRepository
 from app.infrastructure.repositories.projects import ProjectRepository
 from app.infrastructure.repositories.workforce import (
     AgentTemplateRepository,
@@ -94,6 +103,12 @@ from app.infrastructure.repositories.workforce import (
 #: non-zero placeholder and mark it in the usage metadata.
 _UNPRICED = 0.0000001
 
+#: Said when the legacy "run" button is pressed on a task the engine owns.
+_ENGINE_OWNS_TASK = (
+    "این کار را موتور خودکار اجرا می‌کند؛ "
+    "برای توقف، ادامه یا لغو از دکمه‌های همین کار استفاده کنید."
+)
+
 _SYSTEM_PROMPT = (
     "تو دستیار هوش مصنوعی سازمان هستی. پاسخ‌ها را کوتاه، دقیق و عملی بده. "
     "هیچ‌گاه نام مدل، ارائه‌دهنده یا جزئیات زیرساخت را فاش نکن."
@@ -126,7 +141,7 @@ async def record_model_usage(
     card = await 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 ledger.record(
+    balance = await ledger.record(
         usage=UsageRecord(
             organization_id=organization_id,
             project_id=project_id,
@@ -141,6 +156,9 @@ async def record_model_usage(
             metadata_json={"priced": card is not None},
         )
     )
+    if balance is not None:
+        # The debit moved the wallet: the sidebar badge follows it live.
+        await emit_wallet_changed(organization_id=organization_id, balance=balance)
     # Live push: the billing page and the sidebar credit badge react to
     # consumption as it happens instead of at the next page load.
     await emit_usage_recorded(
@@ -406,6 +424,7 @@ class WorkforceService:
         """
         from app.application.container_org import build_memory, build_model_grants
         from app.application.engine.manager import ManagerTurns
+        from app.infrastructure.repositories.billing import CreditWalletRepository
         from app.infrastructure.repositories.engine import (
             EmployeeProfileRepository,
             PipelineStateRepository,
@@ -426,6 +445,7 @@ class WorkforceService:
             memory=build_memory(session),
             profiles=EmployeeProfileRepository(session),
             states=PipelineStateRepository(session),
+            wallets=CreditWalletRepository(session),
         )
 
     # ------------------------------------------------------------------ #
@@ -443,6 +463,13 @@ class WorkforceService:
         thread_id: str | None = None,
         assigned_employee_id: str | None = None,
     ) -> Task:
+        """A user order to the manager: the task enters the engine at manager review.
+
+        The task gets a pipeline state at ``pending_manager_review``, so the scheduler's
+        triage turn takes it from there and picks the real pipeline and assignee (D-D).
+        Without a state the legacy one-shot drain would run it outside the engine. MCP
+        ``tasks_create`` calls this method, so its tasks enter the engine too.
+        """
         if not title.strip():
             raise ValidationError("عنوان کار نمی‌تواند خالی باشد.")
         await self._own_project_or_404(organization_id=organization_id, project_id=project_id)
@@ -451,6 +478,7 @@ class WorkforceService:
             await self.get_employee_or_404(
                 organization_id=organization_id, employee_id=assigned_employee_id
             )
+        project_id = project_id or await self._project_id_for(organization_id)
         task = await self.tasks.create(
             Task(
                 organization_id=organization_id,
@@ -462,6 +490,21 @@ class WorkforceService:
                 created_by_user_id=created_by_user_id,
             )
         )
+        await PipelineStateRepository(self.tasks.session).create(
+            TaskPipelineState(
+                task_id=task.id,
+                organization_id=organization_id,
+                project_id=project_id,
+                pipeline=resolve_pipeline(
+                    manager_choice=None,
+                    task_kind=TaskKind.FEATURE,
+                    has_active_reviewer=False,
+                ),
+                kind=TaskKind.FEATURE,
+                risk=Risk.MEDIUM,
+                stage=Stage.PENDING_MANAGER_REVIEW,
+            )
+        )
         # Live push: the tasks list and the manager-chat sidebar update the
         # moment the task exists, whoever created it and from where.
         await emit_task_created(
@@ -472,6 +515,16 @@ class WorkforceService:
         )
         return task
 
+    async def _project_id_for(self, organization_id: str) -> str | None:
+        """The project a project-less order lands in: the one its employees work in.
+
+        Mirrors ``ManagerTurns._project_id_for``, so a task made here and a task the
+        manager makes from a chat get the same project.
+        """
+        for employee in await self.employees.list_for_org(organization_id, limit=1):
+            return employee.project_id
+        return None
+
     async def list_tasks(
         self,
         organization_id: str,
@@ -506,6 +559,9 @@ class WorkforceService:
         employee_id: str | None = None,
     ) -> Run:
         task = await self.get_task_or_404(organization_id=organization_id, task_id=task_id)
+        if await PipelineStateRepository(self.tasks.session).get(task.id) is not None:
+            # The engine owns this task; its own buttons (pause, resume, cancel, answer) drive it.
+            raise ConflictError(_ENGINE_OWNS_TASK)
         if task.status in (TaskStatus.DONE, TaskStatus.CANCELLED):
             raise ConflictError("این کار قبلاً بسته شده است.")