diff --git a/apps/api/alembic/versions/a2b3c4d5e6f2_ai_workforce.py b/apps/api/alembic/versions/a2b3c4d5e6f2_ai_workforce.py new file mode 100644 index 0000000..6bf951e --- /dev/null +++ b/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") diff --git a/apps/api/app/application/engine/__init__.py b/apps/api/app/application/engine/__init__.py new file mode 100644 index 0000000..375f1c5 --- /dev/null +++ b/apps/api/app/application/engine/__init__.py @@ -0,0 +1 @@ +"""The AI-workforce engine: the tag protocol and the worker-facing helpers.""" diff --git a/apps/api/app/application/engine/protocol.py b/apps/api/app/application/engine/protocol.py new file mode 100644 index 0000000..5504547 --- /dev/null +++ b/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 ``<<>>`` search/replace block inside an EDIT_FILE.""" + + old: str + new: str + + +_EDIT_BLOCK_RE = re.compile( + r"<<.*?)\n===\s*\n(?P.*?)\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 ``<<>>``. 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 <<>> 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", +] diff --git a/apps/api/app/application/engine/state.py b/apps/api/app/application/engine/state.py new file mode 100644 index 0000000..6fc579c --- /dev/null +++ b/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 diff --git a/apps/api/app/domain/workforce/engine.py b/apps/api/app/domain/workforce/engine.py new file mode 100644 index 0000000..0e7f333 --- /dev/null +++ b/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) diff --git a/apps/api/app/domain/workforce/pipeline.py b/apps/api/app/domain/workforce/pipeline.py new file mode 100644 index 0000000..7b1e300 --- /dev/null +++ b/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", +] diff --git a/apps/api/app/infrastructure/db/models_engine.py b/apps/api/app/infrastructure/db/models_engine.py index d02adb1..ded500c 100644 --- a/apps/api/app/infrastructure/db/models_engine.py +++ b/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", +] diff --git a/apps/api/app/infrastructure/repositories/engine.py b/apps/api/app/infrastructure/repositories/engine.py new file mode 100644 index 0000000..3dbc508 --- /dev/null +++ b/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 diff --git a/apps/api/tests/test_engine_pipeline.py b/apps/api/tests/test_engine_pipeline.py new file mode 100644 index 0000000..90ba820 --- /dev/null +++ b/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" diff --git a/apps/api/tests/test_engine_protocol.py b/apps/api/tests/test_engine_protocol.py new file mode 100644 index 0000000..93bc9ee --- /dev/null +++ b/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" + "<<>>\n" + "after\n" + "<<>>" + ) + 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 == "پورت ۸۰۰۰ است." diff --git a/docs/plan/reports/agent-2-phase-1.md b/docs/plan/reports/agent-2-phase-1.md new file mode 100644 index 0000000..9def4cb --- /dev/null +++ b/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` + (`<<>>`), 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`.