u78649634_default/smart-org
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(engine): workforce engine core — tag protocol, pipelines, optimistic state

Browse Source 
Phase 1 of TODO-agent-2: pure logic plus the tables it writes to.

- domain/workforce/pipeline.py: Stage/PipelineKey/Outcome, the four pipelines,
  TRANSITIONS, next_stage, STAGE_ROLE (role+mode), resolve_pipeline with the
  no-reviewer fallback, task-kind defaults, stage -> coarse tasks.status, and
  the TaskPipelineState record (with the optimistic version).
- application/engine/protocol.py: the [[TAG attr="v"]]body[[/TAG]] parser and
  its self-closing form, the per-role/mode allow-list, literal bodies (a
  bracket inside a body is not a tag), typed readers (ASSIGN/CREATE_TASK/
  team changes/KNOWLEDGE/EDIT_FILE blocks) and the per-reply limits
  (3 tasks, 3 team changes, 3 knowledge items).
- application/engine/state.py: advance() with the optimistic compare-and-set
  and park_dependents() so a failed dependency never deadlocks its children.
- db/models_engine.py + migration a2b3c4d5e6f2: task_pipeline_state,
  employee_profiles, engine_turns, gate_results as side tables (tasks and
  employees stay locked). Linear head; alembic check reports no drift.
- repositories/engine.py + 58 new tests (test_engine_protocol,
  test_engine_pipeline) covering every pipeline's happy path, illegal
  transitions, exactly-one-winner concurrent advance, dependency parking and
  all parser edge cases.

Suite: 594 passed, ruff clean, mypy clean.
This commit is contained in:
mjalmoudy94
2026-10-14 15:35:31 +03:30
parent c93b25741e
commit df7da8c75c
11 changed files with 3211 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/tests/test_engine_pipeline.py
+523
View File
@@ -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"