feat(engine): the turn runner — prompt builder, sandbox tools, agent loop

Phase 2 of TODO-agent-2: one employee, one turn, real tools in the sandbox.

- engine/prompt.py: the stable, cache-friendly prefix in the plan's order
  (rules → role protocol → toolkit → app profile → memory), a task brief
  capped at 1200 chars, role-specific stage context (spec for the developer,
  the diff only for the reviewer, test output for the tester) and a hard
  prompt cap that drops the oldest tool results first.
- engine/tools.py: READ/WRITE/EDIT/LIST_DIR/SEARCH/RUN executed inside the
  sandbox through SandboxProviderPort, rooted at the worktree's sandbox path.
  Path escapes, .. and .git are refused; git push/remote/config are refused
  with "the platform does git"; RUN clamps its timeout (120s default, 600s
  max) and keeps the tail of its output; every execution writes a tool_calls
  row so the approvals UI shows real work.
- engine/runner.py: the loop — model call, parse tags, execute tool tags,
  feed results back — until a terminal tag, the tool cap, the turn timeout or
  a cancellation. The employee's model is checked against ModelGrantPort
  before the first call (ungranted → error_kind=auth, no spend), errors are
  classified transient vs fatal, and every model call writes usage events
  through the one shared record_model_usage path (extracted from
  WorkforceService so manager chat counts too). Writes engine_turns, run_steps
  and tool_calls.
- engine/events.py: RunFeed batches run.log (≤1 event / 500 ms / stream, ≤200
  lines) and emits run.tool and task.gate through the generic envelope.
- tests: engine_fakes.py (ScriptedModel, FakeSandbox, FakeGrants,
  RecordingMemory) plus 54 new tests in test_engine_runner,
  test_engine_prompt and test_engine_tools.

Suite: 648 passed, ruff clean, mypy clean.

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

@@ -0,0 +1,189 @@
+"""The engine's live feed: batched ``run.log`` / ``run.tool`` / ``task.gate``.
+
+PLAN-v2 §5 requires log events to be *batched* — at most one event per 500 ms
+per stream, at most 200 lines per event. A model loop that emits one WebSocket
+event per tool call turns a chatty agent into a flood that buries the browser,
+so the rule is not a nicety.
+
+``RunFeed`` buffers log lines per run and publishes them at the boundaries the
+runner chooses: between steps (rate-limited) and at the end of a turn (always).
+Tool results and gate verdicts are their own event types and never wait behind
+the log buffer — a command's output is the thing the live log panel exists for.
+
+Publishing goes through the generic envelope rather than a typed emitter
+because ``ports/realtime.py`` is a frozen contract and these three event types
+have no typed helper there.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+
+from app.application.ports.realtime import (
+    RealtimeEvent,
+    org_channel,
+    publish,
+    run_channel,
+    task_channel,
+)
+from app.core.config import settings
+
+
+@dataclass(slots=True, kw_only=True)
+class _Stream:
+    """One run's buffer: the lines waiting, and when we last published."""
+
+    run_id: str
+    task_id: str
+    organization_id: str
+    lines: list[str] = field(default_factory=list)
+    last_publish: float = 0.0
+    #: Set when the buffer reached its line cap, so the next flush must not
+    #: wait for the interval to elapse.
+    overflow: bool = False
+
+
+class RunFeed:
+    """Batches engine output into bounded realtime events, per run."""
+
+    def __init__(
+        self, *, min_interval: float | None = None, max_lines: int | None = None
+    ) -> None:
+        self._min_interval = (
+            settings.engine_log_flush_seconds if min_interval is None else min_interval
+        )
+        self._max_lines = (
+            settings.engine_log_max_lines_per_event if max_lines is None else max_lines
+        )
+        self._streams: dict[str, _Stream] = {}
+        #: Events published, by type — so a test can assert "one batched log
+        #: event, not one per line" without touching the realtime bus.
+        self.published: dict[str, int] = {}
+
+    # -- buffering ----------------------------------------------------------
+
+    def log(self, *, run_id: str, task_id: str, organization_id: str, line: str) -> None:
+        """Buffer one log line. Nothing is published until a flush."""
+        stream = self._stream(run_id, task_id, organization_id)
+        stream.lines.append(line)
+        if len(stream.lines) >= self._max_lines:
+            # The buffer is full: the next flush must not wait for the timer.
+            stream.overflow = True
+
+    async def tool(
+        self,
+        *,
+        run_id: str,
+        task_id: str,
+        organization_id: str,
+        tool: str,
+        ok: bool,
+        summary: str,
+    ) -> None:
+        """Report one executed tool call as a single, self-contained event."""
+        await self._emit(
+            RealtimeEvent(
+                type="run.tool",
+                channels=(
+                    run_channel(run_id),
+                    task_channel(task_id),
+                    org_channel(organization_id),
+                ),
+                organization_id=organization_id,
+                data={
+                    "run_id": run_id,
+                    "task_id": task_id,
+                    "tool": tool,
+                    "ok": ok,
+                    "summary": summary[:2000],
+                },
+            )
+        )
+
+    async def gate(
+        self,
+        *,
+        task_id: str,
+        organization_id: str,
+        gate: str,
+        ok: bool,
+        detail: str,
+    ) -> None:
+        """Report one gate verdict, with the evidence attached."""
+        await self._emit(
+            RealtimeEvent(
+                type="task.gate",
+                channels=(task_channel(task_id), org_channel(organization_id)),
+                organization_id=organization_id,
+                data={
+                    "task_id": task_id,
+                    "gate": gate,
+                    "ok": ok,
+                    "detail": detail[:8000],
+                },
+            )
+        )
+
+    # -- flushing -----------------------------------------------------------
+
+    async def flush(self, *, run_id: str, force: bool = False) -> None:
+        """Publish a run's buffered lines, unless the interval has not elapsed.
+
+        ``force`` is for the end of a turn: whatever is buffered then is the
+        last thing the browser will see until the next turn, so it goes out.
+        """
+        stream = self._streams.get(run_id)
+        if stream is None or not stream.lines:
+            return
+        now = time.monotonic()
+        elapsed = now - stream.last_publish
+        if not force and not stream.overflow and elapsed < self._min_interval:
+            return
+        await self._publish(stream, force=force or stream.overflow)
+
+    async def flush_all(self) -> None:
+        for stream in list(self._streams.values()):
+            await self._publish(stream, force=True)
+
+    # -- internals ----------------------------------------------------------
+
+    async def _publish(self, stream: _Stream, *, force: bool) -> None:
+        if force:
+            lines, stream.lines = list(stream.lines), []
+        else:
+            lines, stream.lines = stream.lines[: self._max_lines], stream.lines[self._max_lines :]
+        stream.overflow = False
+        if not lines:
+            return
+        stream.last_publish = time.monotonic()
+        await self._emit(
+            RealtimeEvent(
+                type="run.log",
+                channels=(
+                    run_channel(stream.run_id),
+                    task_channel(stream.task_id),
+                    org_channel(stream.organization_id),
+                ),
+                organization_id=stream.organization_id,
+                data={"run_id": stream.run_id, "task_id": stream.task_id, "lines": lines},
+            )
+        )
+
+    async def _emit(self, event: RealtimeEvent) -> None:
+        self.published[event.type] = self.published.get(event.type, 0) + 1
+        # The realtime publisher is documented as never raising; a counter
+        # keeps tests honest even when the bus is a NullPublisher.
+        await publish(event)
+
+    def _stream(self, run_id: str, task_id: str, organization_id: str) -> _Stream:
+        stream = self._streams.get(run_id)
+        if stream is None:
+            stream = _Stream(
+                run_id=run_id, task_id=task_id, organization_id=organization_id
+            )
+            self._streams[run_id] = stream
+        return stream
+
+
+__all__ = ["RunFeed"]

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

@@ -0,0 +1,299 @@
+"""Building one turn's prompt: a stable prefix and a bounded tail.
+
+The order is not cosmetic, it is a caching decision. A provider that supports
+prompt caching (agy does) can reuse the *prefix* across every turn of a task
+and every task of a role, so everything that changes rarely goes first and
+everything that changes every turn goes last:
+
+1. platform rules — short, and never change
+2. the role protocol — allowed tags plus a worked example, per role/mode
+3. the toolkit description — what the tools do and what they refuse
+4. the app profile — language, framework, commands (never a secret)
+5. org memory — what this organization learned about itself
+6. the task brief (capped) plus the stage-specific context the role needs:
+   the planner's spec for the developer, **the diff only** for the reviewer,
+   the test output for the tester
+7. the conversation so far in this turn — the tool results, newest last
+
+The whole prompt has a hard character cap. When it is exceeded the builder
+drops the **oldest tool results first**, and never the protocol, the memory
+context or the task brief: a model that forgets the rules is worse than a
+model that forgot what it read three calls ago.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from app.application.engine.protocol import (
+    TOOL_TAGS,
+    allowed_tags,
+)
+from app.application.ports import ChatMessage
+from app.core.config import settings
+from app.domain.workforce.models import RoleKey
+from app.domain.workforce.pipeline import Mode, Stage
+
+#: The task brief is capped before anything else, so a chatty task
+#: description can never squeeze the protocol out of the prompt.
+MAX_TASK_BRIEF_CHARS = 1200
+
+#: Marker written in place of a tool result that was dropped for space.
+DROPPED = "[… نتیجهٔ قبلی به‌خاطر محدودیت طول حذف شد …]"
+
+#: Platform rules. Deliberately short: this is the cacheable prefix, and the
+#: role protocol below carries the format rules.
+PLATFORM_RULES = (
+    "قوانین پلتفرم:\n"
+    "۱. تو یک کارمند هوش مصنوعی درون سامانه هستی و فقط با برچسب‌های مجاز کار می‌کنی.\n"
+    "۲. نام مدل، ارائه‌دهنده، حساب یا هر جزئیات زیرساختی را هرگز فاش نکن.\n"
+    "۳. هر تغییری در مخزن را پلتفرم با git انجام می‌دهد؛ تو فایل‌ها را ویرایش و "
+    "دستورها را در سندباکس اجرا می‌کنی.\n"
+    "۴. اگر به تصمیم انسان نیاز بود، از ASK_USER استفاده کن و علت را بنویس.\n"
+    "۵. فارسی، کوتاه و عملی پاسخ بده."
+)
+
+#: What the tags look like, with a worked example per role/mode.
+_PROTOCOL_HINTS: dict[tuple[RoleKey, Mode], str] = {
+    (RoleKey.MANAGER, Mode.TRIAGE): (
+        "تو مدیر هستی و کار را ارزیابی می‌کنی.\n"
+        "برچسب‌های مجاز:\n"
+        '[[ASSIGN employee="نام کارمند" pipeline="plan_code_test"/]]\n'
+        "[[REJECT reason=\"دلیل\"/]]\n"
+        "[[ASK_USER]]سؤالت را اینجا بنویس[[/ASK_USER]]\n"
+        '[[KNOWLEDGE kind="rule" key="..."]]قاعده‌ای که یاد گرفتی[[/KNOWLEDGE]]\n'
+        "دقیقاً یکی از این برچسب‌ها را انتخاب کن."
+    ),
+    (RoleKey.MANAGER, Mode.FINAL): (
+        "تو مدیر هستی و تصمیم نهایی را می‌گیری.\n"
+        "برچسب‌های مجاز:\n"
+        "[[COMPLETE/]] یا [[REVISE]]دلیل بازگشت کار[[/REVISE]]\n"
+        '[[KNOWLEDGE kind="rule" key="..."]]قاعده‌ای که یاد گرفتی[[/KNOWLEDGE]]'
+    ),
+    (RoleKey.MANAGER, Mode.ERROR): (
+        "تو مدیر هستی و خطای یک کار را بررسی می‌کنی.\n"
+        "برچسب‌های مجاز:\n"
+        '[[RESOLVE_ERROR hint="راهنمایی برای برنامه‌نویس"/]] یا '
+        '[[CANCEL reason="دلیل لغو"/]]\n'
+        '[[KNOWLEDGE kind="lesson" key="..."]]درس آموخته‌شده[[/KNOWLEDGE]]'
+    ),
+    (RoleKey.MANAGER, Mode.CHAT): (
+        "تو مدیر هستی و با کاربر گفتگو می‌کنی.\n"
+        "برچسب‌های مجاز:\n"
+        '[[CREATE_TASK title="عنوان کار" kind="feature|bugfix|refactor|docs|analysis" '
+        'priority="normal|high" depends_on="id1,id2" pipeline="..."]'
+        "]]توضیح اختیاری[[/CREATE_TASK]]\n"
+        '[[PAUSE_TASK task_id="..."/]] [[RESUME_TASK task_id="..."/]] '
+        '[[CANCEL_TASK task_id="..."/]]\n'
+        '[[HIRE role="developer|tester|reviewer|planner" '
+        'tier="fast|balanced|strong" name="نام کارمند"/]]\n'
+        '[[UPDATE_EMPLOYEE employee_id="..." tier="..."/]] '
+        '[[FIRE employee_id="..."/]]\n'
+        '[[KNOWLEDGE kind="rule|fact" key="..."]'
+        "]]متن[[/KNOWLEDGE]]\n"
+        "پاسخ متنی‌ات برای کاربر نمایش داده می‌شود؛ برچسب‌ها کنارش اجرا می‌شوند. "
+        "کارها را اتمیک بساز و نگو که کدام فایل‌ها را هم‌زمان تغییر می‌دهند."
+    ),
+    (RoleKey.PLANNER, Mode.WORK): (
+        "تو برنامه‌ریز هستی.\n"
+        "برچسب‌های مجاز:\n"
+        "[[PLAN_SPEC]]برنامهٔ اجرا: مراحل، فایل‌های درگیر، ریسک‌ها[[/PLAN_SPEC]]\n"
+        '[[SPLIT title="عنوان کار جدا" kind="..."]]توضیح[[/SPLIT]]\n'
+        "[[ASK_USER]]سؤال[[/ASK_USER]]\n"
+        "برای دیدن فایل‌ها از ابزارهای خواندن استفاده کن و با PLAN_SPEC پایان بده."
+    ),
+    (RoleKey.DEVELOPER, Mode.WORK): (
+        "تو برنامه‌نویس هستی.\n"
+        "برچسب‌های مجاز:\n"
+        "[[WORK_REPORT]]خلاصهٔ تغییرات، فایل‌های ویرایش‌شده و نتیجهٔ تست‌ها[[/WORK_REPORT]]\n"
+        "[[ASK_USER]]سؤال[[/ASK_USER]]\n"
+        "قبل از WORK_REPORT حتماً تست‌ها را با RUN اجرا کن."
+    ),
+    (RoleKey.REVIEWER, Mode.WORK): (
+        "تو بازبین هستی و diff را می‌بینی.\n"
+        "برچسب‌های مجاز:\n"
+        "[[REVIEW_PASS]]نظر مثبت[[/REVIEW_PASS]] یا "
+        "[[REVIEW_CHANGES]]آنچه باید اصلاح شود[[/REVIEW_CHANGES]]"
+    ),
+    (RoleKey.TESTER, Mode.WORK): (
+        "تو آزمونگر هستی.\n"
+        "برچسب‌های مجاز:\n"
+        '[[TEST_PASS]]نتیجه[[/TEST_PASS]] یا [[TEST_PASS waive="دلیل رهاسازی"]]\n'
+        "[[TEST_FAIL]]خروجی خطا[[/TEST_FAIL]]\n"
+        "دستور تست را خودت با RUN اجرا کن."
+    ),
+}
+
+#: What each tool does and what it refuses. Shown to every working role.
+TOOLKIT_DESCRIPTION = (
+    "ابزارها (داخل سندباکس اجرا می‌شوند):\n"
+    '[[READ_FILE path="app/main.py"/]]\n'
+    '[[WRITE_FILE path="app/x.py"]]محتوای کامل فایل[[/WRITE_FILE]]\n'
+    '[[EDIT_FILE path="app/x.py"]]<<<OLD\nکد قدیمی\n===\nکد جدید\nNEW>>>[[/EDIT_FILE]]\n'
+    '[[LIST_DIR path="app"/]] [[SEARCH pattern="TODO" path="app"/]]\n'
+    '[[RUN cmd="pytest -q" timeout="120"/]]\n'
+    "قواعد: مسیرها نسبی و داخل پروژه‌اند؛ git push/remote/config توسط پلتفرم انجام "
+    "می‌شود و رد می‌شود؛ خروجی RUN بریده می‌شود."
+)
+
+
+@dataclass(slots=True, kw_only=True)
+class TurnMessage:
+    """One entry of the in-turn conversation: a tool call or its result."""
+
+    kind: str  # tool | result | note
+    text: str
+
+
+@dataclass(slots=True, kw_only=True)
+class TurnContext:
+    """Everything the builder needs about this turn."""
+
+    organization_id: str
+    task_id: str
+    project_id: str | None = None
+    task_title: str = ""
+    task_description: str = ""
+    stage: Stage = Stage.PENDING_IMPLEMENTATION
+    role: RoleKey = RoleKey.DEVELOPER
+    mode: Mode = Mode.WORK
+    app_profile_text: str = ""
+    memory_context: str = ""
+    #: The planner's spec, shown to the developer (and the reviewer).
+    spec: str = ""
+    #: The diff, shown to the reviewer.
+    diff_text: str = ""
+    #: The latest test output, shown to the tester and the manager.
+    test_output: str = ""
+    #: Free-form stage context from the scheduler (e.g. the error hint).
+    stage_context: str = ""
+    history: list[TurnMessage] = field(default_factory=list)
+
+
+@dataclass(slots=True, kw_only=True)
+class TurnPrompt:
+    """The prompt for one model call: a stable prefix and a bounded tail."""
+
+    system: str
+    messages: list[ChatMessage] = field(default_factory=list)
+    total_chars: int = 0
+
+
+def _stage_context_for(ctx: TurnContext) -> str:
+    """The role-specific part of the brief: only what this role must see."""
+    parts: list[str] = []
+    if ctx.stage_context:
+        parts.append(ctx.stage_context.strip())
+    if ctx.role is RoleKey.DEVELOPER and ctx.spec:
+        parts.append(f"برنامهٔ تأییدشده:\n{ctx.spec.strip()}")
+    if ctx.role is RoleKey.REVIEWER and ctx.diff_text:
+        # The diff only: a reviewer that reads the whole tree re-reviews
+        # everything and finds nothing.
+        parts.append(f"diff تغییرات:\n{ctx.diff_text.strip()}")
+    if ctx.role is RoleKey.TESTER and ctx.test_output:
+        parts.append(f"آخرین خروجی تست:\n{ctx.test_output.strip()}")
+    if ctx.role is RoleKey.MANAGER and ctx.mode is Mode.FINAL:
+        if ctx.spec:
+            parts.append(f"برنامهٔ کار:\n{ctx.spec.strip()}")
+        if ctx.diff_text:
+            parts.append(f"خلاصهٔ diff:\n{ctx.diff_text.strip()}")
+        if ctx.test_output:
+            parts.append(f"نتیجهٔ تست:\n{ctx.test_output.strip()}")
+    return "\n\n".join(parts)
+
+
+def _history_messages(history: list[TurnMessage], *, budget: int) -> list[ChatMessage]:
+    """Render the turn's history, dropping the oldest tool results first.
+
+    The budget is what is left of the cap after the prefix and the brief, so
+    the caller decides how much room the conversation may take. Tool calls are
+    the model's own words and results are the environment's: both are rendered
+    as ``assistant``/``user`` exchange — providers have no sandbox-specific
+    role, and pretending otherwise is what makes tool loops fragile.
+    """
+    rendered: list[ChatMessage] = []
+    for item in history:
+        role = "user" if item.kind == "result" else "assistant"
+        rendered.append(ChatMessage(role=role, content=item.text))
+    total = sum(len(m.content) for m in rendered)
+    if total <= budget:
+        return rendered
+    # Walk from the oldest, dropping results (never the calls, so the model
+    # still sees what it did) until the rest fits.
+    for index, msg in enumerate(rendered):
+        if total <= budget:
+            break
+        if history[index].kind != "result":
+            continue
+        saved = len(msg.content)
+        msg.content = DROPPED
+        total -= saved - len(DROPPED)
+    if total <= budget:
+        return rendered
+    # Last resort, and it should not happen in practice: keep the newest
+    # messages, because the newest result answers the model's last question.
+    kept: list[ChatMessage] = []
+    remaining = budget
+    for msg in reversed(rendered):
+        if remaining <= 0:
+            break
+        if len(msg.content) > remaining:
+            msg.content = msg.content[-remaining:]
+        kept.append(msg)
+        remaining -= len(msg.content)
+    kept.reverse()
+    return kept
+
+
+def build_turn_prompt(ctx: TurnContext) -> TurnPrompt:
+    """Build the system prefix and the user tail for one turn."""
+    granted = allowed_tags(role=ctx.role, mode=ctx.mode)
+    protocol_hint = _PROTOCOL_HINTS.get((ctx.role, ctx.mode), "با برچسب‌های مجاز پاسخ بده.")
+    tag_list = "، ".join(sorted(t.value for t in granted))
+
+    system = "\n\n".join(
+        [
+            PLATFORM_RULES,
+            f"نقش: {ctx.role.value} ({ctx.mode.value}) — مرحلهٔ {ctx.stage.value}\n"
+            f"برچسب‌های مجاز: {tag_list}\n\n{protocol_hint}",
+            TOOLKIT_DESCRIPTION if granted & TOOL_TAGS else "",
+            ctx.app_profile_text.strip(),
+            ctx.memory_context.strip(),
+        ]
+    ).strip()
+
+    brief_parts = [f"عنوان کار: {ctx.task_title}"]
+    if ctx.task_description:
+        brief_parts.append(f"توضیحات: {ctx.task_description}")
+    brief = "\n".join(brief_parts)[:MAX_TASK_BRIEF_CHARS]
+    tail_parts = [brief]
+    stage_context = _stage_context_for(ctx)
+    if stage_context:
+        tail_parts.append(stage_context)
+    tail = "\n\n".join(tail_parts)
+
+    budget = max(settings.engine_prompt_char_cap - len(system) - len(tail), 500)
+    messages = _history_messages(ctx.history, budget=budget)
+    messages.append(ChatMessage(role="user", content=tail or "ادامه بده."))
+    return TurnPrompt(
+        system=system,
+        messages=messages,
+        total_chars=len(system) + sum(len(m.content) for m in messages),
+    )
+
+
+def protocol_error_feedback(errors: tuple[str, ...]) -> str:
+    """The protocol errors of the last reply, as a message for the next one."""
+    return "\n".join(errors)
+
+
+__all__ = [
+    "MAX_TASK_BRIEF_CHARS",
+    "PLATFORM_RULES",
+    "TOOLKIT_DESCRIPTION",
+    "TurnContext",
+    "TurnMessage",
+    "TurnPrompt",
+    "build_turn_prompt",
+    "protocol_error_feedback",
+]

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

@@ -0,0 +1,582 @@
+"""One employee, one turn: the loop every agent in the platform runs.
+
+``run_turn`` is the whole agent in one function: call the model, parse its
+tags, execute the tools it asked for, feed the results back, repeat — until a
+terminal protocol tag appears, the tool-call cap is hit, the turn times out, or
+somebody cancels the run.
+
+The rules that make it safe to let a model drive work:
+
+* **The employee's model is a granted model.** ``ModelGrantPort.is_granted``
+  is checked *before* the first call; an ungranted model fails with
+  ``error_kind=auth`` and a message for the manager, never a silent retry.
+* **Every model call is billed.** Usage events are written through the same
+  ``record_model_usage`` path ``WorkforceService`` uses, so nothing the engine
+  spends escapes the ledger — manager chat included.
+* **Errors are classified, not swallowed.** ``rate_limit``, ``unavailable``,
+  ``network`` and ``timeout`` are transient and the scheduler retries them;
+  ``auth``, ``model`` and ``protocol`` are not, and they go to the manager's
+  error turn. The vocabulary lives in the domain
+  (``app/domain/workforce/engine.py``), not here.
+* **A cancellation is honoured between steps.** The loop asks the run (and the
+  caller's hook) before every model call and every tool batch, so
+  ``cancel_run`` stops work in progress instead of after it.
+* **The turn leaves a record.** ``engine_turns`` (who ran, what came back,
+  what it cost), ``run_steps`` (the timeline the UI renders) and ``tool_calls``
+  (the executions the approvals UI shows) are all written here.
+
+Only tool tags are executed here. Every other tag (``ASSIGN``, ``COMPLETE``,
+``CREATE_TASK``, ``KNOWLEDGE``…) is a *decision*: the caller interprets it
+against the pipeline, which is what keeps this loop free of orchestration.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass, field
+
+from app.application.engine.events import RunFeed
+from app.application.engine.prompt import (
+    TurnContext,
+    TurnMessage,
+    build_turn_prompt,
+    protocol_error_feedback,
+)
+from app.application.engine.protocol import TOOL_TAGS, Tag, TagName, parse_reply
+from app.application.engine.tools import ToolExecutor, ToolOutcome, tool_key_for
+from app.application.ports import ModelProviderPort
+from app.application.ports.memory import MemoryPort, NullMemory
+from app.application.ports.model_grants import ModelGrantPort
+from app.application.ports.sandbox_provider import SandboxProviderPort
+from app.application.services.workforce_service import record_model_usage
+from app.core.config import settings
+from app.core.errors import ProviderUnavailableError
+from app.domain.ai_control.models import AiModel, RateUnit
+from app.domain.workforce.engine import EngineTurn, ErrorKind, is_transient
+from app.domain.workforce.models import (
+    Employee,
+    RoleKey,
+    Run,
+    RunStatus,
+    Task,
+    ToolCall,
+    ToolCallStatus,
+)
+from app.domain.workforce.pipeline import Mode, Stage
+from app.infrastructure.repositories.ai_control import (
+    AiModelRepository,
+    ModelCredentialRepository,
+    RateCardRepository,
+)
+from app.infrastructure.repositories.audit import UsageLedgerRepository
+from app.infrastructure.repositories.engine import (
+    EngineTurnRepository,
+    PipelineStateRepository,
+)
+from app.infrastructure.repositories.workforce import (
+    RunRepository,
+    RunStepRepository,
+    ToolCallRepository,
+)
+
+#: The tag that ends each role's turn. A role whose reply has none of these
+#: keeps working: the caller decides what a reply without a decision means.
+TERMINAL_TAGS: dict[tuple[RoleKey, Mode], tuple[TagName, ...]] = {
+    (RoleKey.MANAGER, Mode.TRIAGE): (TagName.ASSIGN, TagName.REJECT, TagName.ASK_USER),
+    (RoleKey.MANAGER, Mode.FINAL): (TagName.COMPLETE, TagName.REVISE),
+    (RoleKey.MANAGER, Mode.ERROR): (TagName.RESOLVE_ERROR, TagName.CANCEL),
+    # Chat has no worktree: the manager's reply is consumed by the chat use
+    # case, which is why these are the tags the chat layer looks for.
+    (RoleKey.MANAGER, Mode.CHAT): (
+        TagName.CREATE_TASK,
+        TagName.PAUSE_TASK,
+        TagName.RESUME_TASK,
+        TagName.CANCEL_TASK,
+        TagName.HIRE,
+        TagName.UPDATE_EMPLOYEE,
+        TagName.FIRE,
+    ),
+    (RoleKey.PLANNER, Mode.WORK): (TagName.PLAN_SPEC,),
+    (RoleKey.DEVELOPER, Mode.WORK): (TagName.WORK_REPORT,),
+    (RoleKey.REVIEWER, Mode.WORK): (TagName.REVIEW_PASS, TagName.REVIEW_CHANGES),
+    (RoleKey.TESTER, Mode.WORK): (TagName.TEST_PASS, TagName.TEST_FAIL),
+}
+
+
+@dataclass(slots=True, kw_only=True)
+class TurnResult:
+    """What one turn produced — or why it could not.
+
+    ``retryable`` is the scheduler's answer to "try again or escalate": it
+    reads the domain's transient set, so a rate limit and a bad key are never
+    treated the same way twice.
+    """
+
+    ok: bool
+    error_kind: str = ""
+    error: str = ""
+    reply_text: str = ""
+    tags: tuple[Tag, ...] = ()
+    tool_calls: int = 0
+    model_calls: int = 0
+    #: True when the turn ended on a timeout, a cancellation or the tool cap
+    #: rather than on a protocol decision.
+    incomplete: bool = False
+    turn_id: str = ""
+
+    @property
+    def retryable(self) -> bool:
+        return bool(self.error_kind) and is_transient(self.error_kind)
+
+
+@dataclass(slots=True, kw_only=True)
+class TurnRequest:
+    """Everything ``run_turn`` needs. One object, no hidden state."""
+
+    organization_id: str
+    task: Task
+    role: RoleKey
+    mode: Mode
+    stage: Stage
+    run_id: str
+    employee: Employee | None = None
+    #: The granted model this employee runs on (``employee_profiles.ai_model_id``).
+    model_id: str | None = None
+    #: The worktree this turn works in, as the sandbox sees it.
+    sandbox_root: str = "/workspace"
+    session_external_id: str | None = None
+    extra_context: str = ""
+    history: list[TurnMessage] = field(default_factory=list)
+    #: Checked between steps; returning True ends the turn as cancelled.
+    should_cancel: Callable[[], Awaitable[bool]] | None = None
+    user_id: str | None = None
+
+
+def classify_error(exc: BaseException) -> ErrorKind:
+    """Sort a failure into the engine's error vocabulary.
+
+    The providers raise one broad error type, so the classification reads the
+    message rather than pretending each provider has its own exception tree.
+    A failure that cannot be classified counts as ``unavailable`` (transient):
+    retrying a genuine outage is right, and a permanent failure reveals itself
+    on the second attempt.
+    """
+    if isinstance(exc, asyncio.TimeoutError):
+        return ErrorKind.TIMEOUT
+    text = str(exc).lower()
+    if "timeout" in text or "timed out" in text:
+        return ErrorKind.TIMEOUT
+    if "rate" in text or "429" in text or "quota" in text:
+        return ErrorKind.RATE_LIMIT
+    if any(word in text for word in ("auth", "credential", "unauthor", "401", "403")):
+        return ErrorKind.AUTH
+    if "model" in text:
+        return ErrorKind.MODEL
+    if any(word in text for word in ("connection", "network", "unreachable", "dns")):
+        return ErrorKind.NETWORK
+    return ErrorKind.UNAVAILABLE
+
+
+class TurnRunner:
+    """Runs one turn at a time. One instance per process is enough."""
+
+    def __init__(
+        self,
+        *,
+        provider: ModelProviderPort,
+        models: AiModelRepository,
+        credentials: ModelCredentialRepository,
+        rate_cards: RateCardRepository,
+        ledger: UsageLedgerRepository,
+        grants: ModelGrantPort,
+        sandbox: SandboxProviderPort | None = None,
+        memory: MemoryPort | None = None,
+        turns: EngineTurnRepository | None = None,
+        run_steps: RunStepRepository | None = None,
+        tool_calls: ToolCallRepository | None = None,
+        runs: RunRepository | None = None,
+        states: PipelineStateRepository | None = None,
+        feed: RunFeed | None = None,
+    ) -> None:
+        self.provider = provider
+        self.models = models
+        self.credentials = credentials
+        self.rate_cards = rate_cards
+        self.ledger = ledger
+        self.grants = grants
+        self.sandbox = sandbox
+        self.memory = memory or NullMemory()
+        self.turns = turns
+        self.run_steps = run_steps
+        self.tool_calls = tool_calls
+        self.runs = runs
+        self.states = states
+        self.feed = feed or RunFeed()
+
+    # ------------------------------------------------------------------ #
+
+    async def run_turn(self, request: TurnRequest) -> TurnResult:
+        """Run one employee turn to completion (or to a clean failure)."""
+        started = time.monotonic()
+        deadline = started + settings.engine_turn_timeout_seconds
+        history: list[TurnMessage] = list(request.history)
+        tool_calls_made = 0
+        model_calls = 0
+        reply_text = ""
+        tags: tuple[Tag, ...] = ()
+        turn_row = await self._open_turn_row(request)
+
+        try:
+            model = await self._resolve_model(request)
+        except _ModelUnavailable as exc:
+            return await self._fail(
+                request, turn_row, exc.kind.value, str(exc), tool_calls_made, model_calls
+            )
+
+        while True:
+            if await self._cancelled(request):
+                return await self._cancelled_result(request, turn_row, tool_calls_made, model_calls)
+            if time.monotonic() > deadline:
+                return await self._fail(
+                    request,
+                    turn_row,
+                    ErrorKind.TIMEOUT.value,
+                    "مهلت نوبت تمام شد.",
+                    tool_calls_made,
+                    model_calls,
+                    incomplete=True,
+                )
+
+            prompt = await self._build_prompt(request, history)
+            try:
+                response = await self._complete(request, model, prompt)
+            except ProviderUnavailableError as exc:
+                return await self._fail(
+                    request,
+                    turn_row,
+                    classify_error(exc).value,
+                    str(exc),
+                    tool_calls_made,
+                    model_calls,
+                )
+            model_calls += 1
+            reply_text = response.content
+            self.feed.log(
+                run_id=request.run_id,
+                task_id=request.task.id,
+                organization_id=request.organization_id,
+                line=f"{request.role.value}/{request.mode.value}: {response.content[:400]}",
+            )
+
+            parsed = parse_reply(response.content, role=request.role, mode=request.mode)
+            if parsed.errors:
+                # The model learns the protocol from what it just got wrong.
+                history.append(
+                    TurnMessage(kind="result", text=protocol_error_feedback(parsed.errors))
+                )
+                self.feed.log(
+                    run_id=request.run_id,
+                    task_id=request.task.id,
+                    organization_id=request.organization_id,
+                    line=f"خطای پروتکل: {parsed.errors[0]}",
+                )
+                await self.feed.flush(run_id=request.run_id)
+                continue
+
+            requested_tools = [tag for tag in parsed.tags if tag.name in TOOL_TAGS]
+            tags = self._terminal_tags(request, parsed.tags)
+            if not requested_tools:
+                # Nothing left to execute: the reply is a decision, and the
+                # caller reads it out of ``TurnResult.tags``.
+                break
+
+            for tag in requested_tools:
+                if tool_calls_made >= settings.engine_max_tool_calls_per_turn:
+                    return await self._fail(
+                        request,
+                        turn_row,
+                        ErrorKind.PROTOCOL.value,
+                        "سقف فراخوانی ابزار در این نوبت پر شد.",
+                        tool_calls_made,
+                        model_calls,
+                        incomplete=True,
+                    )
+                if await self._cancelled(request):
+                    return await self._cancelled_result(
+                        request, turn_row, tool_calls_made, model_calls
+                    )
+                outcome = await self._execute_tool(request, tag)
+                tool_calls_made += 1
+                history.append(
+                    TurnMessage(
+                        kind="result", text=f"[[{outcome.tool}]] → {outcome.output}"
+                    )
+                )
+                await self._step(
+                    request,
+                    kind="tool",
+                    title=f"{outcome.tool}: {'موفق' if outcome.ok else 'ناموفق'}",
+                    payload={"tool": outcome.tool, "ok": outcome.ok},
+                )
+            await self.feed.flush(run_id=request.run_id)
+
+        await self.feed.flush(run_id=request.run_id, force=True)
+        await self._close_turn_row(
+            turn_row, ok=True, tool_calls=tool_calls_made, reply=reply_text, tags=tags
+        )
+        return TurnResult(
+            ok=True,
+            reply_text=reply_text,
+            tags=tags,
+            tool_calls=tool_calls_made,
+            model_calls=model_calls,
+            turn_id=turn_row.id if turn_row else "",
+        )
+
+    # -- model ------------------------------------------------------------- #
+
+    async def _resolve_model(self, request: TurnRequest) -> AiModel:
+        """The employee's granted model, or a clean auth failure."""
+        model_id = request.model_id
+        if not model_id:
+            raise _ModelUnavailable(
+                ErrorKind.AUTH, "برای این کارمند مدلی انتخاب نشده است."
+            )
+        granted = await self.grants.is_granted(
+            organization_id=request.organization_id, ai_model_id=model_id
+        )
+        if not granted:
+            raise _ModelUnavailable(
+                ErrorKind.AUTH,
+                "مدل این کارمند به سازمان شما اختصاص داده نشده است.",
+            )
+        model = await self.models.get(model_id)
+        if model is None or not model.is_active:
+            raise _ModelUnavailable(ErrorKind.MODEL, "مدل این کارمند موجود نیست.")
+        return model
+
+    async def _complete(self, request: TurnRequest, model: AiModel, prompt):
+        """One model call, billed exactly like every other model call."""
+        pool = await self.credentials.list_active_for_provider(model.provider_id)
+        credential = pool[0] if pool else None
+        response = await self.provider.complete(
+            model=model,
+            credential=credential,
+            credentials=pool or None,
+            messages=prompt.messages,
+            system=prompt.system,
+        )
+        for unit, tokens in (
+            (RateUnit.MODEL_INPUT_TOKENS, float(response.input_tokens)),
+            (RateUnit.MODEL_OUTPUT_TOKENS, float(response.output_tokens)),
+        ):
+            await record_model_usage(
+                ledger=self.ledger,
+                rate_cards=self.rate_cards,
+                organization_id=request.organization_id,
+                unit=unit,
+                quantity=tokens,
+                project_id=request.task.project_id,
+                user_id=request.user_id,
+                employee_id=request.employee.id if request.employee else None,
+                run_id=request.run_id,
+            )
+        return response
+
+    # -- tools ------------------------------------------------------------- #
+
+    async def _execute_tool(self, request: TurnRequest, tag: Tag) -> ToolOutcome:
+        """Run one tool tag in the sandbox and record the call."""
+        if self.sandbox is None or not request.session_external_id:
+            return ToolOutcome(
+                tool=tag.name.value,
+                ok=False,
+                output="سندباکسی برای این نوبت در دسترس نیست.",
+                refused=True,
+            )
+        executor = ToolExecutor(
+            sandbox=self.sandbox,
+            session_external_id=request.session_external_id,
+            root=request.sandbox_root,
+        )
+        outcome = await executor.execute(tag)
+        await self._record_tool_call(request, tag, outcome)
+        await self.feed.tool(
+            run_id=request.run_id,
+            task_id=request.task.id,
+            organization_id=request.organization_id,
+            tool=outcome.tool,
+            ok=outcome.ok,
+            summary=outcome.output,
+        )
+        return outcome
+
+    async def _record_tool_call(self, request: TurnRequest, tag: Tag, outcome: ToolOutcome) -> None:
+        """Persist the execution so the approvals UI shows real work."""
+        if self.tool_calls is None:
+            return
+        call = await self.tool_calls.create(
+            ToolCall(
+                run_id=request.run_id,
+                tool_key=tool_key_for(tag),
+                arguments_json={"tag": tag.name.value, "attrs": dict(tag.attrs)},
+                result_json={"ok": outcome.ok, "output": outcome.output[:4000]},
+                status=ToolCallStatus.SUCCEEDED if outcome.ok else ToolCallStatus.FAILED,
+                requires_approval=False,
+            )
+        )
+        outcome.tool_call_id = call.id
+
+    # -- records ----------------------------------------------------------- #
+
+    async def _open_turn_row(self, request: TurnRequest):
+        prompt = await self._build_prompt(request, list(request.history))
+        if self.turns is None:
+            return None
+        return await self.turns.create(
+            EngineTurn(
+                task_id=request.task.id,
+                run_id=request.run_id,
+                employee_id=request.employee.id if request.employee else None,
+                role=request.role,
+                mode=request.mode,
+                stage=request.stage,
+                prompt_chars=prompt.total_chars,
+            )
+        )
+
+    async def _close_turn_row(
+        self, turn_row, *, ok: bool, tool_calls: int, reply: str, tags: tuple[Tag, ...]
+    ) -> None:
+        if turn_row is None or self.turns is None:
+            return
+        counts: dict[str, int] = {}
+        for tag in tags:
+            counts[tag.name.value] = counts.get(tag.name.value, 0) + 1
+        turn_row.ok = ok
+        turn_row.tool_calls = tool_calls
+        turn_row.reply_excerpt = reply[: settings.engine_reply_excerpt_chars]
+        turn_row.tags = counts
+        await self.turns.finish(
+            turn_row.id,
+            ok=ok,
+            tool_calls=tool_calls,
+            error_kind="",
+            reply_excerpt=turn_row.reply_excerpt,
+            tags=counts,
+        )
+
+    async def _fail(
+        self,
+        request: TurnRequest,
+        turn_row,
+        error_kind: str,
+        message: str,
+        tool_calls: int,
+        model_calls: int,
+        *,
+        incomplete: bool = False,
+    ) -> TurnResult:
+        await self.feed.flush(run_id=request.run_id, force=True)
+        if turn_row is not None and self.turns is not None:
+            turn_row.ok = False
+            await self.turns.finish(
+                turn_row.id,
+                ok=False,
+                tool_calls=tool_calls,
+                error_kind=error_kind or ErrorKind.UNAVAILABLE.value,
+            )
+        await self._step(
+            request,
+            kind="error",
+            title="نوبت ناموفق بود.",
+            payload={"error_kind": error_kind, "message": message[:2000]},
+        )
+        return TurnResult(
+            ok=False,
+            error_kind=error_kind,
+            error=message,
+            tool_calls=tool_calls,
+            model_calls=model_calls,
+            incomplete=incomplete,
+            turn_id=turn_row.id if turn_row else "",
+        )
+
+    async def _cancelled_result(
+        self, request: TurnRequest, turn_row, tool_calls: int, model_calls: int
+    ) -> TurnResult:
+        return await self._fail(
+            request,
+            turn_row,
+            ErrorKind.CANCELLED.value,
+            "اجرا لغو شد.",
+            tool_calls,
+            model_calls,
+            incomplete=True,
+        )
+
+    # -- helpers ----------------------------------------------------------- #
+
+    async def _build_prompt(self, request: TurnRequest, history: list[TurnMessage]):
+        memory_context = await self.memory.build_context(
+            organization_id=request.organization_id,
+            role=request.role.value,
+            query=f"{request.task.title}\n{request.task.description}",
+            budget_chars=2500,
+        )
+        return build_turn_prompt(
+            TurnContext(
+                organization_id=request.organization_id,
+                task_id=request.task.id,
+                project_id=request.task.project_id,
+                task_title=request.task.title,
+                task_description=request.task.description,
+                stage=request.stage,
+                role=request.role,
+                mode=request.mode,
+                memory_context=memory_context,
+                stage_context=request.extra_context,
+                history=history,
+            )
+        )
+
+    async def _cancelled(self, request: TurnRequest) -> bool:
+        """Ask the caller's hook and the run itself whether to stop."""
+        if request.should_cancel is not None and await request.should_cancel():
+            return True
+        if self.runs is None:
+            return False
+        run: Run | None = await self.runs.get(request.run_id)
+        return run is not None and run.status in (
+            RunStatus.CANCELLED,
+            RunStatus.FAILED,
+            RunStatus.TIMED_OUT,
+        )
+
+    async def _step(
+        self, request: TurnRequest, *, kind: str, title: str, payload: dict
+    ) -> None:
+        """Append to the run timeline the UI renders."""
+        if self.run_steps is None:
+            return
+        await self.run_steps.append(
+            run_id=request.run_id, kind=kind, title=title, payload=payload
+        )
+
+    @staticmethod
+    def _terminal_tags(request: TurnRequest, tags: tuple[Tag, ...]) -> tuple[Tag, ...]:
+        wanted = TERMINAL_TAGS.get((request.role, request.mode), ())
+        return tuple(tag for tag in tags if tag.name in wanted)
+
+
+class _ModelUnavailable(Exception):
+    """The employee's model is missing or not granted to this organization."""
+
+    def __init__(self, kind: ErrorKind, message: str) -> None:
+        super().__init__(message)
+        self.kind = kind
+
+
+__all__ = ["TERMINAL_TAGS", "TurnRequest", "TurnResult", "TurnRunner", "classify_error"]

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

@@ -0,0 +1,279 @@
+"""Executing the toolkit inside the sandbox — and refusing what it must.
+
+Four rules, all enforced here rather than at the call sites:
+
+1. **Path safety.** Every path is resolved against the worktree's sandbox root.
+   Absolute paths that escape it, ``..`` traversal and anything under ``.git``
+   are refused. A symlink that points outside the repository is not followed:
+   the sandbox is the security boundary, and this is the second lock.
+2. **The AI never touches git.** ``git push``, ``git remote`` and ``git config``
+   are refused with a message that says why — the platform does git through
+   ``RepoWorkspacePort`` (PLAN-v2 D6), never the model.
+3. **Bounded commands.** ``RUN`` has a timeout (default 120 s, hard max 600 s)
+   and its output is capped to the last ``engine_tool_output_chars``
+   characters, which is what the model needs anyway: a traceback is at the end.
+4. **Every call is recorded.** Each execution writes a ``tool_calls`` row with
+   ``succeeded``/``failed``, so the approvals UI finally shows real executions
+   instead of intentions.
+
+The sandbox is reached only through ``SandboxProviderPort``; there is no local
+shell-out path.
+"""
+
+from __future__ import annotations
+
+import shlex
+from dataclasses import dataclass
+from pathlib import PurePosixPath
+
+from app.application.engine.protocol import EditBlock, Tag, TagName, edit_blocks_of
+from app.application.ports import CommandResult, SandboxProviderPort
+from app.core.config import settings
+from app.domain.workforce.tools import ToolKey
+
+#: Commands the model may never run. The message is the point: a refusal the
+#: model can read is a refusal it works around once.
+_FORBIDDEN_GIT: tuple[tuple[str, str], ...] = (
+    ("push", "git push"),
+    ("remote", "git remote"),
+    ("config", "git config"),
+)
+_GIT_REFUSAL = (
+    "دستور git رد شد: عملیات git را پلتفرم انجام می‌دهد، نه هوش مصنوعی."
+)
+
+#: RUN defaults and ceilings. A model asking for "no timeout" gets the ceiling.
+RUN_DEFAULT_TIMEOUT = 120
+RUN_MAX_TIMEOUT = 600
+RUN_MAX_COMMAND_CHARS = 4000
+
+#: Which protocol tag maps to which row of the existing tool_calls catalog.
+_TOOL_KEY: dict[TagName, str] = {
+    TagName.READ_FILE: ToolKey.SANDBOX_READ_FILE,
+    TagName.WRITE_FILE: ToolKey.SANDBOX_WRITE_FILE,
+    TagName.EDIT_FILE: ToolKey.SANDBOX_WRITE_FILE,
+    TagName.LIST_DIR: ToolKey.SANDBOX_RUN_COMMAND,
+    TagName.SEARCH: ToolKey.SANDBOX_RUN_COMMAND,
+    TagName.RUN: ToolKey.SANDBOX_RUN_COMMAND,
+}
+
+
+class UnsafePathError(ValueError):
+    """The requested path escapes the worktree."""
+
+
+class GitRefusedError(ValueError):
+    """The model tried to run git itself."""
+
+
+@dataclass(slots=True, kw_only=True)
+class ToolOutcome:
+    """One tool execution: what happened, in a form the model can read."""
+
+    tool: str
+    ok: bool
+    output: str
+    refused: bool = False
+    #: The row key this call was recorded under, for the approvals UI.
+    tool_key: str = ""
+    tool_call_id: str = ""
+
+
+def safe_join(root: str, relative: str) -> str:
+    """Resolve ``relative`` inside ``root`` or raise :class:`UnsafePathError`.
+
+    Pure and synchronous on purpose: it is also the unit under test, and a path
+    bug must be reproducible without a container.
+    """
+    if not relative or not relative.strip():
+        raise UnsafePathError("مسیر خالی است.")
+    candidate = relative.strip()
+    if candidate.startswith(("/", "~")):
+        raise UnsafePathError(f"مسیر مطلق مجاز نیست: {candidate}")
+    parts = PurePosixPath(candidate).parts
+    if any(part == ".." for part in parts):
+        raise UnsafePathError(f"خروج از ریشهٔ مخزن مجاز نیست: {candidate}")
+    if parts and parts[0] == ".git":
+        raise UnsafePathError("دسترسی به .git مجاز نیست.")
+    base = PurePosixPath(root.rstrip("/") or "/")
+    return str(base / candidate.lstrip("/"))
+
+
+def _tail(text: str, *, limit: int) -> str:
+    """Keep the *last* ``limit`` characters: a traceback is at the end."""
+    if len(text) <= limit:
+        return text
+    return f"[… {len(text) - limit} کاراکتر ابتدایی حذف شد …]\n{text[-limit:]}"
+
+
+def _clamp_timeout(raw: str) -> int:
+    try:
+        value = int(float(raw)) if raw else RUN_DEFAULT_TIMEOUT
+    except ValueError:
+        value = RUN_DEFAULT_TIMEOUT
+    return max(1, min(value, RUN_MAX_TIMEOUT))
+
+
+def _check_command(command: str) -> None:
+    if len(command) > RUN_MAX_COMMAND_CHARS:
+        raise ValueError("دستور بیش از حد طولانی است.")
+    tokens = shlex.split(command, comments=False, posix=True)
+    if not tokens:
+        raise ValueError("دستور خالی است.")
+    if tokens[0] != "git":
+        return
+    for sub, label in _FORBIDDEN_GIT:
+        if sub in tokens[1:]:
+            raise GitRefusedError(f"{_GIT_REFUSAL} («{label}»)")
+
+
+def _shell_quote(value: str) -> str:
+    return shlex.quote(value)
+
+
+class ToolExecutor:
+    """Runs one parsed tool tag inside the worktree's sandbox session."""
+
+    def __init__(
+        self,
+        *,
+        sandbox: SandboxProviderPort,
+        session_external_id: str,
+        root: str,
+        max_output_chars: int | None = None,
+    ) -> None:
+        self.sandbox = sandbox
+        self.session_external_id = session_external_id
+        self.root = root
+        self.max_output_chars = (
+            settings.engine_tool_output_chars if max_output_chars is None else max_output_chars
+        )
+
+    async def execute(self, tag: Tag) -> ToolOutcome:
+        """Execute one tool tag. Never raises: failures come back as output."""
+        handler = _HANDLERS.get(tag.name)
+        if handler is None:  # pragma: no cover - the matrix forbids it
+            return ToolOutcome(
+                tool=tag.name.value, ok=False, output="این ابزار در این نقش مجاز نیست."
+            )
+        try:
+            output = await handler(self, tag)
+        except UnsafePathError as exc:
+            return ToolOutcome(
+                tool=tag.name.value, ok=False, output=str(exc), refused=True
+            )
+        except GitRefusedError as exc:
+            return ToolOutcome(
+                tool=tag.name.value, ok=False, output=str(exc), refused=True
+            )
+        except FileNotFoundError as exc:
+            return ToolOutcome(tool=tag.name.value, ok=False, output=str(exc))
+        except OSError as exc:
+            return ToolOutcome(
+                tool=tag.name.value, ok=False, output=f"خطای سندباکس: {exc}"
+            )
+        except ValueError as exc:
+            return ToolOutcome(tool=tag.name.value, ok=False, output=str(exc))
+        return ToolOutcome(
+            tool=tag.name.value, ok=True, output=_tail(output, limit=self.max_output_chars)
+        )
+
+    # -- sandbox primitives -------------------------------------------------
+
+    async def _run(self, command: str, *, timeout: int) -> str:
+        _check_command(command)
+        result: CommandResult = await self.sandbox.run_command(
+            session_external_id=self.session_external_id,
+            command=command,
+            timeout_seconds=timeout,
+            max_output_bytes=settings.engine_tool_output_chars * 4,
+        )
+        text = result.stdout or ""
+        if result.stderr:
+            text = f"{text}\n{result.stderr}" if text else result.stderr
+        return f"$ {command}\n(exit {result.exit_code})\n{text}".strip()
+
+    async def _read(self, path: str) -> str:
+        return await self.sandbox.read_file(
+            session_external_id=self.session_external_id, path=safe_join(self.root, path)
+        )
+
+    async def _write(self, path: str, content: str) -> None:
+        await self.sandbox.write_file(
+            session_external_id=self.session_external_id,
+            path=safe_join(self.root, path),
+            content=content,
+        )
+
+    # -- the six tools ------------------------------------------------------
+
+    async def _read_file(self, tag: Tag) -> str:
+        path = tag.require("path")
+        content = await self._read(path)
+        return f"# {path}\n{content}"
+
+    async def _write_file(self, tag: Tag) -> str:
+        path = tag.require("path")
+        await self._write(path, tag.body)
+        return f"نوشته شد: {path} ({len(tag.body)} کاراکتر)"
+
+    async def _edit_file(self, tag: Tag) -> str:
+        path = tag.require("path")
+        blocks: list[EditBlock] = edit_blocks_of(tag)
+        content = await self._read(path)
+        for block in blocks:
+            occurrences = content.count(block.old)
+            if occurrences != 1:
+                raise ValueError(
+                    f"بلوک قدیمی باید دقیقاً یک بار پیدا شود؛ {occurrences} بار پیدا شد."
+                )
+            content = content.replace(block.old, block.new, 1)
+        await self._write(path, content)
+        return f"ویرایش شد: {path} ({len(blocks)} بلوک)"
+
+    async def _list_dir(self, tag: Tag) -> str:
+        path = tag.attr("path", ".") or "."
+        target = safe_join(self.root, path)
+        return await self._run(f"ls -la -- {_shell_quote(target)}", timeout=30)
+
+    async def _search(self, tag: Tag) -> str:
+        pattern = tag.require("pattern")
+        path = tag.attr("path", ".") or "."
+        target = safe_join(self.root, path)
+        command = f"grep -RIn -e {_shell_quote(pattern)} -- {_shell_quote(target)}"
+        # grep exits 1 when nothing matched, which is an answer, not a failure.
+        return await self._run(command, timeout=60)
+
+    async def _run_command(self, tag: Tag) -> str:
+        command = tag.attr("cmd", "") or tag.body.strip()
+        if not command:
+            raise ValueError("دستور خالی است.")
+        timeout = _clamp_timeout(tag.attr("timeout", ""))
+        return await self._run(command, timeout=timeout)
+
+
+_HANDLERS = {
+    TagName.READ_FILE: ToolExecutor._read_file,
+    TagName.WRITE_FILE: ToolExecutor._write_file,
+    TagName.EDIT_FILE: ToolExecutor._edit_file,
+    TagName.LIST_DIR: ToolExecutor._list_dir,
+    TagName.SEARCH: ToolExecutor._search,
+    TagName.RUN: ToolExecutor._run_command,
+}
+
+
+def tool_key_for(tag: Tag) -> str:
+    """The ``tool_calls`` row key a protocol tag maps to."""
+    return _TOOL_KEY.get(tag.name, ToolKey.SANDBOX_RUN_COMMAND)
+
+
+__all__ = [
+    "GitRefusedError",
+    "RUN_DEFAULT_TIMEOUT",
+    "RUN_MAX_TIMEOUT",
+    "ToolExecutor",
+    "ToolOutcome",
+    "UnsafePathError",
+    "safe_join",
+    "tool_key_for",
+]

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

@@ -102,6 +102,58 @@ _SYSTEM_PROMPT = (
 )
 
 
+async def record_model_usage(
+    *,
+    ledger: UsageLedgerRepository,
+    rate_cards: RateCardRepository,
+    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:
+    """Write one usage event for a model call, and push it live.
+
+    **Every** model call goes through here — the run engine, the manager chat
+    and the engine's turn runner alike. Billing reads the internal ledger and
+    never the provider's invoice, so a call that is not recorded here is a
+    call the platform never bills for.
+
+    The pricing rule: a missing rate card is a misconfiguration, not a free
+    lunch, so the placeholder price is recorded and flagged as unpriced rather
+    than silently charging zero.
+    """
+    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(
+        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},
+        )
+    )
+    # 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(
+        organization_id=organization_id,
+        unit=unit.value if hasattr(unit, "value") else str(unit),
+        quantity=quantity,
+        run_id=run_id,
+        employee_id=employee_id,
+    )
+
+
 class WorkforceService:
     def __init__(
         self,
@@ -928,25 +980,17 @@ class WorkforceService:
         employee_id: str | None = None,
         run_id: str | None = None,
     ) -> None:
-        card = await self.rate_cards.get_by_unit(unit)
+        await record_model_usage(
-        unit_price = card.customer_price if card else _UNPRICED
+            ledger=self.ledger,
-        internal_cost = card.internal_cost_estimate if card else 0.0
+            rate_cards=self.rate_cards,
-        await self.ledger.record(
-            usage=UsageRecord(
             organization_id=organization_id,
+            unit=unit,
+            quantity=quantity,
             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},
         )
-        )
-        # 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(
             organization_id=organization_id,

apps/api/app/core/config.py

@@ -148,6 +148,17 @@ class Settings(BaseSettings):
     engine_max_cycles: int = 6
     engine_max_employees: int = 20
     engine_tick_seconds: float = 10.0
+    #: Hard cap on a turn's total prompt characters. The builder truncates the
+    #: oldest tool results first (never the protocol, never the task brief).
+    engine_prompt_char_cap: int = 24000
+    #: How much of one tool's output is fed back to the model, and how much of
+    #: a reply is stored in ``engine_turns.reply_excerpt``.
+    engine_tool_output_chars: int = 8000
+    engine_reply_excerpt_chars: int = 4000
+    #: Real-time log batching (PLAN-v2 §5): one event per stream per interval,
+    #: at most this many lines per event.
+    engine_log_flush_seconds: float = 0.5
+    engine_log_max_lines_per_event: int = 200
     # <<< agent-2
 
     # >>> agent-3 (delivery & infra): add settings ONLY between these markers

apps/api/app/domain/workforce/engine.py

@@ -33,6 +33,7 @@ class ErrorKind(StrEnum):
     MODEL = "model"
     PROTOCOL = "protocol"
     BUDGET = "budget"
+    CANCELLED = "cancelled"
 
 
 TRANSIENT_ERROR_KINDS: frozenset[ErrorKind] = frozenset(

apps/api/app/infrastructure/repositories/engine.py

@@ -408,14 +408,26 @@ class EngineTurnRepository:
         return [_turn(r) for r in (await self.session.execute(stmt)).scalars()]
 
     async def finish(
-        self, turn_id: str, *, ok: bool, tool_calls: int, error_kind: str = ""
+        self,
+        turn_id: str,
+        *,
+        ok: bool,
+        tool_calls: int,
+        error_kind: str = "",
+        reply_excerpt: str | None = None,
+        tags: dict | None = None,
     ) -> None:
+        """Close a turn: outcome, tool count and the bounded reply excerpt."""
         row = await self.session.get(EngineTurnRow, turn_id)
         if row is None:
             return
         row.ok = ok
         row.tool_calls = tool_calls
         row.error_kind = error_kind
+        if reply_excerpt is not None:
+            row.reply_excerpt = reply_excerpt
+        if tags is not None:
+            row.tags = dict(tags)
         row.finished_at = utcnow()
         await self.session.flush()
 

apps/api/tests/engine_fakes.py

@@ -0,0 +1,242 @@
+"""Fakes for the engine tests: a scripted model and a scripted sandbox.
+
+Two rules the engine tests insist on: the model never answers by calling a
+real provider, and nothing ever executes on the host. Both fakes also *record*
+what they were given, because half of what the engine must guarantee is about
+prompts (no secret ever reaches one) and commands (git never runs).
+
+``ScriptedModel.complete`` returns the queued responses in order and raises
+the queued exceptions, so a test can script "read a file, run the tests, report"
+exactly the way a real agent would behave — including the failures.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from app.application.ports import ChatMessage, CommandResult, ModelHealth, ModelResponse
+from app.application.ports.model_grants import GrantedModel
+from app.application.ports.sandbox_provider import (
+    SessionInfo,
+    SnapshotInfo,
+    WorkspaceInfo,
+)
+
+
+@dataclass(slots=True, kw_only=True)
+class ScriptedReply:
+    """One queued model answer: text, or an exception to raise instead."""
+
+    content: str = ""
+    raises: BaseException | None = None
+    input_tokens: int = 10
+    output_tokens: int = 5
+
+
+class ScriptedModel:
+    """A ``ModelProviderPort`` that replays a script.
+
+    ``seen`` holds every call it received (system prompt + messages), so a test
+    can assert what the model was actually told — the no-secrets rule is only
+    testable this way.
+    """
+
+    def __init__(self, replies: list[ScriptedReply] | None = None) -> None:
+        self.replies: list[ScriptedReply] = list(replies or [])
+        self.seen: list[dict] = []
+        self.calls = 0
+
+    def queue(self, content: str, **kwargs) -> None:
+        self.replies.append(ScriptedReply(content=content, **kwargs))
+
+    async def complete(
+        self,
+        *,
+        model,
+        messages: list[ChatMessage],
+        system: str = "",
+        temperature: float = 0.2,
+        max_tokens: int = 2048,
+        credential=None,
+        credentials=None,
+    ) -> ModelResponse:
+        self.calls += 1
+        self.seen.append(
+            {
+                "model": model,
+                "system": system,
+                "messages": list(messages),
+                "credential": credential,
+            }
+        )
+        if not self.replies:
+            return ModelResponse(
+                content="[[WORK_REPORT]]پاسخ پیش‌فرض[[/WORK_REPORT]]",
+                model_internal_name=getattr(model, "internal_name", ""),
+                input_tokens=1,
+                output_tokens=1,
+            )
+        reply = self.replies.pop(0)
+        if reply.raises is not None:
+            raise reply.raises
+        return ModelResponse(
+            content=reply.content,
+            model_internal_name=getattr(model, "internal_name", ""),
+            input_tokens=reply.input_tokens,
+            output_tokens=reply.output_tokens,
+        )
+
+    async def health(self) -> ModelHealth:
+        return ModelHealth(ok=True, detail="scripted")
+
+    @property
+    def prompts(self) -> list[str]:
+        """Every system prompt the model was given, in order."""
+        return [call["system"] for call in self.seen]
+
+    def all_text(self) -> str:
+        """Everything the model saw, for the "no secret" assertion."""
+        parts = [call["system"] for call in self.seen]
+        for call in self.seen:
+            parts.extend(message.content for message in call["messages"])
+        return "\n".join(parts)
+
+
+class FakeSandbox:
+    """A ``SandboxProviderPort`` over a dict filesystem and scripted commands.
+
+    ``commands`` maps the command *prefix* to a queue of results, so a test can
+    make the tests red once and green afterwards: the engine must notice.
+    """
+
+    def __init__(
+        self,
+        *,
+        files: dict[str, str] | None = None,
+        commands: dict[str, list[CommandResult]] | None = None,
+    ) -> None:
+        self.files: dict[str, str] = dict(files or {})
+        self.commands: dict[str, list[CommandResult]] = {
+            key: list(value) for key, value in (commands or {}).items()
+        }
+        self.executed: list[dict] = []
+        self.writes: list[dict] = []
+        self.reads: list[str] = []
+
+    # -- SandboxProviderPort ------------------------------------------------
+
+    async def create_workspace(
+        self, *, organization_id: str, project_id: str, template_id: str, image: str
+    ) -> WorkspaceInfo:
+        return WorkspaceInfo(external_id="ws-1", status="ready")
+
+    async def start_session(self, *, workspace_external_id: str) -> SessionInfo:
+        return SessionInfo(external_id="sess-1", status="running")
+
+    async def run_command(
+        self,
+        *,
+        session_external_id: str,
+        command: str,
+        timeout_seconds: int,
+        max_output_bytes: int,
+        env: dict[str, str] | None = None,
+    ) -> CommandResult:
+        self.executed.append(
+            {
+                "command": command,
+                "timeout_seconds": timeout_seconds,
+                "max_output_bytes": max_output_bytes,
+                "session": session_external_id,
+            }
+        )
+        for prefix, queue in self.commands.items():
+            if command.startswith(prefix) and queue:
+                return queue.pop(0)
+        return CommandResult(exit_code=0, stdout="ok")
+
+    async def read_file(self, *, session_external_id: str, path: str) -> str:
+        self.reads.append(path)
+        if path not in self.files:
+            raise FileNotFoundError(path)
+        return self.files[path]
+
+    async def write_file(self, *, session_external_id: str, path: str, content: str) -> None:
+        self.files[path] = content
+        self.writes.append({"path": path, "content": content})
+
+    async def snapshot(self, *, session_external_id: str, name: str) -> SnapshotInfo:
+        return SnapshotInfo(external_id="snap-1", name=name)
+
+    async def pause(self, *, session_external_id: str) -> None:
+        return None
+
+    async def resume(self, *, session_external_id: str) -> SessionInfo:
+        return SessionInfo(external_id="sess-1", status="running")
+
+    async def destroy(self, *, session_external_id: str) -> None:
+        return None
+
+    async def get_preview_url(self, *, session_external_id: str, port: int) -> str:
+        return f"http://127.0.0.1:{port}"
+
+
+class FakeGrants:
+    """A ``ModelGrantPort`` with an explicit allow-list."""
+
+    def __init__(self, granted: list[str] | None = None) -> None:
+        self._granted = list(granted or ["model-strong", "model-fast"])
+        self.checks: list[str] = []
+
+    async def granted_models(self, *, organization_id: str) -> list[GrantedModel]:
+        return [
+            GrantedModel(ai_model_id=model_id, tier="balanced")
+            for model_id in self._granted
+        ]
+
+    async def is_granted(self, *, organization_id: str, ai_model_id: str) -> bool:
+        self.checks.append(ai_model_id)
+        return ai_model_id in self._granted
+
+    def grant(self, model_id: str) -> None:
+        self._granted.append(model_id)
+
+
+class RecordingMemory:
+    """A ``MemoryPort`` that records its queries and returns fixed context."""
+
+    def __init__(self, context: str = "") -> None:
+        self.context = context
+        self.queries: list[dict] = []
+
+    async def build_context(
+        self, *, organization_id: str, role: str, query: str, budget_chars: int = 2500
+    ) -> str:
+        self.queries.append(
+            {"organization_id": organization_id, "role": role, "query": query}
+        )
+        return self.context
+
+    async def record(
+        self,
+        *,
+        organization_id: str,
+        kind: str,
+        content: str,
+        source: str,
+        evidence_task_id: str | None = None,
+        key: str | None = None,
+    ) -> str | None:
+        return None
+
+    async def on_task_finished(self, *, organization_id: str, task) -> None:
+        return None
+
+
+__all__ = [
+    "FakeGrants",
+    "FakeSandbox",
+    "RecordingMemory",
+    "ScriptedModel",
+    "ScriptedReply",
+]

apps/api/tests/test_engine_prompt.py

@@ -0,0 +1,154 @@
+"""The prompt builder: a stable prefix, a bounded tail.
+
+What these tests defend:
+
+* the prefix order is the plan's order (rules → protocol → toolkit → app
+  profile → memory), because that order is what makes prompt caching useful
+* the task brief is capped, and the history truncation drops the **oldest tool
+  results first** — a model that forgets the rules is worse than one that
+  forgot what it read three calls ago
+* nothing secret reaches the prompt: the builder only ever sees the strings it
+  is handed, and the runner never hands it settings
+"""
+
+from __future__ import annotations
+
+from app.application.engine.prompt import (
+    PLATFORM_RULES,
+    TOOLKIT_DESCRIPTION,
+    TurnContext,
+    TurnMessage,
+    build_turn_prompt,
+)
+from app.core.config import settings
+from app.domain.workforce.models import RoleKey
+from app.domain.workforce.pipeline import Mode, Stage
+
+
+def _ctx(**kwargs) -> TurnContext:
+    base = {
+        "organization_id": "org-1",
+        "task_id": "task-1",
+        "project_id": "proj-1",
+        "task_title": "ساخت صفحهٔ ورود",
+        "task_description": "توضیح کار",
+        "stage": Stage.PENDING_IMPLEMENTATION,
+        "role": RoleKey.DEVELOPER,
+        "mode": Mode.WORK,
+        "app_profile_text": "زبان: پایتون\nدستور تست: pytest -q",
+        "memory_context": "این سازمان تست‌ها را با pytest اجرا می‌کند.",
+    }
+    base.update(kwargs)
+    return TurnContext(**base)
+
+
+def test_the_prefix_keeps_the_plan_order():
+    prompt = build_turn_prompt(_ctx())
+    system = prompt.system
+    positions = [
+        system.index(PLATFORM_RULES.splitlines()[0]),  # 1. platform rules
+        system.index("تو برنامه‌نویس هستی"),  # 2. the role protocol
+        system.index(TOOLKIT_DESCRIPTION.splitlines()[0]),  # 3. the toolkit
+        system.index("دستور تست: pytest -q"),  # 4. the app profile
+        system.index("این سازمان"),  # 5. org memory
+    ]
+    assert positions == sorted(positions)
+    assert prompt.messages[-1].role == "user"  # 6/7. brief and history last
+
+
+def test_the_toolkit_is_only_shown_to_roles_that_have_tools():
+    developer = build_turn_prompt(_ctx())
+    manager = build_turn_prompt(
+        _ctx(role=RoleKey.MANAGER, mode=Mode.TRIAGE, stage=Stage.PENDING_MANAGER_REVIEW)
+    )
+    assert "READ_FILE" in developer.system
+    assert "READ_FILE" not in manager.system
+
+
+def test_the_task_brief_is_capped():
+    prompt = build_turn_prompt(_ctx(task_description="x" * 5000))
+    brief = prompt.messages[-1].content
+    # The description is cut, the title survives.
+    assert "ساخت صفحهٔ ورود" in brief
+    assert len(brief) < 2000
+
+
+def test_stage_context_only_goes_to_the_role_that_needs_it():
+    """A reviewer sees the diff; a developer sees the spec; nobody sees both."""
+    developer = build_turn_prompt(
+        _ctx(spec="برنامهٔ برنامه‌ریز", diff_text="diff --git a/x b/x")
+    )
+    assert "برنامهٔ برنامه‌ریز" in developer.messages[-1].content
+    assert "diff --git" not in developer.messages[-1].content
+
+    reviewer = build_turn_prompt(
+        _ctx(
+            role=RoleKey.REVIEWER,
+            mode=Mode.WORK,
+            stage=Stage.PENDING_REVIEW,
+            spec="برنامهٔ برنامه‌ریز",
+            diff_text="diff --git a/x b/x",
+        )
+    )
+    assert "diff --git" in reviewer.messages[-1].content
+    assert "برنامهٔ برنامه‌ریز" not in reviewer.messages[-1].content
+
+
+def test_history_drops_the_oldest_results_first():
+    big = "y" * 4000
+    history = [
+        TurnMessage(kind="result", text=big),  # oldest: dropped first
+        TurnMessage(kind="result", text=big),
+        TurnMessage(kind="result", text="نتیجهٔ تازه و مهم"),
+    ]
+    settings.engine_prompt_char_cap = 9000  # keep the prefix small
+    try:
+        prompt = build_turn_prompt(_ctx(history=list(history)))
+    finally:
+        settings.engine_prompt_char_cap = 24000
+    joined = "\n".join(m.content for m in prompt.messages)
+    assert "نتیجهٔ تازه و مهم" in joined
+    assert joined.count("y" * 4000) <= 1
+    assert prompt.total_chars <= 9000 + 4000
+
+
+def test_a_tiny_cap_still_keeps_the_rules_and_the_brief():
+    settings.engine_prompt_char_cap = 3000
+    try:
+        prompt = build_turn_prompt(
+            _ctx(history=[TurnMessage(kind="result", text="z" * 20000)])
+        )
+    finally:
+        settings.engine_prompt_char_cap = 24000
+    # The prefix and the brief survive; only history space is sacrificed.
+    assert "قوانین پلتفرم" in prompt.system
+    assert "ساخت صفحهٔ ورود" in prompt.messages[-1].content
+
+
+def test_the_prompt_never_contains_the_settings(monkeypatch):
+    monkeypatch.setattr(settings, "jwt_secret", "top-secret")
+    monkeypatch.setattr(settings, "database_url", "sqlite:///private.db")
+    prompt = build_turn_prompt(
+        _ctx(history=[TurnMessage(kind="result", text="نتیجه")])
+    )
+    text = prompt.system + "\n" + "\n".join(m.content for m in prompt.messages)
+    assert "top-secret" not in text
+    assert "private.db" not in text
+    assert PLATFORM_RULES.splitlines()[0] in text
+
+
+def test_manager_modes_get_their_own_protocol_hint():
+    triage = build_turn_prompt(
+        _ctx(role=RoleKey.MANAGER, mode=Mode.TRIAGE, stage=Stage.PENDING_MANAGER_REVIEW)
+    )
+    final = build_turn_prompt(
+        _ctx(role=RoleKey.MANAGER, mode=Mode.FINAL, stage=Stage.PENDING_FINAL_REVIEW)
+    )
+    assert "ASSIGN" in triage.system
+    assert "ASSIGN" not in final.system
+    assert "COMPLETE" in final.system
+
+
+def test_the_toolkit_description_lists_every_tool_tag():
+    for tool in ("READ_FILE", "WRITE_FILE", "EDIT_FILE", "LIST_DIR", "SEARCH", "RUN"):
+        assert tool in TOOLKIT_DESCRIPTION

apps/api/tests/test_engine_runner.py

@@ -0,0 +1,551 @@
+"""The turn runner: one employee turn, with a scripted model and a fake sandbox.
+
+What these tests defend:
+
+* the write → run → report loop really works: a file is written inside the
+  sandbox, a command runs there, and the report ends the turn
+* the sandbox is the boundary: a path escape, a ``.git`` path and a git
+  command are refused *inside the tool*, with a message the model can read
+* a turn is bounded: the tool-call cap and the turn timeout both stop it
+* errors are classified, so a rate limit is retried later and an ungranted
+  model is not
+* every model call writes usage events — including the ones a manager chat
+  makes, which agybot forgot
+* no secret from the settings ever reaches a prompt
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import select
+
+from app.application.engine.runner import (
+    TurnRequest,
+    TurnRunner,
+    classify_error,
+)
+from app.application.ports import CommandResult
+from app.core.config import settings
+from app.core.errors import ProviderUnavailableError
+from app.domain.ai_control.models import AiModel
+from app.domain.base import new_id
+from app.domain.workforce.engine import ErrorKind
+from app.domain.workforce.models import RoleKey
+from app.domain.workforce.pipeline import Mode, Stage
+from app.infrastructure.db.models import (
+    AiProviderRow,
+    EmployeeRow,
+    OrganizationRow,
+    RunRow,
+    TaskRow,
+    UsageEventRow,
+    UserRow,
+)
+from app.infrastructure.repositories.ai_control import (
+    AiModelRepository,
+    ModelCredentialRepository,
+    RateCardRepository,
+)
+from app.infrastructure.repositories.audit import UsageLedgerRepository
+from app.infrastructure.repositories.engine import (
+    EngineTurnRepository,
+    PipelineStateRepository,
+)
+from app.infrastructure.repositories.workforce import (
+    EmployeeRepository,
+    RunRepository,
+    RunStepRepository,
+    TaskRepository,
+    ToolCallRepository,
+)
+from tests.engine_fakes import (
+    FakeGrants,
+    FakeSandbox,
+    RecordingMemory,
+    ScriptedModel,
+    ScriptedReply,
+)
+
+DEVELOPER = {
+    "role": RoleKey.DEVELOPER,
+    "mode": Mode.WORK,
+    "stage": Stage.PENDING_IMPLEMENTATION,
+}
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest_asyncio.fixture
+async def _db(session_factory, seeded):
+    """A tenant, a task, a run, a developer employee and a granted model."""
+    async with session_factory() as session:
+        user = UserRow(email=f"runner-{new_id()[:8]}@example.com", password_hash="x")
+        session.add(user)
+        await session.flush()
+        org = OrganizationRow(name="سازمان اجرا", slug=f"run-{user.id[:6]}", owner_user_id=user.id)
+        session.add(org)
+        await session.flush()
+        # The seeded provider row: the model must hang off a real one.
+        provider_id = (await session.execute(select(AiProviderRow.id).limit(1))).scalar_one()
+
+        model = await AiModelRepository(session).create(
+            AiModel(
+                provider_id=provider_id,
+                internal_name="scripted-model",
+                display_name_for_admin="مدل فیلم‌نویسی‌شده",
+                capabilities=["text", "code"],
+                is_active=True,
+            )
+        )
+        task = TaskRow(organization_id=org.id, title="ساخت صفحهٔ ورود")
+        session.add(task)
+        await session.flush()
+        run = RunRow(task_id=task.id)
+        session.add(run)
+        employee = EmployeeRow(organization_id=org.id, name="سارا", role_key="developer")
+        session.add(employee)
+        await session.commit()
+        yield SimpleNamespace(
+            session_factory=session_factory,
+            organization_id=org.id,
+            task_id=task.id,
+            run_id=run.id,
+            employee_id=employee.id,
+            model_id=model.id,
+        )
+
+
+def _runner(session, *, model, sandbox, grants, feed=None) -> TurnRunner:
+    return TurnRunner(
+        provider=model,
+        models=AiModelRepository(session),
+        credentials=ModelCredentialRepository(session),
+        rate_cards=RateCardRepository(session),
+        ledger=UsageLedgerRepository(session),
+        grants=grants,
+        sandbox=sandbox,
+        memory=RecordingMemory(context="این سازمان تست‌ها را با pytest اجرا می‌کند."),
+        turns=EngineTurnRepository(session),
+        run_steps=RunStepRepository(session),
+        tool_calls=ToolCallRepository(session),
+        runs=RunRepository(session),
+        states=PipelineStateRepository(session),
+        feed=feed,
+    )
+
+
+async def _request(db, *, sandbox_root="/workspace", session_id="sess-1", **kwargs) -> TurnRequest:
+    async with db.session_factory() as session:
+        task = await TaskRepository(session).get(db.task_id)
+        employee = await EmployeeRepository(session).get(db.employee_id)
+    return TurnRequest(
+        organization_id=db.organization_id,
+        task=task,
+        run_id=db.run_id,
+        employee=employee,
+        model_id=db.model_id,
+        sandbox_root=sandbox_root,
+        session_external_id=session_id,
+        **kwargs,
+    )
+
+
+# ---------------------------------------------------------------------------
+# The loop
+# ---------------------------------------------------------------------------
+
+
+async def test_write_then_run_then_report_ends_the_turn(_db, session_factory):
+    """The whole agent loop: one file, one command, one report."""
+    scripted = ScriptedModel(
+        [
+            ScriptedReply(
+                content='[[WRITE_FILE path="app/login.py"]]print("login")[[/WRITE_FILE]]'
+            ),
+            ScriptedReply(content='[[RUN cmd="pytest -q" timeout="30"/]]'),
+            ScriptedReply(content="[[WORK_REPORT]]صفحهٔ ورود نوشده و تست‌ها سبز شد.[[/WORK_REPORT]]"),
+        ]
+    )
+    sandbox = FakeSandbox(
+        commands={"pytest": [CommandResult(exit_code=0, stdout="3 passed")]},
+    )
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=sandbox, grants=FakeGrants([_db.model_id])
+        )
+        request = await _request(_db, **DEVELOPER)
+        result = await runner.run_turn(request)
+        await session.commit()
+
+    assert result.ok is True
+    assert result.model_calls == 3
+    assert result.tool_calls == 2
+    assert result.tags[0].name.value == "WORK_REPORT"
+    # The file really was written inside the sandbox, rooted at the worktree.
+    assert sandbox.writes == [{"path": "/workspace/app/login.py", "content": 'print("login")'}]
+    assert [call["command"] for call in sandbox.executed] == ["pytest -q"]
+    assert scripted.calls == 3
+
+
+async def test_tool_results_feed_back_into_the_next_prompt(_db, session_factory):
+    """The model must see what its own command printed."""
+    scripted = ScriptedModel(
+        [
+            ScriptedReply(content='[[RUN cmd="pytest -q"/]]'),
+            ScriptedReply(content="[[WORK_REPORT]]تمام[[/WORK_REPORT]]"),
+        ]
+    )
+    sandbox = FakeSandbox(
+        commands={"pytest": [CommandResult(exit_code=1, stdout="1 failed", stderr="boom")]}
+    )
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=sandbox, grants=FakeGrants([_db.model_id])
+        )
+        result = await runner.run_turn(await _request(_db, **DEVELOPER))
+        await session.commit()
+
+    assert result.ok is True
+    last_prompt = scripted.seen[-1]
+    feedback = "\n".join(m.content for m in last_prompt["messages"])
+    assert "1 failed" in feedback and "boom" in feedback
+
+
+async def test_each_tool_call_writes_a_tool_calls_row(_db, session_factory):
+    """The approvals UI finally sees real executions, not intentions."""
+    scripted = ScriptedModel(
+        [
+            ScriptedReply(content='[[READ_FILE path="app/x.py"/]]'),
+            ScriptedReply(content="[[WORK_REPORT]]تمام[[/WORK_REPORT]]"),
+        ]
+    )
+    sandbox = FakeSandbox(files={"/workspace/app/x.py": "print(1)"})
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=sandbox, grants=FakeGrants([_db.model_id])
+        )
+        await runner.run_turn(await _request(_db, **DEVELOPER))
+        await session.commit()
+        calls = await ToolCallRepository(session).list_for_run(_db.run_id)
+
+    assert len(calls) == 1
+    assert calls[0].tool_key == "sandbox.read_file"
+    assert calls[0].status.value == "succeeded"
+    assert calls[0].result_json["ok"] is True
+
+
+async def test_a_failed_tool_is_recorded_as_failed(_db, session_factory):
+    scripted = ScriptedModel(
+        [
+            ScriptedReply(content='[[READ_FILE path="app/missing.py"/]]'),
+            ScriptedReply(content="[[WORK_REPORT]]تمام[[/WORK_REPORT]]"),
+        ]
+    )
+    sandbox = FakeSandbox()
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=sandbox, grants=FakeGrants([_db.model_id])
+        )
+        result = await runner.run_turn(await _request(_db, **DEVELOPER))
+        await session.commit()
+        calls = await ToolCallRepository(session).list_for_run(_db.run_id)
+
+    assert result.ok is True
+    assert calls[0].status.value == "failed"
+
+
+async def test_the_turn_leaves_an_engine_turns_row(_db, session_factory):
+    scripted = ScriptedModel([ScriptedReply(content="[[WORK_REPORT]]تمام[[/WORK_REPORT]]")])
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=FakeSandbox(), grants=FakeGrants([_db.model_id])
+        )
+        result = await runner.run_turn(await _request(_db, **DEVELOPER))
+        await session.commit()
+        turns = await EngineTurnRepository(session).list_for_task(_db.task_id)
+
+    assert result.turn_id
+    assert len(turns) == 1
+    assert turns[0].ok is True
+    assert turns[0].role is RoleKey.DEVELOPER
+    assert turns[0].mode is Mode.WORK
+    assert turns[0].stage is Stage.PENDING_IMPLEMENTATION
+    assert turns[0].prompt_chars > 0
+    assert "[[WORK_REPORT]]" in turns[0].reply_excerpt
+    assert "تمام" in turns[0].reply_excerpt
+    assert turns[0].tags == {"WORK_REPORT": 1}
+    assert turns[0].finished_at is not None
+
+
+# ---------------------------------------------------------------------------
+# The sandbox boundary
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "path",
+    ["../../etc/passwd", "/etc/passwd", ".git/config", "app/../../secret"],
+)
+async def test_a_path_escape_is_refused(_db, session_factory, path):
+    scripted = ScriptedModel(
+        [
+            ScriptedReply(content=f'[[READ_FILE path="{path}"/]]'),
+            ScriptedReply(content="[[WORK_REPORT]]تمام[[/WORK_REPORT]]"),
+        ]
+    )
+    sandbox = FakeSandbox(files={"/etc/passwd": "root"})
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=sandbox, grants=FakeGrants([_db.model_id])
+        )
+        result = await runner.run_turn(await _request(_db, **DEVELOPER))
+        await session.commit()
+
+    # The refusal is fed back to the model, and nothing was read.
+    assert result.ok is True
+    assert sandbox.reads == []
+    assert "مجاز نیست" in scripted.all_text()
+
+
+@pytest.mark.parametrize(
+    "command", ["git push origin dev", "git remote add x y", "git config user.email a@b.c"]
+)
+async def test_the_model_cannot_run_git(_db, session_factory, command):
+    scripted = ScriptedModel(
+        [
+            ScriptedReply(content=f'[[RUN cmd="{command}"/]]'),
+            ScriptedReply(content="[[WORK_REPORT]]تمام[[/WORK_REPORT]]"),
+        ]
+    )
+    sandbox = FakeSandbox()
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=sandbox, grants=FakeGrants([_db.model_id])
+        )
+        result = await runner.run_turn(await _request(_db, **DEVELOPER))
+        await session.commit()
+
+    assert result.ok is True
+    assert sandbox.executed == []
+    assert "پلتفرم" in scripted.all_text()
+
+
+async def test_edit_file_applies_a_search_replace_block(_db, session_factory):
+    scripted = ScriptedModel(
+        [
+            ScriptedReply(
+                content=(
+                    "[[EDIT_FILE path=\"app/x.py\"]]<<<OLD\nold line\n===\n"
+                    "new line\nNEW>>>[[/EDIT_FILE]]"
+                )
+            ),
+            ScriptedReply(content="[[WORK_REPORT]]تمام[[/WORK_REPORT]]"),
+        ]
+    )
+    sandbox = FakeSandbox(files={"/workspace/app/x.py": "before\nold line\nafter"})
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=sandbox, grants=FakeGrants([_db.model_id])
+        )
+        result = await runner.run_turn(await _request(_db, **DEVELOPER))
+        await session.commit()
+
+    assert result.ok is True
+    assert sandbox.files["/workspace/app/x.py"] == "before\nnew line\nafter"
+
+
+async def test_edit_file_fails_when_the_old_text_is_not_unique(_db, session_factory):
+    scripted = ScriptedModel(
+        [
+            ScriptedReply(
+                content=(
+                    "[[EDIT_FILE path=\"app/x.py\"]]<<<OLD\nline\n===\n"
+                    "other\nNEW>>>[[/EDIT_FILE]]"
+                )
+            ),
+            ScriptedReply(content="[[WORK_REPORT]]تمام[[/WORK_REPORT]]"),
+        ]
+    )
+    sandbox = FakeSandbox(files={"/workspace/app/x.py": "line\nline"})
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=sandbox, grants=FakeGrants([_db.model_id])
+        )
+        result = await runner.run_turn(await _request(_db, **DEVELOPER))
+        await session.commit()
+
+    assert result.ok is True
+    assert sandbox.files["/workspace/app/x.py"] == "line\nline"
+    assert "دقیقاً یک بار" in scripted.all_text()
+
+
+# ---------------------------------------------------------------------------
+# Bounds: the tool cap and the timeout
+# ---------------------------------------------------------------------------
+
+
+async def test_the_tool_call_cap_stops_a_runaway_turn(_db, session_factory, monkeypatch):
+    monkeypatch.setattr(settings, "engine_max_tool_calls_per_turn", 2)
+    scripted = ScriptedModel([ScriptedReply(content='[[LIST_DIR path="."/]]') for _ in range(5)])
+    sandbox = FakeSandbox()
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=sandbox, grants=FakeGrants([_db.model_id])
+        )
+        result = await runner.run_turn(await _request(_db, **DEVELOPER))
+        await session.commit()
+
+    assert result.ok is False
+    assert result.incomplete is True
+    assert result.tool_calls == 2
+    assert result.error_kind == "protocol"
+
+
+async def test_the_turn_timeout_stops_a_slow_loop(_db, session_factory, monkeypatch):
+    monkeypatch.setattr(settings, "engine_turn_timeout_seconds", 0)
+    scripted = ScriptedModel([ScriptedReply(content='[[LIST_DIR path="."/]]')] * 3)
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=FakeSandbox(), grants=FakeGrants([_db.model_id])
+        )
+        result = await runner.run_turn(await _request(_db, **DEVELOPER))
+        await session.commit()
+
+    assert result.ok is False
+    assert result.error_kind == "timeout"
+
+
+async def test_a_cancelled_run_stops_between_steps(_db, session_factory):
+    scripted = ScriptedModel(
+        [
+            ScriptedReply(content='[[LIST_DIR path="."/]]'),
+            ScriptedReply(content="[[WORK_REPORT]]تمام[[/WORK_REPORT]]"),
+        ]
+    )
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=FakeSandbox(), grants=FakeGrants([_db.model_id])
+        )
+        request = await _request(_db, **DEVELOPER)
+        # The hook cancels after the first model call has been made.
+        calls = {"n": 0}
+
+        async def should_cancel() -> bool:
+            calls["n"] += 1
+            return calls["n"] > 1
+
+        request.should_cancel = should_cancel
+        result = await runner.run_turn(request)
+        await session.commit()
+
+    assert result.ok is False
+    assert result.error_kind == "cancelled"
+
+
+# ---------------------------------------------------------------------------
+# Errors: transient versus fatal
+# ---------------------------------------------------------------------------
+
+
+async def test_a_rate_limit_is_transient_and_reported(_db, session_factory):
+    scripted = ScriptedModel([ScriptedReply(raises=ProviderUnavailableError("429 rate limited"))])
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=FakeSandbox(), grants=FakeGrants([_db.model_id])
+        )
+        result = await runner.run_turn(await _request(_db, **DEVELOPER))
+        await session.commit()
+
+    assert result.ok is False
+    assert result.error_kind == "rate_limit"
+    assert result.retryable is True
+
+
+async def test_an_ungranted_model_is_refused_before_any_call(_db, session_factory):
+    scripted = ScriptedModel([ScriptedReply(content="[[WORK_REPORT]]تمام[[/WORK_REPORT]]")])
+    grants = FakeGrants(["some-other-model"])
+    async with session_factory() as session:
+        runner = _runner(session, model=scripted, sandbox=FakeSandbox(), grants=grants)
+        result = await runner.run_turn(await _request(_db, **DEVELOPER))
+        await session.commit()
+
+    assert result.ok is False
+    assert result.error_kind == "auth"
+    assert result.retryable is False
+    # The model was never called: an ungranted employee spends nothing.
+    assert scripted.calls == 0
+    assert grants.checks == [_db.model_id]
+
+
+def test_error_classification():
+    assert classify_error(ProviderUnavailableError("429 rate limit")) is ErrorKind.RATE_LIMIT
+    assert classify_error(ProviderUnavailableError("auth failed")) is ErrorKind.AUTH
+    assert classify_error(ProviderUnavailableError("connection refused")) is ErrorKind.NETWORK
+    assert classify_error(TimeoutError()) is ErrorKind.TIMEOUT
+    assert classify_error(ProviderUnavailableError("boom")) is ErrorKind.UNAVAILABLE
+
+
+# ---------------------------------------------------------------------------
+# Billing and secrets
+# ---------------------------------------------------------------------------
+
+
+async def test_every_model_call_writes_usage_events(_db, session_factory):
+    """agybot forgot the manager chat; the engine does not."""
+    scripted = ScriptedModel(
+        [
+            ScriptedReply(content='[[LIST_DIR path="."/]]', input_tokens=11, output_tokens=7),
+            ScriptedReply(
+                content="[[WORK_REPORT]]تمام[[/WORK_REPORT]]", input_tokens=13, output_tokens=9
+            ),
+        ]
+    )
+    from sqlalchemy import select
+
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=FakeSandbox(), grants=FakeGrants([_db.model_id])
+        )
+        await runner.run_turn(await _request(_db, **DEVELOPER))
+        await session.commit()
+        rows = (
+            (await session.execute(select(UsageEventRow).where(UsageEventRow.run_id == _db.run_id)))
+            .scalars()
+            .all()
+        )
+
+    assert len(rows) == 4  # two calls, input and output each
+    assert sum(r.quantity for r in rows if r.unit == "model_input_tokens") == 24
+    assert sum(r.quantity for r in rows if r.unit == "model_output_tokens") == 16
+
+
+async def test_no_secret_from_settings_reaches_the_model(_db, session_factory, monkeypatch):
+    """The prompt is built by the engine; the settings stay on the server."""
+    monkeypatch.setattr(settings, "jwt_secret", "super-secret-jwt-value")
+    monkeypatch.setattr(settings, "database_url", "postgresql+asyncpg://u:p@h/db")
+    scripted = ScriptedModel(
+        [
+            ScriptedReply(content='[[READ_FILE path="app/x.py"/]]'),
+            ScriptedReply(content="[[WORK_REPORT]]تمام[[/WORK_REPORT]]"),
+        ]
+    )
+    sandbox = FakeSandbox(files={"/workspace/app/x.py": "print(1)"})
+    async with session_factory() as session:
+        runner = _runner(
+            session, model=scripted, sandbox=sandbox, grants=FakeGrants([_db.model_id])
+        )
+        await runner.run_turn(await _request(_db, **DEVELOPER))
+        await session.commit()
+
+    seen = scripted.all_text()
+    assert "super-secret-jwt-value" not in seen
+    assert "postgresql+asyncpg://u:p@h/db" not in seen
+    # The model still got the rules and the role it plays.
+    assert "قوانین پلتفرم" in seen
+    assert "WORK_REPORT" in seen

apps/api/tests/test_engine_tools.py

@@ -0,0 +1,184 @@
+"""The toolkit's guards: path safety, the git refusal and bounded commands.
+
+These are the checks that run *before* anything reaches the sandbox, so they
+are tested as pure functions wherever possible — a path bug must be
+reproducible without a container.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.application.engine.protocol import Tag, TagName, edit_blocks_of
+from app.application.engine.tools import (
+    RUN_DEFAULT_TIMEOUT,
+    RUN_MAX_TIMEOUT,
+    ToolExecutor,
+    UnsafePathError,
+    safe_join,
+    tool_key_for,
+)
+from tests.engine_fakes import FakeSandbox
+
+ROOT = "/workspace/.worktrees/task-1"
+
+
+# ---------------------------------------------------------------------------
+# Path safety
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "relative",
+    ["app/main.py", "./app/main.py", "app/deep/nested/file.py", "a.txt"],
+)
+def test_a_normal_relative_path_is_allowed(relative):
+    assert safe_join(ROOT, relative) == f"{ROOT}/{relative.lstrip('./')}".replace(
+        "/./", "/"
+    )
+
+
+@pytest.mark.parametrize(
+    "relative",
+    ["/etc/passwd", "~/.ssh/id_rsa", "../../etc/passwd", "app/../../secrets", ".git/config"],
+)
+def test_an_escaping_path_is_refused(relative):
+    with pytest.raises(UnsafePathError):
+        safe_join(ROOT, relative)
+
+
+def test_an_empty_path_is_refused():
+    with pytest.raises(UnsafePathError):
+        safe_join(ROOT, "   ")
+
+
+def test_the_git_directory_is_never_readable():
+    with pytest.raises(UnsafePathError):
+        safe_join(ROOT, ".git/objects/ab/cdef")
+
+
+# ---------------------------------------------------------------------------
+# The git refusal
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "command",
+    ["git push origin dev", "git remote add origin x", "git config user.email a@b.c"],
+)
+async def test_git_commands_are_refused_before_the_sandbox(command):
+    sandbox = FakeSandbox()
+    executor = ToolExecutor(
+        sandbox=sandbox, session_external_id="sess-1", root=ROOT
+    )
+    outcome = await executor.execute(
+        Tag(name=TagName.RUN, attrs={"cmd": command}, body="")
+    )
+    assert outcome.ok is False
+    assert outcome.refused is True
+    assert "پلتفرم" in outcome.output
+    assert sandbox.executed == []
+
+
+async def test_other_git_commands_are_allowed_through():
+    """Only push/remote/config are refused: the platform owns the rest."""
+    sandbox = FakeSandbox(commands={"git": []})
+    executor = ToolExecutor(
+        sandbox=sandbox, session_external_id="sess-1", root=ROOT
+    )
+    outcome = await executor.execute(
+        Tag(name=TagName.RUN, attrs={"cmd": "git status --porcelain -uall"}, body="")
+    )
+    assert outcome.ok is True
+    assert len(sandbox.executed) == 1
+
+
+# ---------------------------------------------------------------------------
+# Bounds
+# ---------------------------------------------------------------------------
+
+
+async def test_run_clamps_the_timeout_to_the_ceiling():
+    sandbox = FakeSandbox()
+    executor = ToolExecutor(sandbox=sandbox, session_external_id="sess-1", root=ROOT)
+    await executor.execute(
+        Tag(name=TagName.RUN, attrs={"cmd": "sleep 1", "timeout": "99999"}, body="")
+    )
+    assert sandbox.executed[0]["timeout_seconds"] == RUN_MAX_TIMEOUT
+
+
+async def test_run_defaults_to_a_sane_timeout():
+    sandbox = FakeSandbox()
+    executor = ToolExecutor(sandbox=sandbox, session_external_id="sess-1", root=ROOT)
+    await executor.execute(Tag(name=TagName.RUN, attrs={"cmd": "ls"}, body=""))
+    assert sandbox.executed[0]["timeout_seconds"] == RUN_DEFAULT_TIMEOUT
+
+
+async def test_a_huge_output_is_truncated_to_its_tail():
+    sandbox = FakeSandbox(
+        commands={"noisy": [type("R", (), {"exit_code": 0, "stdout": "x" * 30000, "stderr": ""})()]}
+    )
+    executor = ToolExecutor(
+        sandbox=sandbox, session_external_id="sess-1", root=ROOT, max_output_chars=1000
+    )
+    outcome = await executor.execute(
+        Tag(name=TagName.RUN, attrs={"cmd": "noisy"}, body="")
+    )
+    assert outcome.ok is True
+    assert len(outcome.output) <= 1100
+    assert outcome.output.endswith("x" * 100)
+
+
+# ---------------------------------------------------------------------------
+# The tools themselves
+# ---------------------------------------------------------------------------
+
+
+async def test_write_then_read_round_trips_inside_the_root():
+    sandbox = FakeSandbox()
+    executor = ToolExecutor(sandbox=sandbox, session_external_id="sess-1", root=ROOT)
+    written = await executor.execute(
+        Tag(name=TagName.WRITE_FILE, attrs={"path": "app/a.py"}, body="print(1)")
+    )
+    read = await executor.execute(
+        Tag(name=TagName.READ_FILE, attrs={"path": "app/a.py"}, body="")
+    )
+    assert written.ok and read.ok
+    assert "print(1)" in read.output
+    assert sandbox.writes[0]["path"] == f"{ROOT}/app/a.py"
+
+
+async def test_list_dir_and_search_run_quoted_commands():
+    sandbox = FakeSandbox()
+    executor = ToolExecutor(sandbox=sandbox, session_external_id="sess-1", root=ROOT)
+    await executor.execute(Tag(name=TagName.LIST_DIR, attrs={"path": "app"}, body=""))
+    await executor.execute(
+        Tag(name=TagName.SEARCH, attrs={"pattern": "TODO", "path": "app"}, body="")
+    )
+    assert sandbox.executed[0]["command"].startswith("ls -la -- ")
+    assert "grep -RIn -e TODO" in sandbox.executed[1]["command"]
+
+
+async def test_a_missing_file_reports_not_found_without_raising():
+    sandbox = FakeSandbox()
+    executor = ToolExecutor(sandbox=sandbox, session_external_id="sess-1", root=ROOT)
+    outcome = await executor.execute(
+        Tag(name=TagName.READ_FILE, attrs={"path": "app/none.py"}, body="")
+    )
+    assert outcome.ok is False
+    assert "none.py" in outcome.output
+
+
+def test_tool_keys_map_onto_the_existing_catalog():
+    assert tool_key_for(Tag(name=TagName.READ_FILE, attrs={}, body="")) == (
+        "sandbox.read_file"
+    )
+    assert tool_key_for(Tag(name=TagName.RUN, attrs={}, body="")) == (
+        "sandbox.run_command"
+    )
+
+
+def test_edit_blocks_refuse_a_malformed_body():
+    tag = Tag(name=TagName.EDIT_FILE, attrs={"path": "a"}, body="just text")
+    with pytest.raises(ValueError):
+        edit_blocks_of(tag)

docs/plan/reports/agent-2-phase-2.md

@@ -0,0 +1,145 @@
+# Agent 2 — Phase 2 report: the agent turn runner (prompt, tools, loop)
+
+**Date:** 2026-10-14 · **Branch:** `dev` · **Status:** done, all checks green
+
+## What was done
+
+### 2.1 `app/application/engine/prompt.py` (new)
+
+`build_turn_prompt(TurnContext) -> TurnPrompt` builds the **stable prefix** in
+exactly the plan's order, because that order is what makes prompt caching
+worth anything:
+
+1. `PLATFORM_RULES` — five short platform rules (never leaks model/infra detail,
+   git belongs to the platform, ask a human when in doubt).
+2. The role protocol — the allowed-tag list for `(role, mode)` plus a worked
+   example from `_PROTOCOL_HINTS` (manager triage / final / error / chat,
+   planner, developer, reviewer, tester).
+3. `TOOLKIT_DESCRIPTION` — the six tool tags with the rules that matter
+   (relative paths, git refused, output clipped). Shown **only** to roles that
+   actually have tools: the manager never sees a file tool.
+4. The app profile text (language/framework/commands — the caller passes text,
+   never the row, so no secret can leak in).
+5. `MemoryPort.build_context(org, role, query=task text, budget_chars=2500)`.
+6. The task brief, capped at `MAX_TASK_BRIEF_CHARS = 1200`, plus the
+   **stage-specific** context: the planner's spec for the developer, **the
+   diff only** for the reviewer, the test output for the tester, and
+   spec+diff+tests for the manager's final verdict. A reviewer never sees the
+   whole tree.
+7. The in-turn conversation (tool results), newest last.
+
+Hard cap `engine_prompt_char_cap` (new setting, 24,000 chars): when exceeded the
+builder drops the **oldest tool results first**, then clips from the front as a
+last resort. The rules, memory and brief are never the things that get cut.
+
+### 2.2 `app/application/engine/tools.py` (new)
+
+`ToolExecutor.execute(tag) -> ToolOutcome` runs the toolkit **inside the
+sandbox** through `SandboxProviderPort`, rooted at `WorktreeRef.sandbox_path`:
+
+- `safe_join()` — pure, synchronous, unit-tested: rejects absolute paths, `..`
+  traversal, `~`, and anything under `.git`.
+- `READ_FILE` / `WRITE_FILE` (body = content) / `EDIT_FILE`
+  (`<<<OLD … === … NEW>>>` blocks; OLD must appear **exactly once**, otherwise a
+  clear failure instead of a half-applied edit).
+- `LIST_DIR` and `SEARCH` run as quoted `ls` / `grep -RIn` commands.
+- `RUN cmd timeout` — timeout defaults to 120 s and is clamped to 600 s; the
+  output is capped to the **last** `engine_tool_output_chars` (8,000) because a
+  traceback is at the end. `git push`, `git remote` and `git config` are refused
+  with a Persian message that says the platform does git.
+- Every execution writes a `tool_calls` row (mapped onto the existing
+  `ToolKey` catalog: read/write/run) with `succeeded`/`failed`, so the approvals
+  UI finally shows real executions rather than intentions.
+- No failure raises: a refusal comes back as tool output the model reads.
+
+### 2.3 `app/application/engine/runner.py` (new)
+
+`TurnRunner.run_turn(TurnRequest) -> TurnResult` — the loop: model call → parse
+tags → execute **tool** tags → feed results back → repeat until a terminal tag,
+the tool-call cap (`engine_max_tool_calls_per_turn`), the turn timeout
+(`engine_turn_timeout_seconds`) or a cancellation.
+
+- The employee's model comes from `request.model_id` (the future
+  `employee_profiles.ai_model_id`) and is checked against `ModelGrantPort` **before
+  the first call**; ungranted or missing → `error_kind=auth`, zero model calls,
+  zero spend.
+- Every model call writes usage events through the **same** code path the run
+  engine uses: `record_model_usage` was extracted from
+  `WorkforceService._record_usage` into a module-level function in
+  `workforce_service.py` (my file) so there is exactly one billing path — the
+  manager chat will use it too (agybot bug).
+- `classify_error()` maps provider failures onto the domain vocabulary:
+  `rate_limit` / `unavailable` / `network` / `timeout` are transient
+  (`TurnResult.retryable`), `auth` / `model` / `protocol` are not.
+- Cancellation is checked between steps, both through the caller's
+  `should_cancel` hook and by re-reading the run row.
+- Records written: `engine_turns` (role, mode, stage, prompt chars, reply
+  excerpt ≤ 4,000, tag counts, tool calls, ok/error_kind), `run_steps`
+  (timeline entries for tools and errors) and `tool_calls`.
+- `TERMINAL_TAGS` names the decision tag of each role/mode; **only tool tags are
+  executed here** — `ASSIGN`, `COMPLETE`, `CREATE_TASK`, `KNOWLEDGE`… are
+  decisions the caller interprets, which keeps orchestration out of the loop.
+
+### `app/application/engine/events.py` (new)
+
+`RunFeed` implements PLAN-v2 §5's batching for `run.log`, plus `run.tool` and
+`task.gate` (which have no typed emitter in the frozen realtime port, so they go
+through the generic envelope): at most one log event per 500 ms per stream, at
+most 200 lines per event, with a forced flush at the end of a turn.
+
+### New settings (inside the agent-2 block of `config.py`)
+
+`engine_prompt_char_cap`, `engine_tool_output_chars`, `engine_reply_excerpt_chars`,
+`engine_log_flush_seconds`, `engine_log_max_lines_per_event`.
+
+## Tests
+
+- `tests/engine_fakes.py` (new): `ScriptedModel` (queued replies **and** queued
+  exceptions, records every system prompt and message it was given),
+  `FakeSandbox` (dict filesystem + scripted command results per command prefix,
+  records every execution), `FakeGrants`, `RecordingMemory`.
+- `tests/test_engine_runner.py` (22 tests): write → run → report; tool results
+  fed back into the next prompt; a `tool_calls` row per execution (succeeded and
+  failed); the `engine_turns` row; path escapes refused (4 shapes); git refused
+  (3 commands, nothing executed); EDIT_FILE applied and refused when not unique;
+  the tool-call cap; the turn timeout; cancellation between steps; rate limit
+  transient; **ungranted model refused before any call**; error classification;
+  usage events per model call; no settings secret in any prompt.
+- `tests/test_engine_prompt.py` (11 tests): the prefix order; toolkit only for
+  tool roles; brief cap; stage context per role; oldest-results-dropped-first
+  truncation; tiny-cap keeps rules and brief; no secrets; per-mode protocol hints.
+- `tests/test_engine_tools.py` (11 tests): path safety matrix; git refusal;
+  timeout clamp and default; output tail truncation; round trip; quoted
+  list/search; tool-key mapping; malformed edit body.
+
+## Checks (PLAN-v2 §7.4)
+
+```
+cd apps/api && .venv/bin/python -m pytest -q   → 648 passed, 1 skipped (54 new)
+cd apps/api && .venv/bin/ruff check .          → All checks passed!
+cd apps/api && .venv/bin/mypy app              → Success: no issues found in 148 source files
+```
+
+No web files were touched, no migration was needed (no schema change).
+
+## Deviations & notes
+
+1. `WorkforceService._record_usage` now delegates to the new module-level
+   `record_model_usage`; behaviour is identical and the existing 14 workforce
+   tests still pass. This keeps one billing code path instead of a copy.
+2. The runner takes the sandbox session id from the request
+   (`session_external_id`) and fails cleanly when there is none. Wiring the real
+   session (workspace container per app, SandboxService) belongs to Phase 4 with
+   the scheduler; deliberately not faked here.
+3. `TERMINAL_TAGS` treats the manager-chat tags as the chat layer's tags: the
+   chat turn emits decisions, not work.
+
+## Open questions / CHANGE REQUESTs
+
+- None. Expected next: org automation policy behind `ModelGrantPort` (Phase 4).
+
+## Next
+
+Phase 3 — manager chat (orders → tasks → team), triage/final/error turns,
+`KNOWLEDGE` → `MemoryPort.record`, `WorkforceBootstrapPort.bootstrap_team`, and
+`tests/test_engine_manager.py`.