Merge remote-tracking branch 'origin/dev' into dev

apps/api/alembic/versions/a2b3c4d5e6f2_ai_workforce.py

@@ -0,0 +1,186 @@
+"""AI workforce engine: pipeline state, employee profiles, turns, gate results
+
+Revision ID: a2b3c4d5e6f2
+Revises: d5e6f7a8b9c0
+Create Date: 2026-10-14 09:30:00.000000
+
+Owned by the AI-workforce workstream (PLAN-v2 §6).
+
+Deviation from TODO-agent-2, noted so the next reader is not surprised: the
+TODO pins ``down_revision = "a0f0e1d2c3b4"`` (the v2 foundation), which was
+correct while each agent worked on its own branch. On the merged ``dev``
+branch the agents have landed one by one, so this migration chains after the
+most recent head (``d5e6f7a8b9c0``, org setup drafts) to keep a linear
+history. Claude adds the merge revision for the two remaining heads.
+
+No ``tasks`` or ``employees`` column is added: those tables are locked. The
+engine's per-task and per-employee data lives in 1:1 side tables instead.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "a2b3c4d5e6f2"
+down_revision: str | None = "d5e6f7a8b9c0"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "task_pipeline_state",
+        sa.Column(
+            "task_id",
+            sa.String(length=32),
+            sa.ForeignKey("tasks.id", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+        sa.Column(
+            "organization_id",
+            sa.String(length=32),
+            sa.ForeignKey("organizations.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "project_id",
+            sa.String(length=32),
+            sa.ForeignKey("projects.id", ondelete="CASCADE"),
+            nullable=True,
+        ),
+        sa.Column(
+            "pipeline",
+            sa.String(length=32),
+            nullable=False,
+            server_default="plan_code_test",
+        ),
+        sa.Column("kind", sa.String(length=16), nullable=False, server_default="feature"),
+        sa.Column("risk", sa.String(length=8), nullable=False, server_default="medium"),
+        sa.Column(
+            "stage",
+            sa.String(length=32),
+            nullable=False,
+            server_default="pending_manager_review",
+        ),
+        sa.Column("cycle", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("max_cycles", sa.Integer(), nullable=False, server_default="6"),
+        sa.Column("branch", sa.String(length=120), nullable=False, server_default=""),
+        sa.Column("last_diff_hash", sa.String(length=64), nullable=False, server_default=""),
+        sa.Column("same_diff_count", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("retry_count", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("waiver_reason", sa.Text(), nullable=False, server_default=""),
+        sa.Column(
+            "parent_task_id",
+            sa.String(length=32),
+            sa.ForeignKey("tasks.id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+        sa.Column("depends_on", sa.JSON(), nullable=False, server_default="[]"),
+        sa.Column("version", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+    )
+    op.create_index(
+        "ix_task_pipeline_state_organization_id",
+        "task_pipeline_state",
+        ["organization_id"],
+    )
+    op.create_index(
+        "ix_task_pipeline_state_stage", "task_pipeline_state", ["stage"]
+    )
+
+    op.create_table(
+        "employee_profiles",
+        sa.Column(
+            "employee_id",
+            sa.String(length=32),
+            sa.ForeignKey("employees.id", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+        sa.Column("ai_model_id", sa.String(length=32), nullable=True),
+        sa.Column(
+            "effort", sa.String(length=16), nullable=False, server_default="medium"
+        ),
+        sa.Column("roles", sa.String(length=255), nullable=False, server_default=""),
+        sa.Column("hired_by", sa.String(length=8), nullable=False, server_default="user"),
+        sa.Column(
+            "busy_task_id",
+            sa.String(length=32),
+            sa.ForeignKey("tasks.id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+    )
+    op.create_index(
+        "ix_employee_profile_busy", "employee_profiles", ["busy_task_id"]
+    )
+
+    op.create_table(
+        "engine_turns",
+        sa.Column("id", sa.String(length=32), primary_key=True),
+        sa.Column(
+            "task_id",
+            sa.String(length=32),
+            sa.ForeignKey("tasks.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "run_id",
+            sa.String(length=32),
+            sa.ForeignKey("runs.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "employee_id",
+            sa.String(length=32),
+            sa.ForeignKey("employees.id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+        sa.Column("role", sa.String(length=24), nullable=False, server_default="developer"),
+        sa.Column("mode", sa.String(length=16), nullable=False, server_default="work"),
+        sa.Column("stage", sa.String(length=32), nullable=False, server_default=""),
+        sa.Column("prompt_chars", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("reply_excerpt", sa.Text(), nullable=False, server_default=""),
+        sa.Column("tags", sa.JSON(), nullable=False, server_default="{}"),
+        sa.Column("tool_calls", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("ok", sa.Boolean(), nullable=False, server_default=sa.true()),
+        sa.Column("error_kind", sa.String(length=16), nullable=False, server_default=""),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+    )
+    op.create_index("ix_engine_turn_task_stage", "engine_turns", ["task_id", "stage"])
+    op.create_index("ix_engine_turn_run", "engine_turns", ["run_id"])
+
+    op.create_table(
+        "gate_results",
+        sa.Column("id", sa.String(length=32), primary_key=True),
+        sa.Column(
+            "task_id",
+            sa.String(length=32),
+            sa.ForeignKey("tasks.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("gate", sa.String(length=16), nullable=False, server_default="tests"),
+        sa.Column("ok", sa.Boolean(), nullable=False, server_default=sa.true()),
+        sa.Column("detail", sa.Text(), nullable=False, server_default=""),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+    )
+    op.create_index("ix_gate_result_task_gate", "gate_results", ["task_id", "gate"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_gate_result_task_gate", table_name="gate_results")
+    op.drop_table("gate_results")
+    op.drop_index("ix_engine_turn_run", table_name="engine_turns")
+    op.drop_index("ix_engine_turn_task_stage", table_name="engine_turns")
+    op.drop_table("engine_turns")
+    op.drop_index("ix_employee_profile_busy", table_name="employee_profiles")
+    op.drop_table("employee_profiles")
+    op.drop_index("ix_task_pipeline_state_stage", table_name="task_pipeline_state")
+    op.drop_index(
+        "ix_task_pipeline_state_organization_id", table_name="task_pipeline_state"
+    )
+    op.drop_table("task_pipeline_state")

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

@@ -0,0 +1 @@
+"""The AI-workforce engine: the tag protocol and the worker-facing helpers."""

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

@@ -0,0 +1,576 @@
+"""The tag protocol: ``[[TAG attr="v" ...]]body[[/TAG]]`` and ``[[TAG attr="v"/]``.
+
+One parser, one matrix. The rules that make an agent protocol safe all live
+here so no call site can re-decide them:
+
+1. **An allow-list per role and mode.** The manager triages, gives the final
+   verdict, handles errors and chats, and each mode may emit a different set
+   of tags; a planner plans, a developer implements. A tag outside the matrix
+   is ignored and reported back to the model as an error line — the model
+   learns the rule, and nothing dangerous executes by accident.
+2. **The toolkit is the only way to touch files.** ``READ_FILE``,
+   ``WRITE_FILE``, ``EDIT_FILE``, ``LIST_DIR``, ``SEARCH`` and ``RUN`` run
+   inside the sandbox through ``SandboxProviderPort``; the platform still
+   does git (``RepoWorkspacePort``), and ``RUN`` refuses git push/remote/config
+   outright (see ``engine/tools.py``).
+3. **Bounded turns.** A reply may create at most three tasks, change the team
+   at most three times and record at most three knowledge items. A model that
+   ignores the limit gets an error line and the extra tags are dropped, so the
+   effect of one runaway reply stays finite.
+
+Bodies may contain brackets (a source file does), so the parser never
+re-scans a body it already consumed: an outer tag wins and its content is
+literal.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from enum import StrEnum
+
+from app.domain.workforce.models import RoleKey
+from app.domain.workforce.pipeline import Mode, PipelineKey
+
+
+class TagName(StrEnum):
+    """Every tag the protocol knows, whether or not a role may use it."""
+
+    # manager / triage
+    ASSIGN = "ASSIGN"
+    REJECT = "REJECT"
+    ASK_USER = "ASK_USER"
+    KNOWLEDGE = "KNOWLEDGE"
+    # manager / final
+    COMPLETE = "COMPLETE"
+    REVISE = "REVISE"
+    # manager / error
+    RESOLVE_ERROR = "RESOLVE_ERROR"
+    CANCEL = "CANCEL"
+    # manager / chat
+    CREATE_TASK = "CREATE_TASK"
+    PAUSE_TASK = "PAUSE_TASK"
+    RESUME_TASK = "RESUME_TASK"
+    CANCEL_TASK = "CANCEL_TASK"
+    HIRE = "HIRE"
+    UPDATE_EMPLOYEE = "UPDATE_EMPLOYEE"
+    FIRE = "FIRE"
+    # planner
+    PLAN_SPEC = "PLAN_SPEC"
+    SPLIT = "SPLIT"
+    # developer
+    WORK_REPORT = "WORK_REPORT"
+    # reviewer
+    REVIEW_PASS = "REVIEW_PASS"
+    REVIEW_CHANGES = "REVIEW_CHANGES"
+    # tester
+    TEST_PASS = "TEST_PASS"
+    TEST_FAIL = "TEST_FAIL"
+    # toolkit (every working role)
+    READ_FILE = "READ_FILE"
+    WRITE_FILE = "WRITE_FILE"
+    EDIT_FILE = "EDIT_FILE"
+    LIST_DIR = "LIST_DIR"
+    SEARCH = "SEARCH"
+    RUN = "RUN"
+
+
+#: The toolkit. Available to every working role, never to the manager: the
+#: manager decides, it does not edit the repository.
+TOOL_TAGS: frozenset[TagName] = frozenset(
+    {
+        TagName.READ_FILE,
+        TagName.WRITE_FILE,
+        TagName.EDIT_FILE,
+        TagName.LIST_DIR,
+        TagName.SEARCH,
+        TagName.RUN,
+    }
+)
+
+#: Team-changing tags (all manager/chat). Counted together, one limit per turn.
+TEAM_CHANGE_TAGS: frozenset[TagName] = frozenset(
+    {TagName.HIRE, TagName.UPDATE_EMPLOYEE, TagName.FIRE}
+)
+
+#: Tags that carry a body and therefore *must* be closed with ``[[/TAG]]``.
+#: Every other tag is a single token (an order, a tool read, a decision), so
+#: a forgotten trailing ``/`` is tolerated as self-closing instead of failing
+#: the whole reply: ``[[RUN cmd="pytest"]]`` means what ``[[RUN cmd="pytest"/]]``
+#: means.
+BODY_TAGS: frozenset[TagName] = frozenset(
+    {
+        TagName.WRITE_FILE,
+        TagName.EDIT_FILE,
+        TagName.KNOWLEDGE,
+        TagName.ASK_USER,
+        TagName.CREATE_TASK,
+        TagName.SPLIT,
+        TagName.PLAN_SPEC,
+        TagName.WORK_REPORT,
+        TagName.REVIEW_PASS,
+        TagName.REVIEW_CHANGES,
+        TagName.TEST_PASS,
+        TagName.TEST_FAIL,
+    }
+)
+
+#: The per-role/mode matrix. Anything not listed is refused with an error line.
+ALLOWED_TAGS: dict[tuple[RoleKey, Mode], frozenset[TagName]] = {
+    (RoleKey.MANAGER, Mode.TRIAGE): frozenset(
+        {TagName.ASSIGN, TagName.REJECT, TagName.ASK_USER, TagName.KNOWLEDGE}
+    ),
+    (RoleKey.MANAGER, Mode.FINAL): frozenset(
+        {TagName.COMPLETE, TagName.REVISE, TagName.KNOWLEDGE}
+    ),
+    (RoleKey.MANAGER, Mode.ERROR): frozenset(
+        {TagName.RESOLVE_ERROR, TagName.CANCEL, TagName.KNOWLEDGE}
+    ),
+    (RoleKey.MANAGER, Mode.CHAT): frozenset(
+        {
+            TagName.CREATE_TASK,
+            TagName.PAUSE_TASK,
+            TagName.RESUME_TASK,
+            TagName.CANCEL_TASK,
+            TagName.HIRE,
+            TagName.UPDATE_EMPLOYEE,
+            TagName.FIRE,
+            TagName.KNOWLEDGE,
+        }
+    ),
+    (RoleKey.PLANNER, Mode.WORK): frozenset({TagName.PLAN_SPEC, TagName.SPLIT, TagName.ASK_USER})
+    | TOOL_TAGS,
+    (RoleKey.DEVELOPER, Mode.WORK): frozenset({TagName.WORK_REPORT, TagName.ASK_USER}) | TOOL_TAGS,
+    (RoleKey.REVIEWER, Mode.WORK): frozenset(
+        {TagName.REVIEW_PASS, TagName.REVIEW_CHANGES}
+    )
+    | TOOL_TAGS,
+    (RoleKey.TESTER, Mode.WORK): frozenset({TagName.TEST_PASS, TagName.TEST_FAIL}) | TOOL_TAGS,
+}
+
+#: Per-reply limits. Extras are dropped and reported, never executed.
+MAX_CREATE_TASKS_PER_REPLY = 3
+MAX_TEAM_CHANGES_PER_TURN = 3
+MAX_KNOWLEDGE_PER_TURN = 3
+
+#: Which pipeline names ``ASSIGN pipeline=`` may pick. Kept here so the
+#: manager cannot invent a pipeline that does not exist.
+ASSIGNABLE_PIPELINES: frozenset[str] = frozenset(p.value for p in PipelineKey)
+
+
+class ProtocolError(ValueError):
+    """A tag was parsed but its attributes are not usable."""
+
+
+# ---------------------------------------------------------------------------
+# Tag and reply
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True, slots=True)
+class Tag:
+    """One parsed tag: its name, attributes and body (``""`` when self-closing)."""
+
+    name: TagName
+    attrs: dict[str, str] = field(default_factory=dict)
+    body: str = ""
+    #: The raw slice, so an error line can quote exactly what the model wrote.
+    raw: str = ""
+
+    def attr(self, key: str, default: str = "") -> str:
+        return self.attrs.get(key, default)
+
+    def require(self, key: str) -> str:
+        value = self.attrs.get(key, "").strip()
+        if not value:
+            raise ProtocolError(f"[[{self.name}]] requires {key}=...")
+        return value
+
+
+@dataclass(frozen=True, slots=True)
+class ParsedReply:
+    """Everything one model reply contained, in order.
+
+    ``errors`` are lines fed back to the model on its next turn: the protocol
+    is learned from feedback, not from a crash. ``prose`` is the text outside
+    any tag — for a manager chat reply that is the message the user reads.
+    """
+
+    tags: tuple[Tag, ...] = ()
+    errors: tuple[str, ...] = ()
+    prose: str = ""
+
+    def names(self) -> tuple[TagName, ...]:
+        return tuple(tag.name for tag in self.tags)
+
+    def of(self, name: TagName | str) -> tuple[Tag, ...]:
+        wanted = TagName(name)
+        return tuple(tag for tag in self.tags if tag.name is wanted)
+
+    def first(self, name: TagName | str) -> Tag | None:
+        found = self.of(name)
+        return found[0] if found else None
+
+    def has(self, name: TagName | str) -> bool:
+        return bool(self.of(name))
+
+    def error_text(self) -> str:
+        return "\n".join(self.errors)
+
+
+# ---------------------------------------------------------------------------
+# Typed views over tags
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True, slots=True)
+class AssignTag:
+    """``ASSIGN employee="name" pipeline="plan_code_test"``."""
+
+    employee: str
+    pipeline: str = ""
+    task_id: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class TaskDraft:
+    """``CREATE_TASK``: the manager asks for a new task in chat."""
+
+    title: str
+    kind: str = "feature"
+    priority: str = "normal"
+    depends_on: tuple[str, ...] = ()
+    pipeline: str = ""
+    description: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class TeamChange:
+    """``HIRE`` / ``UPDATE_EMPLOYEE`` / ``FIRE``."""
+
+    action: TagName
+    name: str = ""
+    role: str = ""
+    tier: str = ""
+    task_id: str = ""
+    employee_id: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class KnowledgeItem:
+    """``KNOWLEDGE kind="rule|fact" key="..."`` with the fact as the body."""
+
+    kind: str = "rule"
+    key: str = ""
+    content: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class EditBlock:
+    """One ``<<<OLD … === … NEW>>>`` search/replace block inside an EDIT_FILE."""
+
+    old: str
+    new: str
+
+
+_EDIT_BLOCK_RE = re.compile(
+    r"<<<OLD\s*\n(?P<old>.*?)\n===\s*\n(?P<new>.*?)\n\s*NEW>>>",
+    re.DOTALL,
+)
+
+
+def assign_of(tag: Tag) -> AssignTag:
+    """Read an ``ASSIGN`` tag. ``employee=""`` means "pick for me"."""
+    employee = tag.attr("employee", "").strip()
+    if not employee and not tag.attr("task_id", "").strip():
+        raise ProtocolError("[[ASSIGN]] needs employee=... or task_id=...")
+    pipeline = tag.attr("pipeline", "").strip()
+    if pipeline and pipeline not in ASSIGNABLE_PIPELINES:
+        raise ProtocolError(f"unknown pipeline: {pipeline}")
+    return AssignTag(
+        employee=employee, pipeline=pipeline, task_id=tag.attr("task_id", "").strip()
+    )
+
+
+def task_draft_of(tag: Tag) -> TaskDraft:
+    """Read a ``CREATE_TASK`` tag."""
+    title = tag.attr("title", "").strip()
+    if not title:
+        raise ProtocolError("[[CREATE_TASK]] needs title=...")
+    raw_deps = tag.attr("depends_on", "")
+    depends = tuple(d.strip() for d in raw_deps.split(",") if d.strip())
+    return TaskDraft(
+        title=title,
+        kind=tag.attr("kind", "feature").strip() or "feature",
+        priority=tag.attr("priority", "normal").strip() or "normal",
+        depends_on=depends,
+        pipeline=tag.attr("pipeline", "").strip(),
+        description=tag.body.strip()[:4000],
+    )
+
+
+def team_change_of(tag: Tag) -> TeamChange:
+    """Read one of the team tags. Users choose role and tier, never a model."""
+    return TeamChange(
+        action=tag.name,
+        name=tag.attr("name", "").strip(),
+        role=tag.attr("role", "").strip(),
+        tier=tag.attr("tier", "").strip(),
+        task_id=tag.attr("task_id", "").strip(),
+        employee_id=tag.attr("employee_id", "").strip(),
+    )
+
+
+def knowledge_of(tag: Tag) -> KnowledgeItem:
+    """Read a ``KNOWLEDGE`` tag; the body is the content."""
+    return KnowledgeItem(
+        kind=tag.attr("kind", "rule").strip() or "rule",
+        key=tag.attr("key", "").strip(),
+        content=tag.body.strip(),
+    )
+
+
+def edit_blocks_of(tag: Tag) -> list[EditBlock]:
+    """Read the search/replace blocks of an ``EDIT_FILE`` body.
+
+    Blocks are ``<<<OLD`` … ``===`` … ``NEW>>>``. A malformed body raises
+    rather than guessing: a half-applied edit is worse than a failed one.
+    """
+    blocks = [
+        EditBlock(old=m.group("old"), new=m.group("new"))
+        for m in _EDIT_BLOCK_RE.finditer(tag.body)
+    ]
+    if not blocks:
+        raise ProtocolError("[[EDIT_FILE]] needs one or more <<<OLD ... === ... NEW>>> blocks")
+    for block in blocks:
+        if not block.old.strip():
+            raise ProtocolError("[[EDIT_FILE]] OLD block is empty")
+    return blocks
+
+
+def task_drafts(reply: ParsedReply) -> list[TaskDraft]:
+    return [task_draft_of(tag) for tag in reply.of(TagName.CREATE_TASK)]
+
+
+def team_changes(reply: ParsedReply) -> list[TeamChange]:
+    """Every team tag, in the order the model wrote them."""
+    return [team_change_of(tag) for tag in reply.tags if tag.name in TEAM_CHANGE_TAGS]
+
+
+def knowledge_items(reply: ParsedReply) -> list[KnowledgeItem]:
+    return [knowledge_of(tag) for tag in reply.of(TagName.KNOWLEDGE)]
+
+
+# ---------------------------------------------------------------------------
+# Parser
+# ---------------------------------------------------------------------------
+
+#: ``[[ ... ]]`` whose content has no brackets inside: an opening token, a
+#: closing token, or a self-closing one. Attribute values with a ``]`` are
+#: refused by the attribute parser below instead of silently truncating.
+_TOKEN_RE = re.compile(r"\[\[([^\[\]]*)\]\]")
+_ATTR_RE = re.compile(r"([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(?:\"([^\"]*)\"|'([^']*)')")
+
+
+def _closing_re(name: str) -> re.Pattern[str]:
+    return re.compile(r"\[\[/\s*" + re.escape(name) + r"\s*\]\]", re.IGNORECASE)
+
+
+def _split_token(inner: str) -> tuple[str, str, bool]:
+    """``'TAG a="1"'`` -> ``("TAG", ' a="1"', False)``; trailing ``/`` self-closes."""
+    text = inner.strip()
+    self_closing = text.endswith("/")
+    if self_closing:
+        text = text[:-1].strip()
+    match = re.match(r"([A-Za-z_][A-Za-z0-9_]*)", text)
+    if match is None:  # pragma: no cover - the token regex requires a name
+        return "", "", self_closing
+    name = match.group(1)
+    return name.upper(), text[match.end() :], self_closing
+
+
+def _parse_attrs(rest: str) -> tuple[dict[str, str], list[str]]:
+    attrs: dict[str, str] = {}
+    errors: list[str] = []
+    consumed_end = 0
+    for match in _ATTR_RE.finditer(rest):
+        attrs[match.group(1)] = match.group(2) if match.group(2) is not None else match.group(3)
+        consumed_end = match.end()
+    leftover = rest[consumed_end:].strip()
+    if leftover:
+        # One line for the whole fragment: "employee سارا" reads better than two.
+        errors.append(
+            f"ERROR: malformed attribute fragment '{leftover}' (expected key=\"value\")"
+        )
+    return attrs, errors
+
+
+def _consume(name: str, text: str, start: int, self_closing: bool) -> int:
+    """Where the tag ends: its ``[[/NAME]]`` close, or just the token.
+
+    Consuming the whole pair is what keeps a body literal: the scanner moves
+    past the close, so brackets inside the body are never parsed as tags.
+    """
+    if self_closing:
+        return start
+    close = _closing_re(name).search(text, start)
+    return close.end() if close else start
+
+
+def parse_reply(text: str, *, role: RoleKey | str, mode: Mode | str) -> ParsedReply:
+    """Parse one model reply under the role/mode matrix.
+
+    Unknown tags, disallowed tags and unclosed tags are dropped (their body is
+    consumed so it cannot re-appear as stray tags) and reported in
+    ``ParsedReply.errors`` — one line per problem, in order, for the model to
+    read on its next turn.
+    """
+    role_key = RoleKey(role)
+    role_mode = Mode(mode)
+    allowed = ALLOWED_TAGS.get((role_key, role_mode), frozenset())
+
+    tags: list[Tag] = []
+    errors: list[str] = []
+    prose_parts: list[str] = []
+    pos = 0
+
+    while pos < len(text):
+        match = _TOKEN_RE.search(text, pos)
+        if match is None:
+            prose_parts.append(text[pos:])
+            break
+        prose_parts.append(text[pos : match.start()])
+        inner = match.group(1)
+        name, rest, self_closing = _split_token(inner)
+
+        if inner.lstrip().startswith("/"):
+            # A close with no open: the model echoed a tag it never opened.
+            errors.append(f"ERROR: stray closing tag [[{inner.strip()}]] ignored")
+            pos = match.end()
+            continue
+
+        token_end = match.end()
+        try:
+            tag_name = TagName(name)
+        except ValueError:
+            # Unknown tag: consume the pair so its body stays literal text.
+            end = _consume(name, text, token_end, self_closing)
+            errors.append(f"ERROR: unknown tag [[{name}]] ignored")
+            pos = end
+            continue
+
+        if tag_name not in allowed:
+            end = _consume(name, text, token_end, self_closing)
+            errors.append(
+                f"ERROR: tag [[{tag_name}]] is not allowed for "
+                f"role={role_key.value} mode={role_mode.value}"
+            )
+            pos = end
+            continue
+
+        attrs, attr_errors = _parse_attrs(rest)
+        errors.extend(attr_errors)
+
+        body = ""
+        if not self_closing:
+            if tag_name not in BODY_TAGS:
+                # A bodyless tag written without its "/": read it as-is.
+                tags.append(
+                    Tag(
+                        name=tag_name,
+                        attrs=attrs,
+                        body="",
+                        raw=text[match.start() : token_end],
+                    )
+                )
+                pos = token_end
+                continue
+            close = _closing_re(name).search(text, token_end)
+            if close is None:
+                # Unclosed: report it and ignore the tag, body included.
+                errors.append(f"ERROR: unclosed tag [[{name}]] ignored")
+                pos = token_end
+                continue
+            body = text[token_end : close.start()]
+            token_end = close.end()
+
+        tags.append(Tag(name=tag_name, attrs=attrs, body=body, raw=text[match.start() : token_end]))
+        pos = token_end
+
+    limited, limit_errors = _apply_limits(tags)
+    errors.extend(limit_errors)
+    return ParsedReply(
+        tags=tuple(limited), errors=tuple(errors), prose="".join(prose_parts).strip()
+    )
+
+
+def _apply_limits(tags: list[Tag]) -> tuple[list[Tag], list[str]]:
+    """Drop tags past their per-reply limit, keeping the first ones written.
+
+    Team changes share one budget across all three tags: "max 3 changes per
+    turn" means three hires, or two hires and a firing — not three of each.
+    """
+    errors: list[str] = []
+    kept: list[Tag] = []
+    created = 0
+    team_used = 0
+    knowledge_used = 0
+    for tag in tags:
+        if tag.name is TagName.CREATE_TASK:
+            created += 1
+            if created > MAX_CREATE_TASKS_PER_REPLY:
+                errors.append(_limit_error(tag, MAX_CREATE_TASKS_PER_REPLY))
+                continue
+        elif tag.name in TEAM_CHANGE_TAGS:
+            team_used += 1
+            if team_used > MAX_TEAM_CHANGES_PER_TURN:
+                errors.append(_limit_error(tag, MAX_TEAM_CHANGES_PER_TURN))
+                continue
+        elif tag.name is TagName.KNOWLEDGE:
+            knowledge_used += 1
+            if knowledge_used > MAX_KNOWLEDGE_PER_TURN:
+                errors.append(_limit_error(tag, MAX_KNOWLEDGE_PER_TURN))
+                continue
+        kept.append(tag)
+    return kept, errors
+
+
+def _limit_error(tag: Tag, limit: int) -> str:
+    return (
+        f"ERROR: too many [[{tag.name}]] in one reply (max {limit}); the extra was ignored"
+    )
+
+
+def allowed_tags(*, role: RoleKey | str, mode: Mode | str) -> frozenset[TagName]:
+    return ALLOWED_TAGS.get((RoleKey(role), Mode(mode)), frozenset())
+
+
+__all__ = [
+    "ALLOWED_TAGS",
+    "ASSIGNABLE_PIPELINES",
+    "BODY_TAGS",
+    "MAX_CREATE_TASKS_PER_REPLY",
+    "MAX_KNOWLEDGE_PER_TURN",
+    "MAX_TEAM_CHANGES_PER_TURN",
+    "TEAM_CHANGE_TAGS",
+    "TOOL_TAGS",
+    "AssignTag",
+    "EditBlock",
+    "KnowledgeItem",
+    "ParsedReply",
+    "ProtocolError",
+    "Tag",
+    "TagName",
+    "TaskDraft",
+    "TeamChange",
+    "allowed_tags",
+    "assign_of",
+    "edit_blocks_of",
+    "knowledge_items",
+    "knowledge_of",
+    "parse_reply",
+    "task_drafts",
+    "task_draft_of",
+    "team_change_of",
+    "team_changes",
+]

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

@@ -0,0 +1,114 @@
+"""Moving a task between stages — the only place the stage column changes.
+
+Two rules live here rather than in the scheduler:
+
+1. **Optimistic moves.** ``advance`` refuses to guess: it takes the stage the
+   caller believes the task is at, and the repository's compare-and-set either
+   applies the move or raises :class:`~app.core.errors.ConflictError` because
+   another worker got there first. A lost race is a normal outcome, not a
+   crash — the caller re-reads and continues.
+2. **Dependents never get stuck.** When a task fails or is cancelled, every
+   task that declared it in ``depends_on`` is parked in
+   ``pending_user_decision`` instead of waiting for a dependency that will
+   never finish (agybot bug M1: a dead task deadlocked its children).
+
+The coarse ``tasks.status`` is written by the caller in the same transaction
+(see ``engine/scheduler.py`` in a later phase): the stage is the engine's
+truth, the status column is the summary existing APIs keep reading.
+"""
+
+from __future__ import annotations
+
+from app.core.errors import ConflictError, NotFoundError
+from app.domain.workforce.pipeline import (
+    SIDE_STAGES,
+    TERMINAL_STAGES,
+    Stage,
+    TaskPipelineState,
+    can_transition_stage,
+)
+from app.infrastructure.repositories.engine import (
+    InvalidTransitionError,
+    PipelineStateRepository,
+    StaleStageError,
+    UnknownTaskError,
+)
+
+
+async def advance(
+    states: PipelineStateRepository,
+    *,
+    task_id: str,
+    expected_from: Stage,
+    to: Stage,
+) -> TaskPipelineState:
+    """Move one task from ``expected_from`` to ``to``, or fail cleanly.
+
+    Raises ``ConflictError`` when the transition is illegal or when another
+    worker moved the task first, and ``NotFoundError`` when the task has no
+    engine state row at all.
+    """
+    if not can_transition_stage(expected_from, to):
+        raise ConflictError(
+            f"گذار نابجا از {expected_from.value} به {to.value}."
+        )
+    try:
+        return await states.advance(task_id=task_id, expected_from=expected_from, to=to)
+    except UnknownTaskError as exc:
+        raise NotFoundError("این کار وضعیت موتور ندارد.") from exc
+    except InvalidTransitionError as exc:
+        raise ConflictError(str(exc)) from exc
+    except StaleStageError as exc:
+        raise ConflictError(
+            "این کار هم‌اکنون توسط موتور دیگری در حرکت است؛ دوباره تلاش کنید."
+        ) from exc
+
+
+async def park_dependents(
+    states: PipelineStateRepository,
+    *,
+    task_id: str,
+) -> list[TaskPipelineState]:
+    """Park every dependent of a failed or cancelled task with a human.
+
+    A dependency that failed leaves its dependents unable to ever finish, so
+    they move to ``pending_user_decision`` instead of waiting for work that
+    will never happen (agybot bug M1: a dead task deadlocked its children).
+    Dependents that already sit in another side state are left alone — a task
+    a human already paused must not be re-parked.
+
+    The guard is deliberate: a task that merely *exists* is not a reason to
+    park anything. Only a dependency that is actually ``failed`` or
+    ``cancelled`` strands its children, so the function refuses to act on a
+    healthy task however often a tick calls it.
+    """
+    state = await states.get(task_id)
+    if state is None or state.stage not in (Stage.FAILED, Stage.CANCELLED):
+        return []
+    parked: list[TaskPipelineState] = []
+    for dependent in await states.list_dependents(task_id):
+        # A dependent that already waits on a human (paused, another decision)
+        # is left alone: it is already parked, just elsewhere.
+        if dependent.stage in SIDE_STAGES:
+            continue
+        if not can_transition_stage(dependent.stage, Stage.PENDING_USER_DECISION):
+            continue
+        try:
+            parked.append(
+                await advance(
+                    states,
+                    task_id=dependent.task_id,
+                    expected_from=dependent.stage,
+                    to=Stage.PENDING_USER_DECISION,
+                )
+            )
+        except ConflictError:
+            # A worker is already moving it; the next tick will see the truth.
+            continue
+    return parked
+
+
+async def is_terminal_state(states: PipelineStateRepository, task_id: str) -> bool:
+    """True when the task's stage can no longer move (done/cancelled/failed)."""
+    state = await states.get(task_id)
+    return state is None or state.stage in TERMINAL_STAGES

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

@@ -0,0 +1,123 @@
+"""Domain records for the AI-workforce engine.
+
+Pure data: the rows the engine writes about itself (one turn, one gate, one
+employee's model binding) plus the error vocabulary the runner classifies
+with. Kept separate from ``pipeline.py`` because that module is the state
+machine and this one is the ledger the state machine leaves behind.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import StrEnum
+
+from app.domain.base import new_id, utcnow
+from app.domain.workforce.models import RoleKey
+from app.domain.workforce.pipeline import Mode, Stage
+
+
+class ErrorKind(StrEnum):
+    """How a failed model call should be treated.
+
+    ``rate_limit``, ``unavailable``, ``network`` and ``timeout`` are
+    transient: the platform retries later instead of burning a cycle. The
+    others are the caller's problem — a bad key never fixes itself.
+    """
+
+    RATE_LIMIT = "rate_limit"
+    UNAVAILABLE = "unavailable"
+    NETWORK = "network"
+    TIMEOUT = "timeout"
+    AUTH = "auth"
+    MODEL = "model"
+    PROTOCOL = "protocol"
+    BUDGET = "budget"
+
+
+TRANSIENT_ERROR_KINDS: frozenset[ErrorKind] = frozenset(
+    {
+        ErrorKind.RATE_LIMIT,
+        ErrorKind.UNAVAILABLE,
+        ErrorKind.NETWORK,
+        ErrorKind.TIMEOUT,
+    }
+)
+
+#: Retries a transient failure gets before the task goes to error review.
+MAX_TRANSIENT_RETRIES = 2
+
+
+def is_transient(kind: ErrorKind | str) -> bool:
+    return ErrorKind(kind) in TRANSIENT_ERROR_KINDS
+
+
+class GateName(StrEnum):
+    """The gates that decide whether work may move forward."""
+
+    TESTS = "tests"
+    DIFF_GUARD = "diff_guard"
+    NO_PROGRESS = "no_progress"
+    GREEN = "green"
+    LINT = "lint"
+
+
+@dataclass(slots=True, kw_only=True)
+class EmployeeProfile:
+    """The engine's per-employee record: which model it runs on, and its roles.
+
+    A 1:1 side table for the same reason as ``TaskPipelineState`` — the
+    ``employees`` table is locked. Model identity lives here and never
+    crosses an API boundary: users pick a role and a neutral tier.
+    """
+
+    employee_id: str
+    ai_model_id: str | None = None
+    effort: str = "medium"
+    #: CSV of role keys the employee may be hired for (its own role plus extras).
+    roles: list[str] = field(default_factory=list)
+    hired_by: str = "user"  # user | manager
+    busy_task_id: str | None = None
+    updated_at: datetime = field(default_factory=utcnow)
+
+    def role_list(self) -> list[str]:
+        return [r.strip() for r in self.roles if r.strip()]
+
+
+@dataclass(slots=True, kw_only=True)
+class EngineTurn:
+    """One employee turn: who ran, what came back, what it cost.
+
+    ``reply_excerpt`` is bounded (4,000 chars) on purpose — the full reply
+    can be enormous, and this table exists to answer "what happened", not to
+    archive the conversation.
+    """
+
+    task_id: str
+    run_id: str
+    employee_id: str | None = None
+    role: RoleKey = RoleKey.DEVELOPER
+    mode: Mode = Mode.WORK
+    stage: Stage = Stage.PENDING_IMPLEMENTATION
+    prompt_chars: int = 0
+    reply_excerpt: str = ""
+    #: tag name -> count, so the UI can show "3 files edited" without a join.
+    tags: dict = field(default_factory=dict)
+    tool_calls: int = 0
+    ok: bool = True
+    error_kind: str = ""
+    id: str = field(default_factory=new_id)
+    started_at: datetime = field(default_factory=utcnow)
+    finished_at: datetime | None = None
+
+
+@dataclass(slots=True, kw_only=True)
+class GateResult:
+    """One gate verdict, with the detail a human needs to trust it."""
+
+    task_id: str
+    gate: GateName
+    ok: bool
+    detail: str = ""
+    id: str = field(default_factory=new_id)
+    created_at: datetime = field(default_factory=utcnow)

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

@@ -0,0 +1,494 @@
+"""Dynamic task pipelines: stages, transitions and the manager's choices.
+
+Pure logic: no I/O, no framework, no database. Everything here is a decision
+the engine makes *before* it touches a row, which is what makes the state
+machine testable without a session.
+
+The vocabulary comes from agybot (concepts only, never its code): one task
+walks an ordered list of stages, every stage needs exactly one role in one
+mode, and the manager decides which pipeline a task gets. The differences
+that matter here are that every move is optimistic (``version`` in
+``TaskPipelineState``) and that a failed or cancelled dependency parks its
+dependents instead of leaving them stuck forever (agybot bug M1).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import StrEnum
+from typing import NamedTuple
+
+from app.domain.base import utcnow
+from app.domain.workforce.models import RoleKey, TaskStatus
+
+
+class Stage(StrEnum):
+    """Where a task is in its pipeline.
+
+    The first eight values are pipeline stages (they appear in
+    :data:`PIPELINES`); the rest are side states a task can be pushed into
+    from anywhere, which is why they are not part of any pipeline list.
+    """
+
+    PENDING_MANAGER_REVIEW = "pending_manager_review"
+    PENDING_ANALYSIS = "pending_analysis"
+    PENDING_IMPLEMENTATION = "pending_implementation"
+    PENDING_REVIEW = "pending_review"
+    PENDING_TEST = "pending_test"
+    PENDING_FINAL_REVIEW = "pending_final_review"
+    MERGE_REQUEST = "merge_request"
+    COMPLETED = "completed"
+    PENDING_USER_DECISION = "pending_user_decision"
+    ERROR_REVIEW = "error_review"
+    PAUSED = "paused"
+    CANCELLED = "cancelled"
+    FAILED = "failed"
+
+
+#: Side states and terminals: a task in one of these is not walking a pipeline.
+SIDE_STAGES: frozenset[Stage] = frozenset(
+    {
+        Stage.PENDING_USER_DECISION,
+        Stage.ERROR_REVIEW,
+        Stage.PAUSED,
+        Stage.CANCELLED,
+        Stage.FAILED,
+    }
+)
+TERMINAL_STAGES: frozenset[Stage] = frozenset(
+    {Stage.COMPLETED, Stage.CANCELLED, Stage.FAILED}
+)
+
+
+class PipelineKey(StrEnum):
+    """The four pipelines. Names match the plan (agybot's, minus its extras)."""
+
+    PLAN_CODE_TEST = "plan_code_test"
+    PLAN_CODE_REVIEW_TEST = "plan_code_review_test"
+    CODE_TEST = "code_test"
+    ANALYSIS_ONLY = "analysis_only"
+
+
+class TaskKind(StrEnum):
+    FEATURE = "feature"
+    BUGFIX = "bugfix"
+    REFACTOR = "refactor"
+    DOCS = "docs"
+    ANALYSIS = "analysis"
+
+
+class Risk(StrEnum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+
+
+class Mode(StrEnum):
+    """The hat a role wears in one turn.
+
+    Only the manager has more than one: the same employee triages, gives the
+    final verdict and handles errors, and each mode allows a different tag set
+    (see ``engine/protocol.py``). ``work`` is what every other role uses.
+    """
+
+    TRIAGE = "triage"
+    FINAL = "final"
+    ERROR = "error"
+    CHAT = "chat"
+    WORK = "work"
+
+
+class Outcome(StrEnum):
+    """The verdict of a finished turn, as the pipeline understands it."""
+
+    #: The turn produced its tag: move to the next pipeline stage.
+    OK = "ok"
+    #: Reviewer/tester/manager sent the work back: cycle again.
+    REVISE = "revise"
+    #: The manager rejected the task outright.
+    REJECT = "reject"
+    #: The turn needs a human answer before it can continue.
+    ASK_USER = "ask_user"
+    #: Something failed hard enough for the manager's error turn.
+    ERROR = "error"
+    #: The error turn decided to retry with a hint.
+    RESOLVE = "resolve"
+
+
+#: Ordered stages per pipeline. ``completed`` is part of every list so that
+#: "advance past the last stage" is just an index lookup, not a special case.
+PIPELINES: dict[PipelineKey, tuple[Stage, ...]] = {
+    PipelineKey.PLAN_CODE_TEST: (
+        Stage.PENDING_MANAGER_REVIEW,
+        Stage.PENDING_ANALYSIS,
+        Stage.PENDING_IMPLEMENTATION,
+        Stage.PENDING_TEST,
+        Stage.PENDING_FINAL_REVIEW,
+        Stage.MERGE_REQUEST,
+        Stage.COMPLETED,
+    ),
+    PipelineKey.PLAN_CODE_REVIEW_TEST: (
+        Stage.PENDING_MANAGER_REVIEW,
+        Stage.PENDING_ANALYSIS,
+        Stage.PENDING_IMPLEMENTATION,
+        Stage.PENDING_REVIEW,
+        Stage.PENDING_TEST,
+        Stage.PENDING_FINAL_REVIEW,
+        Stage.MERGE_REQUEST,
+        Stage.COMPLETED,
+    ),
+    PipelineKey.CODE_TEST: (
+        Stage.PENDING_MANAGER_REVIEW,
+        Stage.PENDING_IMPLEMENTATION,
+        Stage.PENDING_TEST,
+        Stage.PENDING_FINAL_REVIEW,
+        Stage.MERGE_REQUEST,
+        Stage.COMPLETED,
+    ),
+    #: No worktree, no merge: the planner answers and the task is done.
+    PipelineKey.ANALYSIS_ONLY: (
+        Stage.PENDING_MANAGER_REVIEW,
+        Stage.PENDING_ANALYSIS,
+        Stage.COMPLETED,
+    ),
+}
+
+#: The pipeline used when nobody expressed a preference.
+APP_DEFAULT_PIPELINE: PipelineKey = PipelineKey.PLAN_CODE_TEST
+
+#: Default pipeline and risk per task kind. The manager's choice wins over both.
+TASK_KIND_DEFAULTS: dict[TaskKind, tuple[PipelineKey, Risk]] = {
+    TaskKind.FEATURE: (PipelineKey.PLAN_CODE_TEST, Risk.MEDIUM),
+    TaskKind.BUGFIX: (PipelineKey.CODE_TEST, Risk.LOW),
+    TaskKind.REFACTOR: (PipelineKey.PLAN_CODE_REVIEW_TEST, Risk.HIGH),
+    TaskKind.DOCS: (PipelineKey.CODE_TEST, Risk.LOW),
+    TaskKind.ANALYSIS: (PipelineKey.ANALYSIS_ONLY, Risk.LOW),
+}
+
+#: Legal stage moves. Anything not listed here is a bug in the caller, and
+#: ``engine/state.py`` refuses it before it reaches the database.
+TRANSITIONS: dict[Stage, frozenset[Stage]] = {
+    Stage.PENDING_MANAGER_REVIEW: frozenset(
+        {
+            Stage.PENDING_ANALYSIS,
+            Stage.PENDING_IMPLEMENTATION,
+            Stage.PENDING_USER_DECISION,
+            Stage.ERROR_REVIEW,
+            Stage.PAUSED,
+            Stage.CANCELLED,
+        }
+    ),
+    Stage.PENDING_ANALYSIS: frozenset(
+        {
+            Stage.PENDING_IMPLEMENTATION,
+            # analysis_only ends here; the planner's answer *is* the deliverable.
+            Stage.COMPLETED,
+            Stage.PENDING_USER_DECISION,
+            Stage.ERROR_REVIEW,
+            Stage.PAUSED,
+            Stage.CANCELLED,
+        }
+    ),
+    Stage.PENDING_IMPLEMENTATION: frozenset(
+        {
+            Stage.PENDING_REVIEW,
+            Stage.PENDING_TEST,
+            Stage.PENDING_FINAL_REVIEW,
+            Stage.PENDING_USER_DECISION,
+            Stage.ERROR_REVIEW,
+            Stage.PAUSED,
+            Stage.CANCELLED,
+        }
+    ),
+    Stage.PENDING_REVIEW: frozenset(
+        {
+            Stage.PENDING_TEST,
+            Stage.PENDING_IMPLEMENTATION,
+            Stage.PENDING_USER_DECISION,
+            Stage.ERROR_REVIEW,
+            Stage.PAUSED,
+            Stage.CANCELLED,
+        }
+    ),
+    Stage.PENDING_TEST: frozenset(
+        {
+            Stage.PENDING_FINAL_REVIEW,
+            Stage.PENDING_IMPLEMENTATION,
+            Stage.PENDING_USER_DECISION,
+            Stage.ERROR_REVIEW,
+            Stage.PAUSED,
+            Stage.CANCELLED,
+        }
+    ),
+    Stage.PENDING_FINAL_REVIEW: frozenset(
+        {
+            Stage.MERGE_REQUEST,
+            Stage.PENDING_IMPLEMENTATION,
+            Stage.PENDING_USER_DECISION,
+            Stage.ERROR_REVIEW,
+            Stage.PAUSED,
+            Stage.CANCELLED,
+        }
+    ),
+    #: merge_request is executed by the platform, never by an AI turn: git
+    #: lives behind RepoWorkspacePort (PLAN-v2 D6).
+    Stage.MERGE_REQUEST: frozenset(
+        {
+            Stage.COMPLETED,
+            Stage.PENDING_IMPLEMENTATION,
+            Stage.ERROR_REVIEW,
+            Stage.PAUSED,
+            Stage.CANCELLED,
+        }
+    ),
+    Stage.PENDING_USER_DECISION: frozenset(
+        {
+            # The user answered, so the task goes back to triage: a manager
+            # decides the pipeline again with the new information.
+            Stage.PENDING_MANAGER_REVIEW,
+            Stage.ERROR_REVIEW,
+            Stage.PAUSED,
+            Stage.CANCELLED,
+        }
+    ),
+    Stage.ERROR_REVIEW: frozenset(
+        {
+            Stage.PENDING_ANALYSIS,
+            Stage.PENDING_IMPLEMENTATION,
+            Stage.PENDING_USER_DECISION,
+            Stage.FAILED,
+            Stage.PAUSED,
+            Stage.CANCELLED,
+        }
+    ),
+    #: A paused task resumes through triage as well: the manager re-reads the
+    # task before spending more budget on it.
+    Stage.PAUSED: frozenset(
+        {Stage.PENDING_MANAGER_REVIEW, Stage.ERROR_REVIEW, Stage.CANCELLED}
+    ),
+    Stage.COMPLETED: frozenset(),
+    Stage.CANCELLED: frozenset(),
+    Stage.FAILED: frozenset(),
+}
+
+
+class StageActor(NamedTuple):
+    """Who acts in a stage: a role and the mode it must use."""
+
+    role: RoleKey
+    mode: Mode
+
+
+#: stage -> (role, mode). ``None`` means "no AI turn acts here": the stage is
+#: either terminal, platform-executed (merge_request) or waiting on a human.
+STAGE_ROLE: dict[Stage, StageActor | None] = {
+    Stage.PENDING_MANAGER_REVIEW: StageActor(RoleKey.MANAGER, Mode.TRIAGE),
+    Stage.PENDING_ANALYSIS: StageActor(RoleKey.PLANNER, Mode.WORK),
+    Stage.PENDING_IMPLEMENTATION: StageActor(RoleKey.DEVELOPER, Mode.WORK),
+    Stage.PENDING_REVIEW: StageActor(RoleKey.REVIEWER, Mode.WORK),
+    Stage.PENDING_TEST: StageActor(RoleKey.TESTER, Mode.WORK),
+    Stage.PENDING_FINAL_REVIEW: StageActor(RoleKey.MANAGER, Mode.FINAL),
+    Stage.MERGE_REQUEST: None,
+    Stage.COMPLETED: None,
+    Stage.PENDING_USER_DECISION: None,
+    Stage.ERROR_REVIEW: StageActor(RoleKey.MANAGER, Mode.ERROR),
+    Stage.PAUSED: None,
+    Stage.CANCELLED: None,
+    Stage.FAILED: None,
+}
+
+#: The coarse ``tasks.status`` each stage maps onto. Existing APIs and the old
+#: worker read that column, so the engine keeps it truthful at every move
+#: instead of introducing a second source of truth.
+STAGE_STATUS: dict[Stage, TaskStatus] = {
+    Stage.PENDING_MANAGER_REVIEW: TaskStatus.RUNNING,
+    Stage.PENDING_ANALYSIS: TaskStatus.RUNNING,
+    Stage.PENDING_IMPLEMENTATION: TaskStatus.RUNNING,
+    Stage.PENDING_REVIEW: TaskStatus.RUNNING,
+    Stage.PENDING_TEST: TaskStatus.RUNNING,
+    Stage.PENDING_FINAL_REVIEW: TaskStatus.RUNNING,
+    Stage.MERGE_REQUEST: TaskStatus.RUNNING,
+    Stage.COMPLETED: TaskStatus.DONE,
+    Stage.PENDING_USER_DECISION: TaskStatus.WAITING_APPROVAL,
+    Stage.ERROR_REVIEW: TaskStatus.RUNNING,
+    Stage.PAUSED: TaskStatus.WAITING_APPROVAL,
+    Stage.CANCELLED: TaskStatus.CANCELLED,
+    Stage.FAILED: TaskStatus.FAILED,
+}
+
+#: Fixed targets for outcomes that are not "the next stage": work that is sent
+#: back always returns to the developer, because that is the role that can
+#: actually change the diff.
+_OUTCOME_TARGET: dict[Outcome, Stage] = {
+    Outcome.REVISE: Stage.PENDING_IMPLEMENTATION,
+    Outcome.RESOLVE: Stage.PENDING_IMPLEMENTATION,
+    Outcome.ASK_USER: Stage.PENDING_USER_DECISION,
+    Outcome.ERROR: Stage.ERROR_REVIEW,
+    Outcome.REJECT: Stage.CANCELLED,
+}
+
+
+def is_terminal(stage: Stage) -> bool:
+    return stage in TERMINAL_STAGES
+
+
+def is_side_stage(stage: Stage) -> bool:
+    """True when the stage is not part of any pipeline's ordered list."""
+    return stage in SIDE_STAGES
+
+
+def can_transition_stage(current: Stage, target: Stage) -> bool:
+    return target in TRANSITIONS.get(current, frozenset())
+
+
+def pipeline_stages(pipeline: PipelineKey | str) -> tuple[Stage, ...]:
+    """The ordered stages of a pipeline (including the terminal)."""
+    return PIPELINES[PipelineKey(pipeline)]
+
+
+def requires_worktree(pipeline: PipelineKey | str) -> bool:
+    """``analysis_only`` never creates a worktree and never merges."""
+    return PipelineKey(pipeline) is not PipelineKey.ANALYSIS_ONLY
+
+
+def next_stage(pipeline: PipelineKey | str, current: Stage | str, outcome: Outcome | str) -> Stage:
+    """The stage a task moves to after a turn ends.
+
+    ``Outcome.OK`` walks the pipeline list one step forward (past the last
+    stage there is nothing left, which only happens for a malformed call);
+    every other outcome has a fixed target. The result is *not* checked
+    against :data:`TRANSITIONS` here — callers go through
+    ``engine/state.py``, which refuses illegal moves.
+    """
+    stage = Stage(current)
+    verdict = Outcome(outcome)
+    if verdict is Outcome.OK:
+        stages = PIPELINES[PipelineKey(pipeline)]
+        index = stages.index(stage)
+        return stages[index + 1] if index + 1 < len(stages) else stage
+    try:
+        return _OUTCOME_TARGET[verdict]
+    except KeyError as exc:  # pragma: no cover - the enum is closed
+        raise ValueError(f"نتیجهٔ ناشناخته برای مرحله: {verdict}") from exc
+
+
+def resolve_pipeline(
+    *,
+    manager_choice: PipelineKey | str | None = None,
+    task_kind: TaskKind | str | None = None,
+    app_default: PipelineKey | str | None = None,
+    has_active_reviewer: bool = True,
+) -> PipelineKey:
+    """Decide which pipeline a task gets.
+
+    Priority: the manager's explicit choice, then the task kind's default,
+    then the app profile's default, then ``plan_code_test``.
+
+    A review pipeline without an active reviewer falls back to
+    ``plan_code_test`` — including an explicit manager choice, because a
+    review stage nobody can act on is a dead end. The manager who really
+    wants a reviewer should hire one (``HIRE role="reviewer"``), not start a
+    task that will wait forever.
+    """
+    chosen: PipelineKey | None = None
+    if manager_choice:
+        chosen = PipelineKey(manager_choice)
+    elif task_kind:
+        chosen = TASK_KIND_DEFAULTS[TaskKind(task_kind)][0]
+    elif app_default:
+        chosen = PipelineKey(app_default)
+    if chosen is None:
+        chosen = APP_DEFAULT_PIPELINE
+    if chosen is PipelineKey.PLAN_CODE_REVIEW_TEST and not has_active_reviewer:
+        return PipelineKey.PLAN_CODE_TEST
+    return chosen
+
+
+def risk_for(task_kind: TaskKind | str) -> Risk:
+    """The risk a task kind carries when the manager did not set one."""
+    return TASK_KIND_DEFAULTS[TaskKind(task_kind)][1]
+
+
+def coarse_status(stage: Stage | str) -> TaskStatus:
+    return STAGE_STATUS[Stage(stage)]
+
+
+@dataclass(slots=True, kw_only=True)
+class TaskPipelineState:
+    """The engine's per-task record: pipeline, position and loop guards.
+
+    Lives in a 1:1 side table (``task_pipeline_state``) because ``tasks`` is
+    a locked table: the engine must not add columns to it.
+
+    ``version`` is the optimistic lock. ``advance`` only writes when the
+    row still has the version the caller read, so two workers moving the
+    same task cannot both win.
+    """
+
+    task_id: str
+    organization_id: str
+    project_id: str | None = None
+    pipeline: PipelineKey = APP_DEFAULT_PIPELINE
+    kind: TaskKind = TaskKind.FEATURE
+    risk: Risk = Risk.MEDIUM
+    stage: Stage = Stage.PENDING_MANAGER_REVIEW
+    cycle: int = 0
+    max_cycles: int = 6
+    branch: str = ""
+    #: Hash of the last diff, for the no-progress gate.
+    last_diff_hash: str = ""
+    same_diff_count: int = 0
+    retry_count: int = 0
+    waiver_reason: str = ""
+    parent_task_id: str | None = None
+    depends_on: list[str] = field(default_factory=list)
+    version: int = 0
+    updated_at: datetime = field(default_factory=utcnow)
+
+    @property
+    def status(self) -> TaskStatus:
+        return coarse_status(self.stage)
+
+    @property
+    def needs_turn(self) -> bool:
+        """True when the stage is waiting for an AI turn."""
+        actor = STAGE_ROLE.get(self.stage)
+        return actor is not None and not is_terminal(self.stage)
+
+    @property
+    def is_blocked(self) -> bool:
+        """True while the task waits on a human (decision, pause or budget)."""
+        return self.stage in SIDE_STAGES and not is_terminal(self.stage)
+
+    def actor(self) -> StageActor | None:
+        return STAGE_ROLE.get(self.stage)
+
+
+__all__ = [
+    "APP_DEFAULT_PIPELINE",
+    "PIPELINES",
+    "SIDE_STAGES",
+    "STAGE_ROLE",
+    "STAGE_STATUS",
+    "TASK_KIND_DEFAULTS",
+    "TERMINAL_STAGES",
+    "TRANSITIONS",
+    "Mode",
+    "Outcome",
+    "PipelineKey",
+    "Risk",
+    "Stage",
+    "StageActor",
+    "TaskKind",
+    "TaskPipelineState",
+    "can_transition_stage",
+    "coarse_status",
+    "is_side_stage",
+    "is_terminal",
+    "next_stage",
+    "pipeline_stages",
+    "requires_worktree",
+    "resolve_pipeline",
+    "risk_for",
+]

apps/api/app/infrastructure/db/models_engine.py

@@ -5,6 +5,170 @@ Use ``from app.core.database import Base`` and the helpers in
 ``app.infrastructure.db.models`` here: that module imports this one (import cycle).
 Foreign keys to existing tables are written as plain strings, e.g.
 ``ForeignKey("organizations.id", ondelete="CASCADE")``. See docs/plan/PLAN-v2.md §6.
+
+Four tables, deliberately split so ``tasks`` and ``employees`` stay untouched
+(they are locked):
+
+* ``task_pipeline_state`` — the engine's own view of a task: pipeline, stage,
+  cycle counters, branch and the optimistic ``version``. One row per task.
+* ``employee_profiles`` — which granted model an employee runs on, and how
+  busy it is. Model identity never leaves the server.
+* ``engine_turns`` — one row per agent turn: the audit trail of who thought
+  what, with a bounded reply excerpt.
+* ``gate_results`` — one row per gate verdict (tests, diff guard, no-progress,
+  green), with the detail that justifies it.
+
+Both side tables cascade from their parent, so deleting a task or an employee
+never leaves an orphaned engine row behind.
 """
 
 from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import (
+    JSON,
+    Boolean,
+    DateTime,
+    ForeignKey,
+    Index,
+    Integer,
+    String,
+    Text,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.database import Base
+from app.domain.base import utcnow
+from app.infrastructure.db.columns import created, pk
+
+#: The engine's default limits, mirroring the settings block in ``config.py``.
+#: They are column defaults (not settings) so that a database written by an
+#: older deployment still carries a sane value.
+DEFAULT_MAX_CYCLES = 6
+
+
+class TaskPipelineStateRow(Base):
+    __tablename__ = "task_pipeline_state"
+
+    task_id: Mapped[str] = mapped_column(
+        ForeignKey("tasks.id", ondelete="CASCADE"), primary_key=True
+    )
+    #: Denormalised from the task so the tenant filter stays a single-table
+    #: WHERE: the scheduler's hot query never joins back to ``tasks``.
+    organization_id: Mapped[str] = mapped_column(
+        ForeignKey("organizations.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    project_id: Mapped[str | None] = mapped_column(
+        ForeignKey("projects.id", ondelete="CASCADE"), nullable=True
+    )
+    pipeline: Mapped[str] = mapped_column(
+        String(32), default="plan_code_test", nullable=False
+    )
+    kind: Mapped[str] = mapped_column(String(16), default="feature", nullable=False)
+    risk: Mapped[str] = mapped_column(String(8), default="medium", nullable=False)
+    #: Indexed: every scheduler tick asks "which tasks are at a turn stage".
+    stage: Mapped[str] = mapped_column(
+        String(32), default="pending_manager_review", nullable=False, index=True
+    )
+    cycle: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    max_cycles: Mapped[int] = mapped_column(
+        Integer, default=DEFAULT_MAX_CYCLES, nullable=False
+    )
+    branch: Mapped[str] = mapped_column(String(120), default="", nullable=False)
+    last_diff_hash: Mapped[str] = mapped_column(String(64), default="", nullable=False)
+    same_diff_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    retry_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    #: A waived test pass must always be explainable: the reason is stored,
+    #: never inferred (PLAN-v2 §3, green gate).
+    waiver_reason: Mapped[str] = mapped_column(Text, default="", nullable=False)
+    parent_task_id: Mapped[str | None] = mapped_column(
+        ForeignKey("tasks.id", ondelete="SET NULL"), nullable=True
+    )
+    depends_on: Mapped[list] = mapped_column(JSON, default=list, nullable=False)
+    #: Optimistic lock. Every stage move bumps it; a stale write is refused.
+    version: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=utcnow, onupdate=utcnow, nullable=False
+    )
+
+
+class EmployeeProfileRow(Base):
+    __tablename__ = "employee_profiles"
+    __table_args__ = (Index("ix_employee_profile_busy", "busy_task_id"),)
+
+    employee_id: Mapped[str] = mapped_column(
+        ForeignKey("employees.id", ondelete="CASCADE"), primary_key=True
+    )
+    #: A granted model id (``org_model_grants.ai_model_id``). Nullable until the
+    #: owner grants something: an employee exists but is inactive, which is a
+    #: state the UI shows instead of hiding the employee.
+    ai_model_id: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    effort: Mapped[str] = mapped_column(String(16), default="medium", nullable=False)
+    roles: Mapped[str] = mapped_column(String(255), default="", nullable=False)
+    hired_by: Mapped[str] = mapped_column(String(8), default="user", nullable=False)
+    busy_task_id: Mapped[str | None] = mapped_column(
+        ForeignKey("tasks.id", ondelete="SET NULL"), nullable=True
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=utcnow, onupdate=utcnow, nullable=False
+    )
+
+
+class EngineTurnRow(Base):
+    __tablename__ = "engine_turns"
+    __table_args__ = (
+        Index("ix_engine_turn_task_stage", "task_id", "stage"),
+        Index("ix_engine_turn_run", "run_id"),
+    )
+
+    id: Mapped[str] = pk()
+    task_id: Mapped[str] = mapped_column(
+        ForeignKey("tasks.id", ondelete="CASCADE"), nullable=False
+    )
+    run_id: Mapped[str] = mapped_column(
+        ForeignKey("runs.id", ondelete="CASCADE"), nullable=False
+    )
+    employee_id: Mapped[str | None] = mapped_column(
+        ForeignKey("employees.id", ondelete="SET NULL"), nullable=True
+    )
+    role: Mapped[str] = mapped_column(String(24), default="developer", nullable=False)
+    mode: Mapped[str] = mapped_column(String(16), default="work", nullable=False)
+    stage: Mapped[str] = mapped_column(String(32), default="", nullable=False)
+    prompt_chars: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    #: Bounded on write: a reply is unbounded, a row is not.
+    reply_excerpt: Mapped[str] = mapped_column(Text, default="", nullable=False)
+    tags: Mapped[dict] = mapped_column(JSON, default=dict, nullable=False)
+    tool_calls: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    ok: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
+    error_kind: Mapped[str] = mapped_column(String(16), default="", nullable=False)
+    started_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=utcnow, nullable=False
+    )
+    finished_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    created_at: Mapped[datetime] = created()
+
+
+class GateResultRow(Base):
+    __tablename__ = "gate_results"
+    __table_args__ = (Index("ix_gate_result_task_gate", "task_id", "gate"),)
+
+    id: Mapped[str] = pk()
+    task_id: Mapped[str] = mapped_column(
+        ForeignKey("tasks.id", ondelete="CASCADE"), nullable=False
+    )
+    gate: Mapped[str] = mapped_column(String(16), default="tests", nullable=False)
+    ok: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
+    detail: Mapped[str] = mapped_column(Text, default="", nullable=False)
+    created_at: Mapped[datetime] = created()
+
+
+__all__ = [
+    "DEFAULT_MAX_CYCLES",
+    "EmployeeProfileRow",
+    "EngineTurnRow",
+    "GateResultRow",
+    "TaskPipelineStateRow",
+]

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

@@ -0,0 +1,475 @@
+"""Repositories for the AI-workforce engine's own tables.
+
+Four repositories, one per engine table, all following the house rule: the
+ORM row never leaves this module — the domain dataclasses in
+``app.domain.workforce.pipeline`` / ``engine`` are what the application layer
+holds.
+
+The interesting one is :class:`PipelineStateRepository`. Its ``advance`` is an
+optimistic compare-and-set: the ``UPDATE`` carries the caller's expected stage
+and version in its ``WHERE`` clause, so two workers moving the same task
+cannot both win — the loser sees zero updated rows and gets a
+:class:`StaleStageError` instead of silently overwriting the winner.
+"""
+
+from __future__ import annotations
+
+from sqlalchemy import func, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.domain.base import utcnow
+from app.domain.workforce.engine import (
+    EmployeeProfile,
+    EngineTurn,
+    GateName,
+    GateResult,
+)
+from app.domain.workforce.models import RoleKey
+from app.domain.workforce.pipeline import (
+    TRANSITIONS,
+    Mode,
+    PipelineKey,
+    Risk,
+    Stage,
+    TaskKind,
+    TaskPipelineState,
+)
+from app.infrastructure.db.models_engine import (
+    EmployeeProfileRow,
+    EngineTurnRow,
+    GateResultRow,
+    TaskPipelineStateRow,
+)
+
+
+class StaleStageError(Exception):
+    """Another worker moved the task first; the caller must re-read it."""
+
+
+class UnknownTaskError(LookupError):
+    """The task has no pipeline state row (it was never registered)."""
+
+
+class InvalidTransitionError(ValueError):
+    """The requested move is not in the stage transition table."""
+
+
+# ---------------------------------------------------------------------------
+# task_pipeline_state
+# ---------------------------------------------------------------------------
+
+#: Columns ``patch`` may touch. An allow-list, so a typo can never become SQL.
+_PATCHABLE = frozenset(
+    {
+        "pipeline",
+        "kind",
+        "risk",
+        "stage",
+        "cycle",
+        "max_cycles",
+        "branch",
+        "last_diff_hash",
+        "same_diff_count",
+        "retry_count",
+        "waiver_reason",
+        "parent_task_id",
+        "depends_on",
+    }
+)
+
+
+def _state(row: TaskPipelineStateRow) -> TaskPipelineState:
+    return TaskPipelineState(
+        task_id=row.task_id,
+        organization_id=row.organization_id,
+        project_id=row.project_id,
+        pipeline=PipelineKey(row.pipeline),
+        kind=TaskKind(row.kind),
+        risk=Risk(row.risk),
+        stage=Stage(row.stage),
+        cycle=row.cycle,
+        max_cycles=row.max_cycles,
+        branch=row.branch,
+        last_diff_hash=row.last_diff_hash,
+        same_diff_count=row.same_diff_count,
+        retry_count=row.retry_count,
+        waiver_reason=row.waiver_reason,
+        parent_task_id=row.parent_task_id,
+        depends_on=list(row.depends_on or []),
+        version=row.version,
+        updated_at=row.updated_at,
+    )
+
+
+def _legal_targets(stage: Stage) -> frozenset[Stage]:
+    """The stage moves the domain allows from here."""
+    return TRANSITIONS.get(stage, frozenset())
+
+
+class PipelineStateRepository:
+    """1:1 with a task: pipeline, stage, cycle guards and the version lock."""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+
+    async def get(self, task_id: str) -> TaskPipelineState | None:
+        row = await self.session.get(TaskPipelineStateRow, task_id)
+        return _state(row) if row else None
+
+    async def get_or_raise(self, task_id: str) -> TaskPipelineState:
+        state = await self.get(task_id)
+        if state is None:
+            raise UnknownTaskError(task_id)
+        return state
+
+    async def list_for_org(
+        self, organization_id: str, *, limit: int = 500
+    ) -> list[TaskPipelineState]:
+        stmt = (
+            select(TaskPipelineStateRow)
+            .where(TaskPipelineStateRow.organization_id == organization_id)
+            .order_by(TaskPipelineStateRow.updated_at.desc())
+            .limit(min(limit, 1000))
+        )
+        return [_state(r) for r in (await self.session.execute(stmt)).scalars()]
+
+    async def list_at_stages(
+        self, stages: tuple[Stage, ...], *, limit: int = 50
+    ) -> list[TaskPipelineState]:
+        """The scheduler's hot query: tasks currently at a turn stage."""
+        if not stages:
+            return []
+        stmt = (
+            select(TaskPipelineStateRow)
+            .where(TaskPipelineStateRow.stage.in_([str(s) for s in stages]))
+            .order_by(TaskPipelineStateRow.updated_at.asc())
+            .limit(min(limit, 500))
+        )
+        return [_state(r) for r in (await self.session.execute(stmt)).scalars()]
+
+    async def list_dependents(self, task_id: str) -> list[TaskPipelineState]:
+        """Tasks that declare ``task_id`` in ``depends_on`` and are not closed.
+
+        JSON columns have no portable containment operator (SQLite differs from
+        PostgreSQL), so the match happens in Python. The set is small — it is
+        the tasks waiting on one specific task — so this is not a scan of the
+        table.
+        """
+        stmt = select(TaskPipelineStateRow).where(
+            TaskPipelineStateRow.stage.notin_(
+                [str(Stage.COMPLETED), str(Stage.CANCELLED), str(Stage.FAILED)]
+            )
+        )
+        dependents: list[TaskPipelineState] = []
+        for row in (await self.session.execute(stmt)).scalars():
+            if task_id in list(row.depends_on or []):
+                dependents.append(_state(row))
+        return dependents
+
+    async def create(self, state: TaskPipelineState) -> TaskPipelineState:
+        row = TaskPipelineStateRow(
+            task_id=state.task_id,
+            organization_id=state.organization_id,
+            project_id=state.project_id,
+            pipeline=str(state.pipeline),
+            kind=str(state.kind),
+            risk=str(state.risk),
+            stage=str(state.stage),
+            cycle=state.cycle,
+            max_cycles=state.max_cycles,
+            branch=state.branch,
+            last_diff_hash=state.last_diff_hash,
+            same_diff_count=state.same_diff_count,
+            retry_count=state.retry_count,
+            waiver_reason=state.waiver_reason,
+            parent_task_id=state.parent_task_id,
+            depends_on=list(state.depends_on),
+            version=state.version,
+        )
+        self.session.add(row)
+        await self.session.flush()
+        return _state(row)
+
+    async def advance(
+        self, *, task_id: str, expected_from: Stage, to: Stage
+    ) -> TaskPipelineState:
+        """Move a task one stage forward, but only if nobody moved it yet.
+
+        The row is read for its current version, then written with both the
+        expected stage and that version in the ``WHERE``. If another worker
+        got there first, zero rows match and :class:`StaleStageError` is
+        raised — the caller re-reads and retries (or gives up cleanly).
+        """
+        row = await self.session.get(TaskPipelineStateRow, task_id)
+        if row is None:
+            raise UnknownTaskError(task_id)
+        if Stage(row.stage) is not Stage(expected_from):
+            raise StaleStageError(
+                f"task {task_id} is at {row.stage}, not {expected_from.value}"
+            )
+        if Stage(to) not in _legal_targets(Stage(row.stage)):
+            raise InvalidTransitionError(f"{row.stage} -> {to.value} is not allowed")
+        expected_version = row.version
+        # synchronize_session="fetch" keeps the identity-mapped object truthful
+        # (SQLite has no RETURNING for this), so the check below reads what the
+        # database actually did rather than what we hoped it would do.
+        await self.session.execute(
+            update(TaskPipelineStateRow)
+            .where(
+                TaskPipelineStateRow.task_id == task_id,
+                TaskPipelineStateRow.stage == str(expected_from),
+                TaskPipelineStateRow.version == expected_version,
+            )
+            .values(stage=str(to), version=expected_version + 1, updated_at=utcnow())
+            .execution_options(synchronize_session="fetch")
+        )
+        await self.session.flush()
+        # A version that did not move means the WHERE matched nothing: another
+        # worker wrote this row first and this move never happened.
+        if row.version == expected_version or Stage(row.stage) is not Stage(to):
+            raise StaleStageError(f"task {task_id} was moved by another worker")
+        return _state(row)
+
+    async def patch(
+        self, *, task_id: str, expected_version: int, **fields: object
+    ) -> TaskPipelineState:
+        """Change counters without moving the stage, under the version lock."""
+        unknown = set(fields) - _PATCHABLE
+        if unknown:
+            raise ValueError(f"non-patchable columns: {sorted(unknown)}")
+        row = await self.session.get(TaskPipelineStateRow, task_id)
+        if row is None:
+            raise UnknownTaskError(task_id)
+        if not fields:
+            return _state(row)
+        before = row.version
+        await self.session.execute(
+            update(TaskPipelineStateRow)
+            .where(
+                TaskPipelineStateRow.task_id == task_id,
+                TaskPipelineStateRow.version == expected_version,
+            )
+            .values(**fields, version=expected_version + 1, updated_at=utcnow())
+            .execution_options(synchronize_session="fetch")
+        )
+        await self.session.flush()
+        if row.version == before:
+            raise StaleStageError(f"task {task_id} was moved by another worker")
+        return _state(row)
+
+    async def count_at(self, stage: Stage) -> int:
+        stmt = (
+            select(func.count())
+            .select_from(TaskPipelineStateRow)
+            .where(TaskPipelineStateRow.stage == str(stage))
+        )
+        return int((await self.session.execute(stmt)).scalar_one())
+
+
+# ---------------------------------------------------------------------------
+# employee_profiles
+# ---------------------------------------------------------------------------
+
+
+def _profile(row: EmployeeProfileRow) -> EmployeeProfile:
+    return EmployeeProfile(
+        employee_id=row.employee_id,
+        ai_model_id=row.ai_model_id,
+        effort=row.effort,
+        roles=[r.strip() for r in (row.roles or "").split(",") if r.strip()],
+        hired_by=row.hired_by,
+        busy_task_id=row.busy_task_id,
+        updated_at=row.updated_at,
+    )
+
+
+class EmployeeProfileRepository:
+    """The engine's view of an employee: granted model, roles, current task."""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+
+    async def get(self, employee_id: str) -> EmployeeProfile | None:
+        row = await self.session.get(EmployeeProfileRow, employee_id)
+        return _profile(row) if row else None
+
+    async def list_free_with_role(self, role: str, *, limit: int = 10) -> list[str]:
+        """Employee ids that hold ``role`` and are not busy, least busy first."""
+        stmt = select(EmployeeProfileRow).where(
+            EmployeeProfileRow.busy_task_id.is_(None),
+            EmployeeProfileRow.roles.like(f"%{role}%"),
+        )
+        return [r.employee_id for r in (await self.session.execute(stmt)).scalars()][:limit]
+
+    async def upsert(self, profile: EmployeeProfile) -> EmployeeProfile:
+        row = await self.session.get(EmployeeProfileRow, profile.employee_id)
+        if row is None:
+            row = EmployeeProfileRow(
+                employee_id=profile.employee_id,
+                ai_model_id=profile.ai_model_id,
+                effort=profile.effort,
+                roles=",".join(profile.role_list()),
+                hired_by=profile.hired_by,
+                busy_task_id=profile.busy_task_id,
+            )
+            self.session.add(row)
+        else:
+            row.ai_model_id = profile.ai_model_id
+            row.effort = profile.effort
+            row.roles = ",".join(profile.role_list())
+            row.hired_by = profile.hired_by
+            row.busy_task_id = profile.busy_task_id
+        await self.session.flush()
+        return _profile(row)
+
+    async def set_busy(self, *, employee_id: str, task_id: str | None) -> None:
+        row = await self.session.get(EmployeeProfileRow, employee_id)
+        if row is None:
+            return
+        row.busy_task_id = task_id
+        await self.session.flush()
+
+    async def set_model(self, *, employee_id: str, ai_model_id: str | None) -> None:
+        row = await self.session.get(EmployeeProfileRow, employee_id)
+        if row is None:
+            return
+        row.ai_model_id = ai_model_id
+        await self.session.flush()
+
+
+# ---------------------------------------------------------------------------
+# engine_turns
+# ---------------------------------------------------------------------------
+
+
+def _turn(row: EngineTurnRow) -> EngineTurn:
+    return EngineTurn(
+        id=row.id,
+        task_id=row.task_id,
+        run_id=row.run_id,
+        employee_id=row.employee_id,
+        role=RoleKey(row.role),
+        mode=Mode(row.mode),
+        stage=Stage(row.stage) if row.stage else Stage.PENDING_IMPLEMENTATION,
+        prompt_chars=row.prompt_chars,
+        reply_excerpt=row.reply_excerpt,
+        tags=dict(row.tags or {}),
+        tool_calls=row.tool_calls,
+        ok=row.ok,
+        error_kind=row.error_kind,
+        started_at=row.started_at,
+        finished_at=row.finished_at,
+    )
+
+
+class EngineTurnRepository:
+    """One row per agent turn: the audit trail of who thought what."""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+
+    async def create(self, turn: EngineTurn) -> EngineTurn:
+        row = EngineTurnRow(
+            id=turn.id,
+            task_id=turn.task_id,
+            run_id=turn.run_id,
+            employee_id=turn.employee_id,
+            role=str(turn.role),
+            mode=str(turn.mode),
+            stage=str(turn.stage),
+            prompt_chars=turn.prompt_chars,
+            reply_excerpt=turn.reply_excerpt,
+            tags=dict(turn.tags),
+            tool_calls=turn.tool_calls,
+            ok=turn.ok,
+            error_kind=turn.error_kind,
+            started_at=turn.started_at,
+            finished_at=turn.finished_at,
+        )
+        self.session.add(row)
+        await self.session.flush()
+        return _turn(row)
+
+    async def list_for_task(self, task_id: str, *, limit: int = 200) -> list[EngineTurn]:
+        stmt = (
+            select(EngineTurnRow)
+            .where(EngineTurnRow.task_id == task_id)
+            .order_by(EngineTurnRow.started_at.asc())
+            .limit(min(limit, 1000))
+        )
+        return [_turn(r) for r in (await self.session.execute(stmt)).scalars()]
+
+    async def list_for_run(self, run_id: str) -> list[EngineTurn]:
+        stmt = (
+            select(EngineTurnRow)
+            .where(EngineTurnRow.run_id == run_id)
+            .order_by(EngineTurnRow.started_at.asc())
+        )
+        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 = ""
+    ) -> None:
+        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
+        row.finished_at = utcnow()
+        await self.session.flush()
+
+
+# ---------------------------------------------------------------------------
+# gate_results
+# ---------------------------------------------------------------------------
+
+
+def _gate(row: GateResultRow) -> GateResult:
+    return GateResult(
+        id=row.id,
+        task_id=row.task_id,
+        gate=GateName(row.gate),
+        ok=row.ok,
+        detail=row.detail,
+        created_at=row.created_at,
+    )
+
+
+class GateResultRepository:
+    """One row per gate verdict, with the detail that justifies it."""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+
+    async def create(self, result: GateResult) -> GateResult:
+        row = GateResultRow(
+            id=result.id,
+            task_id=result.task_id,
+            gate=str(result.gate),
+            ok=result.ok,
+            detail=result.detail,
+            created_at=result.created_at,
+        )
+        self.session.add(row)
+        await self.session.flush()
+        return _gate(row)
+
+    async def list_for_task(self, task_id: str, *, limit: int = 200) -> list[GateResult]:
+        stmt = (
+            select(GateResultRow)
+            .where(GateResultRow.task_id == task_id)
+            .order_by(GateResultRow.created_at.asc())
+            .limit(min(limit, 1000))
+        )
+        return [_gate(r) for r in (await self.session.execute(stmt)).scalars()]
+
+    async def latest_for(self, task_id: str, gate: str) -> GateResult | None:
+        stmt = (
+            select(GateResultRow)
+            .where(GateResultRow.task_id == task_id, GateResultRow.gate == gate)
+            .order_by(GateResultRow.created_at.desc())
+            .limit(1)
+        )
+        row = (await self.session.execute(stmt)).scalar_one_or_none()
+        return _gate(row) if row else None

apps/api/tests/test_engine_pipeline.py

@@ -0,0 +1,523 @@
+"""The engine's state machine: pipelines, transitions, optimistic moves.
+
+What these tests defend:
+
+* every pipeline really walks from the manager's review to ``completed`` on
+  the happy path — a stage nobody can leave is a stuck task
+* an illegal move is refused *before* it reaches the database, so a bug in a
+  later phase cannot corrupt a task's position
+* two workers racing on one task: exactly one wins, the loser learns about it
+  instead of silently overwriting the winner
+* a failed dependency parks its dependents with a human instead of leaving
+  them waiting for work that will never happen (agybot bug M1)
+"""
+
+from __future__ import annotations
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import delete
+
+from app.application.engine.state import advance, park_dependents
+from app.core.errors import ConflictError, NotFoundError
+from app.domain.base import new_id
+from app.domain.workforce.engine import (
+    EmployeeProfile,
+    EngineTurn,
+    GateName,
+    GateResult,
+)
+from app.domain.workforce.models import RoleKey, TaskStatus
+from app.domain.workforce.pipeline import (
+    APP_DEFAULT_PIPELINE,
+    PIPELINES,
+    SIDE_STAGES,
+    STAGE_ROLE,
+    STAGE_STATUS,
+    TERMINAL_STAGES,
+    TRANSITIONS,
+    Mode,
+    Outcome,
+    PipelineKey,
+    Risk,
+    Stage,
+    StageActor,
+    TaskKind,
+    TaskPipelineState,
+    can_transition_stage,
+    coarse_status,
+    is_side_stage,
+    is_terminal,
+    next_stage,
+    pipeline_stages,
+    requires_worktree,
+    resolve_pipeline,
+    risk_for,
+)
+from app.infrastructure.db.models import (
+    EmployeeRow,
+    OrganizationRow,
+    RunRow,
+    TaskRow,
+    UserRow,
+)
+from app.infrastructure.repositories.engine import (
+    EmployeeProfileRepository,
+    EngineTurnRepository,
+    GateResultRepository,
+    PipelineStateRepository,
+    StaleStageError,
+)
+
+# ---------------------------------------------------------------------------
+# Pure logic
+# ---------------------------------------------------------------------------
+
+#: The happy path of each pipeline: the stages after the manager's review.
+_HAPPY_PATHS: dict[PipelineKey, tuple[Stage, ...]] = {
+    PipelineKey.PLAN_CODE_TEST: (
+        Stage.PENDING_ANALYSIS,
+        Stage.PENDING_IMPLEMENTATION,
+        Stage.PENDING_TEST,
+        Stage.PENDING_FINAL_REVIEW,
+        Stage.MERGE_REQUEST,
+        Stage.COMPLETED,
+    ),
+    PipelineKey.PLAN_CODE_REVIEW_TEST: (
+        Stage.PENDING_ANALYSIS,
+        Stage.PENDING_IMPLEMENTATION,
+        Stage.PENDING_REVIEW,
+        Stage.PENDING_TEST,
+        Stage.PENDING_FINAL_REVIEW,
+        Stage.MERGE_REQUEST,
+        Stage.COMPLETED,
+    ),
+    PipelineKey.CODE_TEST: (
+        Stage.PENDING_IMPLEMENTATION,
+        Stage.PENDING_TEST,
+        Stage.PENDING_FINAL_REVIEW,
+        Stage.MERGE_REQUEST,
+        Stage.COMPLETED,
+    ),
+    PipelineKey.ANALYSIS_ONLY: (Stage.PENDING_ANALYSIS, Stage.COMPLETED),
+}
+
+
+def test_every_pipeline_walks_from_manager_review_to_completed():
+    """The happy path, one ``Outcome.OK`` step at a time — all four pipelines."""
+    for pipeline, walked in _HAPPY_PATHS.items():
+        stages = pipeline_stages(pipeline)
+        assert stages[0] is Stage.PENDING_MANAGER_REVIEW, pipeline
+        assert stages[-1] is Stage.COMPLETED, pipeline
+        assert stages == (Stage.PENDING_MANAGER_REVIEW, *walked), pipeline
+        previous = Stage.PENDING_MANAGER_REVIEW
+        for wanted in walked:
+            assert next_stage(pipeline, previous, Outcome.OK) is wanted, pipeline
+            # Every step the happy path takes is also a legal database move.
+            assert can_transition_stage(previous, wanted) is True, (pipeline, previous, wanted)
+            previous = wanted
+
+
+def test_analysis_only_needs_no_worktree_and_the_others_do():
+    assert requires_worktree(PipelineKey.ANALYSIS_ONLY) is False
+    for pipeline in PipelineKey:
+        if pipeline is PipelineKey.ANALYSIS_ONLY:
+            continue
+        assert requires_worktree(pipeline) is True, pipeline
+
+
+def test_sent_back_work_returns_to_the_developer():
+    """A revise from review, test or final review cycles the same task."""
+    for stage in (Stage.PENDING_REVIEW, Stage.PENDING_TEST, Stage.PENDING_FINAL_REVIEW):
+        assert next_stage(
+            PipelineKey.PLAN_CODE_REVIEW_TEST, stage, Outcome.REVISE
+        ) is Stage.PENDING_IMPLEMENTATION
+    assert next_stage(
+        PipelineKey.CODE_TEST, Stage.ERROR_REVIEW, Outcome.RESOLVE
+    ) is Stage.PENDING_IMPLEMENTATION
+
+
+def test_outcomes_route_to_their_fixed_targets():
+    assert next_stage(
+        PipelineKey.PLAN_CODE_TEST, Stage.PENDING_ANALYSIS, Outcome.ASK_USER
+    ) is Stage.PENDING_USER_DECISION
+    assert next_stage(
+        PipelineKey.PLAN_CODE_TEST, Stage.PENDING_ANALYSIS, Outcome.ERROR
+    ) is Stage.ERROR_REVIEW
+    assert next_stage(
+        PipelineKey.PLAN_CODE_TEST, Stage.PENDING_MANAGER_REVIEW, Outcome.REJECT
+    ) is Stage.CANCELLED
+
+
+def test_illegal_transitions_are_refused():
+    # Terminal stages never move.
+    for stage in TERMINAL_STAGES:
+        assert TRANSITIONS[stage] == frozenset()
+        assert can_transition_stage(stage, Stage.PENDING_IMPLEMENTATION) is False
+    # A paused task resumes through triage, never straight into code.
+    assert can_transition_stage(Stage.PAUSED, Stage.PENDING_IMPLEMENTATION) is False
+    assert can_transition_stage(Stage.PAUSED, Stage.PENDING_MANAGER_REVIEW) is True
+    # Nobody skips the manager's review by jumping straight to done.
+    assert can_transition_stage(Stage.PENDING_MANAGER_REVIEW, Stage.COMPLETED) is False
+    # Side states cannot jump into each other arbitrarily.
+    assert can_transition_stage(Stage.PENDING_USER_DECISION, Stage.COMPLETED) is False
+
+
+def test_every_pipeline_stage_can_leave_it():
+    """A stage that appears in a pipeline must have a way out of it."""
+    for pipeline, stages in PIPELINES.items():
+        for stage in stages:
+            assert stage is Stage.COMPLETED or TRANSITIONS.get(stage), (pipeline, stage)
+
+
+def test_side_stages_are_not_part_of_any_pipeline():
+    pipeline_stage_set = {stage for stages in PIPELINES.values() for stage in stages}
+    for stage in SIDE_STAGES:
+        assert stage not in pipeline_stage_set
+        assert is_side_stage(stage) is True
+
+
+def test_stage_roles_match_the_pipeline():
+    """Every turn stage has exactly one actor, and it is the right role."""
+    assert STAGE_ROLE[Stage.PENDING_MANAGER_REVIEW] == StageActor(RoleKey.MANAGER, Mode.TRIAGE)
+    assert STAGE_ROLE[Stage.PENDING_ANALYSIS] == StageActor(RoleKey.PLANNER, Mode.WORK)
+    assert STAGE_ROLE[Stage.PENDING_IMPLEMENTATION] == StageActor(RoleKey.DEVELOPER, Mode.WORK)
+    assert STAGE_ROLE[Stage.PENDING_REVIEW] == StageActor(RoleKey.REVIEWER, Mode.WORK)
+    assert STAGE_ROLE[Stage.PENDING_TEST] == StageActor(RoleKey.TESTER, Mode.WORK)
+    assert STAGE_ROLE[Stage.PENDING_FINAL_REVIEW] == StageActor(RoleKey.MANAGER, Mode.FINAL)
+    assert STAGE_ROLE[Stage.ERROR_REVIEW] == StageActor(RoleKey.MANAGER, Mode.ERROR)
+    # The platform merges; nobody acts in merge_request.
+    assert STAGE_ROLE[Stage.MERGE_REQUEST] is None
+    assert STAGE_ROLE[Stage.PAUSED] is None
+
+
+def test_stages_map_to_the_coarse_task_status():
+    assert coarse_status(Stage.COMPLETED) is TaskStatus.DONE
+    assert coarse_status(Stage.PENDING_USER_DECISION) is TaskStatus.WAITING_APPROVAL
+    assert coarse_status(Stage.PAUSED) is TaskStatus.WAITING_APPROVAL
+    assert coarse_status(Stage.FAILED) is TaskStatus.FAILED
+    assert coarse_status(Stage.CANCELLED) is TaskStatus.CANCELLED
+    for stage in (
+        Stage.PENDING_MANAGER_REVIEW,
+        Stage.PENDING_ANALYSIS,
+        Stage.PENDING_IMPLEMENTATION,
+        Stage.PENDING_REVIEW,
+        Stage.PENDING_TEST,
+        Stage.PENDING_FINAL_REVIEW,
+        Stage.MERGE_REQUEST,
+        Stage.ERROR_REVIEW,
+    ):
+        assert STAGE_STATUS[stage] is TaskStatus.RUNNING, stage
+
+
+def test_pipeline_resolution_priority():
+    """manager > kind > app default > plan_code_test."""
+    assert resolve_pipeline(
+        manager_choice=PipelineKey.ANALYSIS_ONLY, task_kind=TaskKind.FEATURE
+    ) is PipelineKey.ANALYSIS_ONLY
+    assert resolve_pipeline(task_kind=TaskKind.BUGFIX) is PipelineKey.CODE_TEST
+    assert resolve_pipeline(app_default=PipelineKey.CODE_TEST) is PipelineKey.CODE_TEST
+    assert resolve_pipeline() is APP_DEFAULT_PIPELINE
+    assert APP_DEFAULT_PIPELINE is PipelineKey.PLAN_CODE_TEST
+
+
+def test_a_review_pipeline_without_a_reviewer_falls_back():
+    """A task waiting forever for a reviewer that does not exist is worse."""
+    assert resolve_pipeline(task_kind=TaskKind.REFACTOR, has_active_reviewer=True) is (
+        PipelineKey.PLAN_CODE_REVIEW_TEST
+    )
+    assert resolve_pipeline(task_kind=TaskKind.REFACTOR, has_active_reviewer=False) is (
+        PipelineKey.PLAN_CODE_TEST
+    )
+    # Even an explicit manager choice falls back: the fix is hiring a
+    # reviewer, not starting a task nobody can act on.
+    assert resolve_pipeline(
+        manager_choice=PipelineKey.PLAN_CODE_REVIEW_TEST, has_active_reviewer=False
+    ) is PipelineKey.PLAN_CODE_TEST
+
+
+def test_task_kinds_carry_a_pipeline_and_a_risk():
+    assert resolve_pipeline(task_kind=TaskKind.ANALYSIS) is PipelineKey.ANALYSIS_ONLY
+    assert risk_for(TaskKind.REFACTOR) is Risk.HIGH
+    assert risk_for(TaskKind.BUGFIX) is Risk.LOW
+
+
+def test_terminal_and_side_state_predicates():
+    assert is_terminal(Stage.COMPLETED) is True
+    assert is_terminal(Stage.PENDING_IMPLEMENTATION) is False
+    assert is_side_stage(Stage.PAUSED) is True
+    assert is_side_stage(Stage.PENDING_IMPLEMENTATION) is False
+
+
+def test_pipeline_state_defaults_are_a_fresh_triage():
+    state = TaskPipelineState(task_id="t1", organization_id="o1")
+    assert state.stage is Stage.PENDING_MANAGER_REVIEW
+    assert state.status is TaskStatus.RUNNING
+    assert state.needs_turn is True
+    assert state.is_blocked is False
+    assert state.actor() == StageActor(RoleKey.MANAGER, Mode.TRIAGE)
+    assert state.max_cycles == 6
+
+
+# ---------------------------------------------------------------------------
+# Database-backed: optimistic advance and dependency parking
+# ---------------------------------------------------------------------------
+
+
+@pytest_asyncio.fixture
+async def _db(session_factory):
+    """Write directly to the test database: user -> org -> task -> state."""
+    async with session_factory() as session:
+        user = UserRow(
+            email=f"engine-{new_id()[:8]}@example.com", password_hash="x"
+        )
+        session.add(user)
+        await session.flush()
+        org = OrganizationRow(
+            name="سازمان موتور", slug=f"eng-{new_id()[:8]}", owner_user_id=user.id
+        )
+        session.add(org)
+        await session.commit()
+
+        async def make_task(title: str) -> str:
+            async with session_factory() as task_session:
+                task = TaskRow(organization_id=org.id, title=title)
+                task_session.add(task)
+                await task_session.flush()
+                await task_session.commit()
+                return task.id
+
+        async def make_run(task_id: str) -> str:
+            async with session_factory() as run_session:
+                run = RunRow(task_id=task_id)
+                run_session.add(run)
+                await run_session.flush()
+                await run_session.commit()
+                return run.id
+
+        async def make_employee(name: str) -> str:
+            async with session_factory() as employee_session:
+                employee = EmployeeRow(
+                    organization_id=org.id, name=name, role_key="developer"
+                )
+                employee_session.add(employee)
+                await employee_session.flush()
+                await employee_session.commit()
+                return employee.id
+
+        yield session_factory, org.id, make_task, make_run, make_employee
+
+
+async def _state_for(
+    session_factory, organization_id: str, task_id: str, **kwargs
+) -> TaskPipelineState:
+    async with session_factory() as session:
+        repo = PipelineStateRepository(session)
+        state = await repo.create(
+            TaskPipelineState(task_id=task_id, organization_id=organization_id, **kwargs)
+        )
+        await session.commit()
+        return state
+
+
+async def test_advance_moves_the_task_and_bumps_the_version(_db):
+    session_factory, org_id, make_task, make_run, make_employee = _db
+    task_id = await make_task("کار یک")
+    state = await _state_for(session_factory, org_id, task_id)
+    assert state.version == 0
+
+    async with session_factory() as session:
+        moved = await advance(
+            PipelineStateRepository(session),
+            task_id=task_id,
+            expected_from=Stage.PENDING_MANAGER_REVIEW,
+            to=Stage.PENDING_ANALYSIS,
+        )
+        await session.commit()
+    assert moved.stage is Stage.PENDING_ANALYSIS
+    assert moved.version == 1
+    assert moved.status is TaskStatus.RUNNING
+
+
+async def test_advance_refuses_an_illegal_move(_db):
+    session_factory, org_id, make_task, make_run, make_employee = _db
+    task_id = await make_task("کار دو")
+    await _state_for(session_factory, org_id, task_id)
+
+    async with session_factory() as session:
+        with pytest.raises(ConflictError):
+            await advance(
+                PipelineStateRepository(session),
+                task_id=task_id,
+                expected_from=Stage.PENDING_MANAGER_REVIEW,
+                to=Stage.COMPLETED,
+            )
+        await session.rollback()
+
+
+async def test_concurrent_advance_has_exactly_one_winner(_db):
+    """The optimistic check: the loser is told, not silently overwritten."""
+    session_factory, org_id, make_task, make_run, make_employee = _db
+    task_id = await make_task("کار سه")
+    await _state_for(session_factory, org_id, task_id)
+
+    async with session_factory() as first, session_factory() as second:
+        winner = await advance(
+            PipelineStateRepository(first),
+            task_id=task_id,
+            expected_from=Stage.PENDING_MANAGER_REVIEW,
+            to=Stage.PENDING_ANALYSIS,
+        )
+        # The second session still believes the task is at manager review.
+        with pytest.raises(ConflictError):
+            await advance(
+                PipelineStateRepository(second),
+                task_id=task_id,
+                expected_from=Stage.PENDING_MANAGER_REVIEW,
+                to=Stage.PENDING_IMPLEMENTATION,
+            )
+        await first.commit()
+        await second.rollback()
+
+    assert winner.stage is Stage.PENDING_ANALYSIS
+
+
+async def test_advance_on_an_unknown_task_is_a_404(_db):
+    session_factory, _org_id, _make_task, _make_run, _make_employee = _db
+    async with session_factory() as session:
+        with pytest.raises(NotFoundError):
+            await advance(
+                PipelineStateRepository(session),
+                task_id="missing",
+                expected_from=Stage.PENDING_MANAGER_REVIEW,
+                to=Stage.PENDING_ANALYSIS,
+            )
+
+
+async def test_a_failed_dependency_parks_its_dependents(_db):
+    """agybot bug M1: a dead dependency must not deadlock its children."""
+    session_factory, org_id, make_task, make_run, make_employee = _db
+    first = await make_task("کار پایه")
+    second = await make_task("کار وابسته")
+    # The dependency dies from error review: that is the only door to FAILED.
+    await _state_for(session_factory, org_id, first, stage=Stage.ERROR_REVIEW)
+    await _state_for(
+        session_factory, org_id, second, stage=Stage.PENDING_ANALYSIS, depends_on=[first]
+    )
+
+    async with session_factory() as session:
+        states = PipelineStateRepository(session)
+        await advance(
+            states,
+            task_id=first,
+            expected_from=Stage.ERROR_REVIEW,
+            to=Stage.FAILED,
+        )
+        parked = await park_dependents(states, task_id=first)
+        await session.commit()
+
+    assert [p.task_id for p in parked] == [second]
+    assert parked[0].stage is Stage.PENDING_USER_DECISION
+    assert parked[0].status is TaskStatus.WAITING_APPROVAL
+
+    # A dependent that already waits on a human is not parked twice.
+    async with session_factory() as session:
+        again = await park_dependents(PipelineStateRepository(session), task_id=first)
+    assert again == []
+
+
+async def test_a_healthy_dependency_leaves_its_dependents_alone(_db):
+    session_factory, org_id, make_task, make_run, make_employee = _db
+    first = await make_task("کار پایهٔ موفق")
+    second = await make_task("کار وابستهٔ دوم")
+    await _state_for(session_factory, org_id, first)
+    await _state_for(
+        session_factory, org_id, second, stage=Stage.PENDING_ANALYSIS, depends_on=[first]
+    )
+    async with session_factory() as session:
+        # Nothing failed, so nothing is parked.
+        parked = await park_dependents(PipelineStateRepository(session), task_id=first)
+    assert parked == []
+
+
+async def test_patch_bumps_the_version_under_the_lock(_db):
+    """Counter updates (cycles, retries, waivers) are optimistic too."""
+    session_factory, org_id, make_task, make_run, make_employee = _db
+    task_id = await make_task("کار چهارم")
+    state = await _state_for(session_factory, org_id, task_id)
+
+    async with session_factory() as session:
+        repo = PipelineStateRepository(session)
+        patched = await repo.patch(
+            task_id=task_id, expected_version=state.version, cycle=2, waiver_reason="دلیل"
+        )
+        await session.commit()
+    assert patched.cycle == 2
+    assert patched.waiver_reason == "دلیل"
+    assert patched.version == state.version + 1
+
+    async with session_factory() as session:
+        repo = PipelineStateRepository(session)
+        with pytest.raises(StaleStageError):
+            await repo.patch(task_id=task_id, expected_version=state.version, cycle=9)
+        await session.rollback()
+
+
+async def test_engine_rows_cascade_from_their_task(_db):
+    """Deleting the task must not leave engine rows behind."""
+    session_factory, org_id, make_task, make_run, make_employee = _db
+    task_id = await make_task("کار حذف‌شونده")
+    run_id = await make_run(task_id)
+    await _state_for(session_factory, org_id, task_id)
+
+    async with session_factory() as session:
+        turn = await EngineTurnRepository(session).create(
+            EngineTurn(
+                task_id=task_id,
+                run_id=run_id,
+                role=RoleKey.DEVELOPER,
+                mode=Mode.WORK,
+                stage=Stage.PENDING_IMPLEMENTATION,
+                reply_excerpt="چیزی",
+            )
+        )
+        await GateResultRepository(session).create(
+            GateResult(task_id=task_id, gate=GateName.TESTS, ok=True)
+        )
+        await session.commit()
+    assert turn.task_id == task_id
+
+    async with session_factory() as session:
+        await session.execute(delete(TaskRow).where(TaskRow.id == task_id))
+        await session.commit()
+
+    async with session_factory() as session:
+        assert await PipelineStateRepository(session).get(task_id) is None
+        assert await EngineTurnRepository(session).list_for_task(task_id) == []
+
+
+async def test_employee_profile_upsert_round_trip(_db):
+    session_factory, _org_id, make_task, _make_run, make_employee = _db
+    employee_id = await make_employee("سارا")
+    task_id = await make_task("کار در دست")
+    async with session_factory() as session:
+        repo = EmployeeProfileRepository(session)
+        await repo.upsert(
+            EmployeeProfile(
+                employee_id=employee_id,
+                ai_model_id="model-1",
+                roles=["developer"],
+                hired_by="manager",
+            )
+        )
+        await session.commit()
+        # A narrow change does not rewrite the whole record...
+        await repo.set_busy(employee_id=employee_id, task_id=task_id)
+        await repo.set_model(employee_id=employee_id, ai_model_id="model-2")
+        await session.commit()
+        profile = await repo.get(employee_id)
+        assert profile is not None
+        assert profile.ai_model_id == "model-2"
+        assert profile.busy_task_id == task_id
+        assert profile.role_list() == ["developer"]
+        assert profile.hired_by == "manager"

apps/api/tests/test_engine_protocol.py

@@ -0,0 +1,413 @@
+"""The tag protocol: parsing, the role/mode matrix and the per-reply limits.
+
+What these tests defend:
+
+* a tag pair with brackets inside its body is parsed as *one* tag with a
+  literal body — a source file full of ``[[`` must survive a round trip
+* an unknown or disallowed tag never executes: it is dropped and reported so
+  the model can correct itself next turn
+* the per-role matrix is exhaustive (every role/mode that acts has a row) and
+  the toolkit is granted to working roles only — the manager never edits code
+* limits are enforced: three tasks, three team changes, three knowledge
+  items, and the extras are dropped rather than executed
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.application.engine.protocol import (
+    ALLOWED_TAGS,
+    MAX_CREATE_TASKS_PER_REPLY,
+    MAX_KNOWLEDGE_PER_TURN,
+    MAX_TEAM_CHANGES_PER_TURN,
+    ProtocolError,
+    TagName,
+    assign_of,
+    edit_blocks_of,
+    knowledge_items,
+    knowledge_of,
+    parse_reply,
+    task_draft_of,
+    task_drafts,
+    team_change_of,
+    team_changes,
+)
+from app.domain.workforce.models import RoleKey
+from app.domain.workforce.pipeline import Mode
+
+MANAGER_TRIAGE = {"role": RoleKey.MANAGER, "mode": Mode.TRIAGE}
+MANAGER_CHAT = {"role": RoleKey.MANAGER, "mode": Mode.CHAT}
+DEVELOPER = {"role": RoleKey.DEVELOPER, "mode": Mode.WORK}
+PLANNER = {"role": RoleKey.PLANNER, "mode": Mode.WORK}
+
+
+# ---------------------------------------------------------------------------
+# The happy path
+# ---------------------------------------------------------------------------
+
+
+def test_a_paired_tag_carries_its_attrs_and_body():
+    reply = parse_reply(
+        'سلام [[KNOWLEDGE kind="rule" key="style"]]همیشه از فارسی ساده استفاده کن.[[/KNOWLEDGE]]',
+        **MANAGER_TRIAGE,
+    )
+    assert reply.errors == ()
+    assert reply.prose == "سلام"
+    (tag,) = reply.of(TagName.KNOWLEDGE)
+    assert tag.attr("kind") == "rule"
+    assert tag.body == "همیشه از فارسی ساده استفاده کن."
+    items = knowledge_items(reply)
+    assert items[0].kind == "rule"
+    assert items[0].content == "همیشه از فارسی ساده استفاده کن."
+
+
+def test_a_self_closing_tag_has_an_empty_body():
+    reply = parse_reply('[[ASSIGN employee="سارا" pipeline="plan_code_test"/]]', **MANAGER_TRIAGE)
+    assert reply.errors == ()
+    (tag,) = reply.tags
+    assert tag.name is TagName.ASSIGN
+    assert tag.body == ""
+    assign = assign_of(tag)
+    assert assign.employee == "سارا"
+    assert assign.pipeline == "plan_code_test"
+
+
+def test_whitespace_and_case_are_tolerated():
+    reply = parse_reply(
+        '[[  assign   employee = "سارا"  /  ]]',
+        **MANAGER_TRIAGE,
+    )
+    # Attribute values are read even with loose spacing...
+    assert reply.errors == ()
+    assert assign_of(reply.first(TagName.ASSIGN)).employee == "سارا"
+
+
+def test_nested_brackets_in_a_body_stay_literal():
+    """A file's content is not re-parsed as tags: the outer tag wins."""
+    body = 'def render():\n    return "[[NOT_A_TAG]] and [[/NOT_A_TAG]]"'
+    reply = parse_reply(
+        f'[[WRITE_FILE path="app/render.ts"]]{body}[[/WRITE_FILE]]',
+        **DEVELOPER,
+    )
+    assert reply.errors == ()
+    (tag,) = reply.tags
+    assert tag.name is TagName.WRITE_FILE
+    assert tag.body == body
+
+
+def test_several_tags_keep_their_order():
+    reply = parse_reply(
+        '[[READ_FILE path="a.py"]][[LIST_DIR path="."]][[RUN cmd="pytest" timeout="60"/]]',
+        **DEVELOPER,
+    )
+    assert reply.errors == ()
+    assert reply.names() == (TagName.READ_FILE, TagName.LIST_DIR, TagName.RUN)
+    assert reply.of(TagName.RUN)[0].attr("cmd") == "pytest"
+    assert reply.of(TagName.RUN)[0].attr("timeout") == "60"
+
+
+# ---------------------------------------------------------------------------
+# Rejections, all of which the model hears about
+# ---------------------------------------------------------------------------
+
+
+def test_unknown_tags_are_ignored_and_reported():
+    reply = parse_reply('[[HACK theme="green"]] payload [[/HACK]]done', **DEVELOPER)
+    assert reply.tags == ()
+    assert reply.errors == ("ERROR: unknown tag [[HACK]] ignored",)
+    # The unknown pair's body does not leak into the prose either.
+    assert reply.prose == "done"
+
+
+def test_a_tag_outside_the_role_matrix_is_refused():
+    """A manager cannot write files, and a developer cannot assign work."""
+    reply = parse_reply('[[WRITE_FILE path="x"]]hello[[/WRITE_FILE]]', **MANAGER_TRIAGE)
+    assert reply.tags == ()
+    assert reply.errors == (
+        "ERROR: tag [[WRITE_FILE]] is not allowed for role=manager mode=triage",
+    )
+
+    reply = parse_reply('[[ASSIGN employee="سارا"/]]', **DEVELOPER)
+    assert reply.tags == ()
+    assert reply.errors == (
+        "ERROR: tag [[ASSIGN]] is not allowed for role=developer mode=work",
+    )
+
+
+def test_a_stray_closing_tag_is_reported_once():
+    reply = parse_reply("[[/WORK_REPORT]]", **DEVELOPER)
+    assert reply.tags == ()
+    assert reply.errors == ("ERROR: stray closing tag [[/WORK_REPORT]] ignored",)
+
+
+def test_an_unclosed_tag_is_ignored_not_half_parsed():
+    reply = parse_reply("[[WORK_REPORT]] کار انجام شد بدون بستن", **DEVELOPER)
+    assert reply.tags == ()
+    assert any("unclosed tag [[WORK_REPORT]]" in e for e in reply.errors)
+
+
+def test_malformed_attributes_are_reported_and_the_tag_keeps_going():
+    reply = parse_reply("[[ASSIGN employee سارا/]]", **MANAGER_TRIAGE)
+    assert any("malformed attribute fragment 'employee سارا'" in e for e in reply.errors)
+    # The tag itself is kept: the typed reader is what refuses it.
+    (tag,) = reply.tags
+    assert tag.name is TagName.ASSIGN
+
+
+def test_an_empty_matrix_cell_refuses_everything():
+    """A role/mode with no row gets no tags at all (deny by default)."""
+    reply = parse_reply("[[KNOWLEDGE]]x[[/KNOWLEDGE]]", role=RoleKey.RESEARCHER, mode=Mode.WORK)
+    assert reply.tags == ()
+    assert "not allowed" in reply.errors[0]
+
+
+# ---------------------------------------------------------------------------
+# The role/mode matrix
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("role", "mode", "expected"),
+    [
+        (RoleKey.MANAGER, Mode.TRIAGE, {"ASSIGN", "REJECT", "ASK_USER", "KNOWLEDGE"}),
+        (RoleKey.MANAGER, Mode.FINAL, {"COMPLETE", "REVISE", "KNOWLEDGE"}),
+        (RoleKey.MANAGER, Mode.ERROR, {"RESOLVE_ERROR", "CANCEL", "KNOWLEDGE"}),
+        (
+            RoleKey.MANAGER,
+            Mode.CHAT,
+            {
+                "CREATE_TASK",
+                "PAUSE_TASK",
+                "RESUME_TASK",
+                "CANCEL_TASK",
+                "HIRE",
+                "UPDATE_EMPLOYEE",
+                "FIRE",
+                "KNOWLEDGE",
+            },
+        ),
+        (
+            RoleKey.PLANNER,
+            Mode.WORK,
+            {"PLAN_SPEC", "SPLIT", "ASK_USER"}
+            | {"READ_FILE", "WRITE_FILE", "EDIT_FILE", "LIST_DIR", "SEARCH", "RUN"},
+        ),
+        (
+            RoleKey.DEVELOPER,
+            Mode.WORK,
+            {"WORK_REPORT", "ASK_USER"}
+            | {"READ_FILE", "WRITE_FILE", "EDIT_FILE", "LIST_DIR", "SEARCH", "RUN"},
+        ),
+        (
+            RoleKey.REVIEWER,
+            Mode.WORK,
+            {"REVIEW_PASS", "REVIEW_CHANGES"}
+            | {"READ_FILE", "WRITE_FILE", "EDIT_FILE", "LIST_DIR", "SEARCH", "RUN"},
+        ),
+        (
+            RoleKey.TESTER,
+            Mode.WORK,
+            {"TEST_PASS", "TEST_FAIL"}
+            | {"READ_FILE", "WRITE_FILE", "EDIT_FILE", "LIST_DIR", "SEARCH", "RUN"},
+        ),
+    ],
+)
+def test_the_matrix_grants_exactly_the_planned_tags(role, mode, expected):
+    granted = {tag.value for tag in ALLOWED_TAGS[(role, mode)]}
+    assert granted == expected
+
+
+def test_the_toolkit_is_available_to_working_roles_only():
+    toolkit = {"READ_FILE", "WRITE_FILE", "EDIT_FILE", "LIST_DIR", "SEARCH", "RUN"}
+    for (role, mode), granted in ALLOWED_TAGS.items():
+        has_tools = bool({t.value for t in granted} & toolkit)
+        if role is RoleKey.MANAGER:
+            assert has_tools is False, (role, mode)
+        else:
+            assert has_tools is True, (role, mode)
+
+
+def test_manager_modes_have_distinct_tag_sets():
+    """The same employee must wear different hats in different modes."""
+    triage = {t.value for t in ALLOWED_TAGS[(RoleKey.MANAGER, Mode.TRIAGE)]}
+    final = {t.value for t in ALLOWED_TAGS[(RoleKey.MANAGER, Mode.FINAL)]}
+    chat = {t.value for t in ALLOWED_TAGS[(RoleKey.MANAGER, Mode.CHAT)]}
+    assert triage != final != chat
+    assert "COMPLETE" in final and "COMPLETE" not in triage
+    assert "CREATE_TASK" in chat and "CREATE_TASK" not in triage
+
+
+# ---------------------------------------------------------------------------
+# Typed readers
+# ---------------------------------------------------------------------------
+
+
+def test_assign_requires_an_employee_or_a_task():
+    tag = parse_reply("[[ASSIGN/]]", **MANAGER_TRIAGE).first(TagName.ASSIGN)
+    with pytest.raises(ProtocolError):
+        assign_of(tag)
+
+
+def test_assign_refuses_an_unknown_pipeline():
+    reply = parse_reply('[[ASSIGN employee="سارا" pipeline="plan_ship_test"/]]', **MANAGER_TRIAGE)
+    with pytest.raises(ProtocolError):
+        assign_of(reply.first(TagName.ASSIGN))
+
+
+def test_create_task_reads_dependencies_and_kind():
+    reply = parse_reply(
+        '[[CREATE_TASK title="نصب فرم ورود" kind="feature" priority="high" depends_on="t1,t2"]]'
+        "توضیحات کار[[/CREATE_TASK]]",
+        **MANAGER_CHAT,
+    )
+    (draft,) = task_drafts(reply)
+    assert draft.title == "نصب فرم ورود"
+    assert draft.kind == "feature"
+    assert draft.priority == "high"
+    assert draft.depends_on == ("t1", "t2")
+    assert draft.description == "توضیحات کار"
+
+
+def test_create_task_without_a_title_is_refused():
+    tag = parse_reply(
+        "[[CREATE_TASK kind=\"feature\"]]x[[/CREATE_TASK]]", **MANAGER_CHAT
+    ).first(TagName.CREATE_TASK)
+    with pytest.raises(ProtocolError):
+        task_draft_of(tag)
+
+
+def test_team_changes_keep_their_order_across_the_three_tags():
+    reply = parse_reply(
+        '[[HIRE role="developer" tier="strong" name="سارا"]]'
+        "[[UPDATE_EMPLOYEE employee_id=\"e1\" tier=\"fast\"]]"
+        '[[FIRE employee_id="e2"]]',
+        **MANAGER_CHAT,
+    )
+    changes = team_changes(reply)
+    assert [c.action for c in changes] == [
+        TagName.HIRE,
+        TagName.UPDATE_EMPLOYEE,
+        TagName.FIRE,
+    ]
+    assert changes[0].name == "سارا"
+    assert changes[0].tier == "strong"
+    # A model name is never part of the protocol: users pick role + tier.
+    for tag in reply.tags:
+        assert not any("model" in key for key in tag.attrs), tag
+
+
+def test_knowledge_defaults_to_a_rule_and_reads_its_key():
+    reply = parse_reply(
+        '[[KNOWLEDGE key="deploy"]]قبل از پوش تست بگیر.[[/KNOWLEDGE]]', **MANAGER_CHAT
+    )
+    (item,) = knowledge_items(reply)
+    assert item.kind == "rule"
+    assert item.key == "deploy"
+    assert item.content == "قبل از پوش تست بگیر."
+
+
+def test_edit_file_body_splits_into_search_replace_blocks():
+    body = (
+        "before\n"
+        "<<<OLD\n"
+        "old line\n"
+        "===\n"
+        "new line\n"
+        "NEW>>>\n"
+        "after\n"
+        "<<<OLD\n"
+        "second old\n"
+        "===\n"
+        "second new\n"
+        "NEW>>>"
+    )
+    reply = parse_reply(f'[[EDIT_FILE path="app/x.py"]]{body}[[/EDIT_FILE]]', **DEVELOPER)
+    (tag,) = reply.of(TagName.EDIT_FILE)
+    blocks = edit_blocks_of(tag)
+    assert [b.old for b in blocks] == ["old line", "second old"]
+    assert [b.new for b in blocks] == ["new line", "second new"]
+
+
+def test_edit_file_without_blocks_is_refused():
+    reply = parse_reply(
+        "[[EDIT_FILE path=\"a\"]]just text[[/EDIT_FILE]]", **DEVELOPER
+    )
+    with pytest.raises(ProtocolError):
+        edit_blocks_of(reply.first(TagName.EDIT_FILE))
+
+
+# ---------------------------------------------------------------------------
+# Limits
+# ---------------------------------------------------------------------------
+
+
+def test_at_most_three_tasks_per_reply():
+    tasks = "".join(
+        f'[[CREATE_TASK title="کار {i}"]]x[[/CREATE_TASK]]' for i in range(3)
+    )
+    reply = parse_reply(tasks, **MANAGER_CHAT)
+    assert len(task_drafts(reply)) == MAX_CREATE_TASKS_PER_REPLY
+    assert reply.errors == ()
+
+
+def test_a_fourth_task_is_dropped_and_reported():
+    tasks = "".join(
+        f'[[CREATE_TASK title="کار {i}"]]x[[/CREATE_TASK]]' for i in range(4)
+    )
+    reply = parse_reply(tasks, **MANAGER_CHAT)
+    assert len(task_drafts(reply)) == 3
+    assert len(reply.errors) == 1
+    assert "too many [[CREATE_TASK]]" in reply.errors[0]
+    assert f"max {MAX_CREATE_TASKS_PER_REPLY}" in reply.errors[0]
+
+
+def test_three_team_changes_per_turn_across_all_three_tags():
+    reply = parse_reply(
+        '[[HIRE role="developer" name="الف"]]'
+        '[[HIRE role="tester" name="ب"]]'
+        '[[HIRE role="reviewer" name="ج"]]'
+        '[[FIRE employee_id="e9"]]',
+        **MANAGER_CHAT,
+    )
+    assert len(team_changes(reply)) == MAX_TEAM_CHANGES_PER_TURN
+    assert any("too many [[FIRE]]" in e for e in reply.errors)
+
+
+def test_three_knowledge_items_per_turn():
+    items = "".join(f'[[KNOWLEDGE key="k{i}"]]متن {i}[[/KNOWLEDGE]]' for i in range(4))
+    reply = parse_reply(items, **MANAGER_CHAT)
+    assert len(knowledge_items(reply)) == MAX_KNOWLEDGE_PER_TURN
+    assert any("too many [[KNOWLEDGE]]" in e for e in reply.errors)
+
+
+def test_limits_do_not_cross_feed_each_other():
+    """Team changes and knowledge are counted separately from tasks."""
+    reply = parse_reply(
+        '[[CREATE_TASK title="a"]]x[[/CREATE_TASK]]'
+        '[[CREATE_TASK title="b"]]x[[/CREATE_TASK]]'
+        '[[HIRE role="developer" name="سارا"]]'
+        '[[KNOWLEDGE key="k"]]متن[[/KNOWLEDGE]]',
+        **MANAGER_CHAT,
+    )
+    assert len(task_drafts(reply)) == 2
+    assert len(team_changes(reply)) == 1
+    assert len(knowledge_items(reply)) == 1
+    assert reply.errors == ()
+
+
+def test_typed_readers_work_on_one_tag_at_a_time():
+    """The runner reads single tags too, not only the whole reply."""
+    reply = parse_reply(
+        '[[HIRE role="developer" tier="balanced" name="سارا"/]]', **MANAGER_CHAT
+    )
+    change = team_change_of(reply.first(TagName.HIRE))
+    assert change.action is TagName.HIRE
+    assert change.role == "developer"
+
+    reply = parse_reply(
+        '[[KNOWLEDGE kind="fact" key="port"]]پورت ۸۰۰۰ است.[[/KNOWLEDGE]]', **MANAGER_CHAT
+    )
+    item = knowledge_of(reply.first(TagName.KNOWLEDGE))
+    assert item.kind == "fact"
+    assert item.content == "پورت ۸۰۰۰ است."

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

@@ -0,0 +1,142 @@
+# Agent 2 — Phase 1 report: engine core (protocol, state machine, pipelines)
+
+**Date:** 2026-10-14 · **Branch:** `dev` (this repo's integration branch) · **Status:** done, all checks green
+
+> Environment note: the TODO file assumes the ext4 worktree (`/home/mamad/work/smart-org-wt/agent2`,
+> branch `agent2/ai-workforce`). This checkout is the `dev` repo itself and already contains the
+> other agents' work (org wizard, invites, delivery infra), so the engine is being implemented here
+> directly on `dev` at the user's instruction. No locked file was touched; no `CHANGE REQUEST` is
+> needed so far.
+
+## What was done
+
+### 1.1 `app/domain/workforce/pipeline.py` (new, pure)
+
+- `Stage` (13 values: the eight pipeline stages plus the side states `pending_user_decision`,
+  `error_review`, `paused`, `cancelled`, `failed`).
+- `PIPELINES`: `plan_code_test` (default), `plan_code_review_test`, `code_test`, `analysis_only`
+  (no worktree, no merge). Each is the ordered tuple of stages ending in `completed`.
+- `TRANSITIONS` — every legal stage move, with the reasoning in comments: a paused or parked task
+  resumes **through triage** (no `paused_from` column exists, and re-triage is the safe route);
+  `merge_request` is platform-executed and nobody acts in it; terminal stages never move.
+- `next_stage(pipeline, current, outcome)` with `Outcome` (`ok`, `revise`, `reject`, `ask_user`,
+  `error`, `resolve`). `ok` walks the pipeline list one step; the others have fixed targets —
+  work sent back always returns to the developer.
+- `STAGE_ROLE` → `(role, mode)` or `None` for stages with no AI turn. Manager modes are
+  `triage` / `final` / `error` (plus `chat`), every other role is `work`.
+- `resolve_pipeline(manager_choice, task_kind, app_default, has_active_reviewer)`: priority
+  manager > kind > app default > `plan_code_test`; a review pipeline without an active reviewer
+  falls back to `plan_code_test`.
+- `TASK_KIND_DEFAULTS`: feature → `plan_code_test`/medium, bugfix → `code_test`/low,
+  refactor → `plan_code_review_test`/high, docs → `code_test`/low, analysis → `analysis_only`/low.
+- `STAGE_STATUS`: stage → coarse `tasks.status` (`new`/`running`/`waiting_approval`/`done`/`failed`/
+  `cancelled`) so existing APIs and the old worker keep working against one truthful column.
+- `TaskPipelineState` dataclass: pipeline, kind, risk, stage, cycle/max_cycles, branch,
+  `last_diff_hash`/`same_diff_count`, `retry_count`, `waiver_reason`, parent/depends, `version`
+  (optimistic lock).
+
+### 1.2 `app/application/engine/protocol.py` (new)
+
+- Parser for `[[TAG attr="v" ...]]body[[/TAG]]` and `[[TAG attr="v"/]]`, tolerant of whitespace,
+  single- or double-quoted values.
+- A body is **literal**: an outer tag wins and the scanner moves past its close, so brackets
+  inside a body (a source file full of `[[`) are never re-parsed.
+- `TagName` enum for all 28 tags; `ALLOWED_TAGS` is the per-role/mode matrix from the TODO
+  (manager triage/final/error/chat, planner, developer, reviewer, tester) plus the six tool tags
+  granted to working roles only — the manager never edits code.
+- Unknown tags, disallowed tags, stray closings, unclosed tags and malformed attribute fragments
+  are dropped and reported as one error line each, in order, for the model to read next turn.
+- `BODY_TAGS`: only tags that carry a body must be closed; a forgotten `/` on a bodyless tag
+  (`[[RUN cmd="pytest"]]`) is read as self-closing instead of failing the reply.
+- Limits: max 3 `CREATE_TASK`, **max 3 team changes across HIRE+UPDATE_EMPLOYEE+FIRE together**,
+  max 3 `KNOWLEDGE` per reply. Extras are dropped and reported.
+- Typed readers: `AssignTag`, `TaskDraft`, `TeamChange`, `KnowledgeItem`, `EditBlock`
+  (`<<<OLD … === … NEW>>>`), plus `edit_blocks_of` which refuses a malformed block rather than
+  half-applying it. `ASSIGN` refuses an unknown pipeline name.
+
+### 1.3 `app/infrastructure/db/models_engine.py` (stub filled) + migration `a2b3c4d5e6f2`
+
+Four tables, all 1:1/side tables — no column was added to the locked `tasks` / `employees`:
+
+- `task_pipeline_state` (PK `task_id`), with the denormalised `organization_id` so the scheduler's
+  hot query stays single-table.
+- `employee_profiles` (PK `employee_id`): `ai_model_id`, `effort`, roles CSV, `hired_by`,
+  `busy_task_id`.
+- `engine_turns`: id, task/run/employee, role, mode, stage, `prompt_chars`, `reply_excerpt`
+  (Text, bounded on write), tags JSON, tool_calls, ok, error_kind, started/finished.
+- `gate_results`: id, task, gate, ok, detail (bounded), created.
+
+Migration `alembic/versions/a2b3c4d5e6f2_ai_workforce.py`:
+`alembic upgrade head` applies cleanly and `alembic check` reports **no drift** between the models
+and the migration chain.
+
+**Deviation (documented in the migration docstring):** the TODO pins
+`down_revision = "a0f0e1d2c3b4"`, which was right while each agent had its own branch. On the
+merged `dev` chain the heads already form a line (`a3b4c5d6e7f4` → `a1b2c3d4e5f1` →
+`d5e6f7a8b9c0`), so this migration chains after `d5e6f7a8b9c0`. Result: **one linear head**
+(`a2b3c4d5e6f2`), no merge revision needed.
+
+### 1.4 `app/application/engine/state.py` (new)
+
+- `advance(states, task_id, expected_from, to)`: checks legality in the domain first, then the
+  repository's compare-and-set (`WHERE task_id AND stage AND version`) performs the move. A lost
+  race raises `ConflictError` ("این کار هم‌اکنون توسط موتور دیگری در حرکت است"), an unknown task
+  raises `NotFoundError`.
+- `park_dependents(states, task_id)`: a failed or cancelled dependency moves every dependent to
+  `pending_user_decision` (agybot bug M1). The function **refuses to act on a healthy task** — the
+  guard is in the function, not in the caller — and leaves dependents that already wait on a human
+  alone. A lost race on a dependent is skipped, not fatal.
+- `app/infrastructure/repositories/engine.py` (new): `PipelineStateRepository` (advance/patch/
+  list_at_stages/list_dependents), `EmployeeProfileRepository`, `EngineTurnRepository`,
+  `GateResultRepository`. `advance`/`patch` verify the version actually moved rather than trusting
+  `rowcount`, so the check holds on SQLite and PostgreSQL alike.
+
+### 1.5 Tests
+
+- `tests/test_engine_pipeline.py` (24 tests): all four pipelines walk manager review → completed
+  step by step, each step also legal; sent-back work returns to the developer; illegal transitions
+  refused; the stage/role/mode map; stage → coarse status; pipeline resolution priority and the
+  reviewer fallback; plus DB-backed tests for the version bump, the illegal move, **exactly one
+  winner in a concurrent advance**, the 404, the dependency-parking rule (and the healthy-task
+  guard), optimistic `patch`, cascade deletes and the employee profile round trip.
+- `tests/test_engine_protocol.py` (34 tests): paired/self-closing parsing, whitespace tolerance,
+  nested brackets staying literal, tag order, unknown/disallowed/stray/unclosed/malformed
+  rejections, the full per-role matrix, toolkit-for-working-roles-only, typed readers
+  (assign/task draft/team changes/knowledge/edit blocks) and all three limits.
+
+## Checks (PLAN-v2 §7.4)
+
+```
+cd apps/api && .venv/bin/python -m pytest -q        → 594 passed, 1 skipped (whole suite; 58 new)
+cd apps/api && .venv/bin/ruff check .               → All checks passed!
+cd apps/api && .venv/bin/mypy app                   → Success: no issues found in 144 source files
+alembic upgrade head (scratch sqlite)               → applies through a2b3c4d5e6f2, single head
+alembic check (scratch sqlite)                      → No new upgrade operations detected.
+```
+
+The web suite was not touched (no UI in Phase 1).
+
+## Deviations & notes for the reviewer
+
+1. Implemented on `dev` instead of an `agent2/ai-workforce` worktree (user instruction; the
+   worktree path in the TODO is a Linux path unavailable here).
+2. Migration `down_revision` chained to the merged head (above) to keep one linear chain.
+3. `task_pipeline_state.organization_id` added beyond the TODO's column list: it is denormalised
+   from the task so the scheduler never joins back, and it keeps the tenant filter single-table.
+4. `resolve_pipeline` applies the no-reviewer fallback to an explicit manager choice too. A review
+   stage nobody can act on is a dead end; the manager who wants a reviewer should hire one.
+5. Team-change limit counts HIRE + UPDATE_EMPLOYEE + FIRE **together** (one budget per turn), which
+   is what "max 3 team changes per turn" means.
+6. New repository module `app/infrastructure/repositories/engine.py` (the engine columns live
+   beside `models_engine.py`); `repositories/workforce.py` was left untouched.
+
+## Open questions / CHANGE REQUESTs
+
+- None for Phase 1. Expected for later phases: org automation policy behind `ModelGrantPort`
+  (TODO 4.1 says to assume enabled until then) and the worker hook for memory expiry.
+
+## Next
+
+Phase 2 — `engine/prompt.py`, `engine/tools.py` (sandbox execution, git refusal, path safety),
+`engine/runner.py` (`run_turn` with the scripted fake model, error classification, usage events)
+and `tests/test_engine_runner.py` + `tests/engine_fakes.py`.