diff --git a/apps/api/app/application/engine/events.py b/apps/api/app/application/engine/events.py new file mode 100644 index 0000000..78f7344 --- /dev/null +++ b/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"] diff --git a/apps/api/app/application/engine/prompt.py b/apps/api/app/application/engine/prompt.py new file mode 100644 index 0000000..b8cadb6 --- /dev/null +++ b/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"]]<<>>[[/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", +] diff --git a/apps/api/app/application/engine/runner.py b/apps/api/app/application/engine/runner.py new file mode 100644 index 0000000..9fcca0f --- /dev/null +++ b/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"] diff --git a/apps/api/app/application/engine/tools.py b/apps/api/app/application/engine/tools.py new file mode 100644 index 0000000..18dafe6 --- /dev/null +++ b/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", +] diff --git a/apps/api/app/application/services/workforce_service.py b/apps/api/app/application/services/workforce_service.py index 613025c..a23e270 100644 --- a/apps/api/app/application/services/workforce_service.py +++ b/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) - unit_price = card.customer_price if card else _UNPRICED - internal_cost = card.internal_cost_estimate if card else 0.0 - await self.ledger.record( - usage=UsageRecord( - organization_id=organization_id, - project_id=project_id, - user_id=user_id, - employee_id=employee_id, - run_id=run_id, - unit=unit, - quantity=quantity, - unit_price=unit_price, - internal_cost_estimate=quantity * internal_cost, - total_price=quantity * unit_price, - metadata_json={"priced": card is not None}, - ) + await record_model_usage( + ledger=self.ledger, + rate_cards=self.rate_cards, + organization_id=organization_id, + unit=unit, + quantity=quantity, + project_id=project_id, + user_id=user_id, + employee_id=employee_id, + run_id=run_id, ) - # 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, diff --git a/apps/api/app/core/config.py b/apps/api/app/core/config.py index 6da12f8..63c0674 100644 --- a/apps/api/app/core/config.py +++ b/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 diff --git a/apps/api/app/domain/workforce/engine.py b/apps/api/app/domain/workforce/engine.py index 0e7f333..a73ab35 100644 --- a/apps/api/app/domain/workforce/engine.py +++ b/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( diff --git a/apps/api/app/infrastructure/repositories/engine.py b/apps/api/app/infrastructure/repositories/engine.py index 3dbc508..1ad7b29 100644 --- a/apps/api/app/infrastructure/repositories/engine.py +++ b/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() diff --git a/apps/api/tests/engine_fakes.py b/apps/api/tests/engine_fakes.py new file mode 100644 index 0000000..18780fa --- /dev/null +++ b/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", +] diff --git a/apps/api/tests/test_engine_prompt.py b/apps/api/tests/test_engine_prompt.py new file mode 100644 index 0000000..924c691 --- /dev/null +++ b/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 diff --git a/apps/api/tests/test_engine_runner.py b/apps/api/tests/test_engine_runner.py new file mode 100644 index 0000000..ca2f295 --- /dev/null +++ b/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\"]]<<>>[[/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\"]]<<>>[[/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 diff --git a/apps/api/tests/test_engine_tools.py b/apps/api/tests/test_engine_tools.py new file mode 100644 index 0000000..d4a09dd --- /dev/null +++ b/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) diff --git a/docs/plan/reports/agent-2-phase-2.md b/docs/plan/reports/agent-2-phase-2.md new file mode 100644 index 0000000..228a2b0 --- /dev/null +++ b/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` + (`<<>>` 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`.