diff --git a/CHANGELOG.md b/CHANGELOG.md index 8ba06db0..302bf25e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), - **A gateway verb on a vanished task/agent fails cleanly instead of crashing cryptically.** The verb runner's atomic steps dereference `task.id` / `agent.id` with no guard, so a verb invoked when the task or agent could not be resolved (e.g. a task forced into an unexpected state out-of-band) crashed with an opaque `'NoneType' object has no attribute 'id'`. The runner now fails fast with an actionable `INVALID_STATE` error that tells the agent to re-fetch and re-issue its claim verb. +- **An agent could be permanently wedged in a respawn loop by duplicate work sessions on one task.** A task is owned by one agent at a time, so it must have at most one *active* git work session — but nothing enforced that: when a task was re-claimed by a **different** agent (after a pool release, reaper unclaim, or escalation redirect) the prior holder's active session was left open. `WorkSessionService.get_active_for_task` then ran a one-row query across the duplicates and raised `MultipleResultsFound`; the caught failure surfaced as the cryptic `'NoneType' object has no attribute 'id'` that crashed the claim/plan/start flow, so the task could never advance — the orchestrator re-spawned its PM every ~30s forever and the task's dependents stayed blocked. (This was the real root cause behind the verb-runner `INVALID_STATE` guard above, which only made the crash legible.) Fixed at three layers: the active-session lookups now return the most-recent session instead of raising; claiming a task supersedes any other agent's stale active session (the single-active-per-task invariant); and a partial unique index — migration 047, which first de-duplicates existing rows, keeping the most recent — enforces it at the database level so it can never recur. + - **A dev claiming a new task no longer gets stuck on `BRANCH_MISMATCH`.** Each developer has one persistent clone shared across all their tasks, so a finished or abandoned prior task could leave the clone dirty and sitting on a sibling task's branch. The claim's git work (creating/checking out the new task's branch) runs as a side-effect *after* the claim's DB transition commits — so when the checkout failed on that dirty tree, the task was already marked assigned while the workspace stayed on the wrong branch, and the dev's next commit was rejected with `BRANCH_MISMATCH` (stalling, then blocking, the task). The claim now does a `git reset --hard` to clean the tree before the checkouts. It runs only on a fresh claim (resume short-circuits earlier), so the discarded changes are abandoned cruft from a finished task — never committed work, and never the gitignored `.venv`. - **The `note` tool no longer times out under load.** Writing a journal entry / note synchronously waited on RAG indexing, which embeds via Ollama — and Ollama is CPU-bound, so under concurrent load that embed slowed enough to time the `note` gateway tool out entirely (despite a "non-blocking" comment on the code). The entry is already persisted before indexing, so indexing is pure best-effort enrichment: it now runs fire-and-forget on the event loop, and the note/journal write returns immediately. diff --git a/alembic/versions/047_ws_single_active.py b/alembic/versions/047_ws_single_active.py new file mode 100644 index 00000000..252dca44 --- /dev/null +++ b/alembic/versions/047_ws_single_active.py @@ -0,0 +1,71 @@ +"""Enforce one ACTIVE work session per task. + +A task is owned by exactly one agent at a time, so it must carry at most one +ACTIVE ``work_sessions`` row. Nothing enforced this: ``create`` only blocked a +duplicate for the *same* agent, so a re-claim by a different agent (pool +release, reaper unclaim, escalation redirect) left the prior holder's ACTIVE +session open. ``WorkSessionService.get_active_for_task`` then ran +``scalar_one_or_none()`` over the duplicate rows and raised +``MultipleResultsFound``; the caught failure left a ``None`` that the verb flow +dereferenced as ``'NoneType' object has no attribute 'id'`` — wedging +``i_will_plan`` into an infinite PM respawn loop. + +This migration (1) deduplicates existing rows — keeping the most recent ACTIVE +session per task and abandoning the rest — then (2) adds a partial unique index +so the invariant can never be violated again. The service layer now also +supersedes stale sessions on claim; this is the DB backstop. + +Revision ID: 047_ws_single_active +Revises: 046_batch_intake +Create Date: 2026-06-24 + +NOTE: revision id is 20 chars — alembic's ``alembic_version.version_num`` is +``VARCHAR(32)`` and a longer id raises at record time. +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = "047_ws_single_active" +down_revision = "046_batch_intake" +branch_labels = None +depends_on = None + +_INDEX = "uq_work_sessions_one_active_per_task" + + +def upgrade() -> None: + # 1) Deduplicate: for every task with >1 ACTIVE session keep the most recent + # (latest started_at, id as deterministic tie-break) and abandon the rest. + op.execute( + sa.text( + """ + UPDATE work_sessions ws + SET status = 'abandoned', + ended_at = COALESCE(ws.ended_at, now()) + WHERE ws.status = 'active' + AND ws.id <> ( + SELECT keep.id + FROM work_sessions keep + WHERE keep.task_id = ws.task_id + AND keep.status = 'active' + ORDER BY keep.started_at DESC, keep.id DESC + LIMIT 1 + ) + """ + ) + ) + # 2) Enforce the invariant at the DB level: at most one ACTIVE row per task. + op.create_index( + _INDEX, + "work_sessions", + ["task_id"], + unique=True, + postgresql_where=sa.text("status = 'active'"), + ) + + +def downgrade() -> None: + op.drop_index(_INDEX, table_name="work_sessions") diff --git a/roboco/db/tables.py b/roboco/db/tables.py index 43ce53f3..8f6851e1 100644 --- a/roboco/db/tables.py +++ b/roboco/db/tables.py @@ -24,6 +24,7 @@ from sqlalchemy import ( String, Text, UniqueConstraint, + text, ) from sqlalchemy.dialects.postgresql import ARRAY, JSONB, UUID from sqlalchemy.orm import Mapped, mapped_column, relationship @@ -774,6 +775,16 @@ class WorkSessionTable(Base): Index("ix_work_sessions_project_status", "project_id", "status"), Index("ix_work_sessions_task", "task_id"), Index("ix_work_sessions_agent_status", "agent_id", "status"), + # A task is owned by one agent at a time → at most one ACTIVE session. + # Without this, a re-claim by a different agent left duplicate ACTIVE + # rows and get_active_for_task crashed with MultipleResultsFound, + # wedging i_will_plan into a respawn loop. Mirrored by migration 047. + Index( + "uq_work_sessions_one_active_per_task", + "task_id", + unique=True, + postgresql_where=text("status = 'active'"), + ), ) diff --git a/roboco/services/task.py b/roboco/services/task.py index a66c26ed..d96fbd04 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -2147,17 +2147,21 @@ class TaskService(BaseService): ) return None - # Check if session already exists for this task+agent + # Check if session already exists for this task+agent (resilient to any + # legacy duplicate ACTIVE rows — see + # WorkSessionService.get_active_for_task). existing = await self.session.execute( - select(WorkSessionTable).where( + select(WorkSessionTable) + .where( and_( WorkSessionTable.task_id == task.id, WorkSessionTable.agent_id == agent_id, WorkSessionTable.status == WorkSessionStatus.ACTIVE, ) ) + .order_by(WorkSessionTable.started_at.desc()) ) - if existing.scalar_one_or_none(): + if existing.scalars().first(): self.log.debug( "Work session already exists", task_id=str(task.id), @@ -2165,6 +2169,25 @@ class TaskService(BaseService): ) return None + # Single-active-per-task invariant: close any OTHER agent's stale ACTIVE + # session for this task before opening a new one. A re-claim by a + # different agent (pool release, reaper unclaim, escalation redirect) + # otherwise left duplicate ACTIVE rows that crashed get_active_for_task + # with MultipleResultsFound — the i_will_plan respawn-loop wedge. + stale = await self.session.execute( + select(WorkSessionTable).where( + and_( + WorkSessionTable.task_id == task.id, + WorkSessionTable.agent_id != agent_id, + WorkSessionTable.status == WorkSessionStatus.ACTIVE, + ) + ) + ) + for prior in stale.scalars().all(): + prior.status = WorkSessionStatus.ABANDONED + prior.ended_at = datetime.now(UTC) + await self.session.flush() + # Determine target branch: # - For subtasks: merge into parent task's branch # - For parent tasks: merge into default branch (master) diff --git a/roboco/services/work_session.py b/roboco/services/work_session.py index 5c534d5b..1581362b 100644 --- a/roboco/services/work_session.py +++ b/roboco/services/work_session.py @@ -99,6 +99,14 @@ class WorkSessionService(BaseService): resource_type="work_session", ) + # Single-active-per-task invariant: a task has one owner, so close any + # OTHER agent's stale ACTIVE session before opening this one. A re-claim + # by a different agent otherwise left duplicate ACTIVE rows that crashed + # get_active_for_task with MultipleResultsFound (the i_will_plan wedge). + await self.supersede_active_sessions_for_task( + task_id=data.task_id, keep_agent_id=data.agent_id + ) + work_session = WorkSessionTable( project_id=data.project_id, task_id=data.task_id, @@ -185,16 +193,28 @@ class WorkSessionService(BaseService): Returns: Active work session or None + + Resilient to the historical duplicate-ACTIVE-rows defect: a task is + owned by one agent at a time, but a re-claim by a different agent used + to leave the prior holder's ACTIVE session open, so a task could carry + more than one ACTIVE row. ``scalar_one_or_none`` *raised* + ``MultipleResultsFound`` on those rows — the failure surfaced as the + cryptic ``'NoneType' object has no attribute 'id'`` that wedged + ``i_will_plan`` into an infinite respawn loop. Returning the most recent + ACTIVE session instead never raises; the invariant is now also enforced + at creation and by a partial unique index (migration 047). """ result = await self.session.execute( - select(WorkSessionTable).where( + select(WorkSessionTable) + .where( and_( WorkSessionTable.task_id == task_id, WorkSessionTable.status == WorkSessionStatus.ACTIVE, ) ) + .order_by(WorkSessionTable.started_at.desc()) ) - return result.scalar_one_or_none() + return result.scalars().first() async def get_active_for_task_and_agent( self, @@ -212,15 +232,52 @@ class WorkSessionService(BaseService): Active work session or None """ result = await self.session.execute( - select(WorkSessionTable).where( + select(WorkSessionTable) + .where( and_( WorkSessionTable.task_id == task_id, WorkSessionTable.agent_id == agent_id, WorkSessionTable.status == WorkSessionStatus.ACTIVE, ) ) + .order_by(WorkSessionTable.started_at.desc()) ) - return result.scalar_one_or_none() + return result.scalars().first() + + async def supersede_active_sessions_for_task( + self, + task_id: UUID, + keep_agent_id: UUID | None = None, + ) -> int: + """Abandon every ACTIVE work session for a task (single-active invariant). + + A task is owned by exactly one agent at a time, so it must have at most + one ACTIVE work session. A re-claim by a different agent (pool release, + reaper unclaim, escalation redirect) used to leave the prior holder's + ACTIVE session open, accumulating duplicate ACTIVE rows; + ``get_active_for_task`` then hit ``MultipleResultsFound`` and the verb + flow crashed with ``'NoneType' object has no attribute 'id'`` — + wedging the task into an infinite respawn loop. Closing stale sessions + before a new claim keeps the invariant. Pass ``keep_agent_id`` to spare + the incoming claimant's own session. Returns the count superseded. + """ + stmt = select(WorkSessionTable).where( + and_( + WorkSessionTable.task_id == task_id, + WorkSessionTable.status == WorkSessionStatus.ACTIVE, + ) + ) + if keep_agent_id is not None: + stmt = stmt.where(WorkSessionTable.agent_id != keep_agent_id) + result = await self.session.execute(stmt) + superseded = 0 + for ws in result.scalars().all(): + ws.status = WorkSessionStatus.ABANDONED + ws.ended_at = datetime.now(UTC) + superseded += 1 + if superseded: + await self.session.flush() + return superseded async def list_by_agent( self, diff --git a/tests/integration/test_work_session_service.py b/tests/integration/test_work_session_service.py index 82de45e7..e4e71a35 100644 --- a/tests/integration/test_work_session_service.py +++ b/tests/integration/test_work_session_service.py @@ -7,7 +7,7 @@ from uuid import uuid4 import pytest import pytest_asyncio -from roboco.db.tables import AgentTable, ProjectTable, TaskTable +from roboco.db.tables import AgentTable, ProjectTable, TaskTable, WorkSessionTable from roboco.models import AgentRole, AgentStatus, Team from roboco.models.base import ( TaskNature, @@ -21,6 +21,7 @@ from roboco.models.work_session import ( ) from roboco.services.base import ConflictError, NotFoundError, ValidationError from roboco.services.work_session import WorkSessionService +from sqlalchemy.exc import IntegrityError if TYPE_CHECKING: from collections.abc import AsyncIterator @@ -461,3 +462,91 @@ async def test_has_unpushed_commits_false_after_pr(ws_setup: dict) -> None: await svc.add_commit(ws.id, "abc123") await svc.create_pr(ws.id, 1, "u") assert await svc.has_unpushed_commits(ws.id) is False + + +# --------------------------------------------------------------------------- +# Single-active-per-task invariant (migration 047) — the i_will_plan wedge fix +# --------------------------------------------------------------------------- + + +async def _second_agent(db_session: AsyncSession) -> AgentTable: + """A distinct agent to simulate a re-claim by someone else.""" + agent = AgentTable( + id=uuid4(), + name="Dev2", + slug=f"be-dev-{uuid4().hex[:8]}", + role=AgentRole.DEVELOPER, + team=Team.BACKEND, + status=AgentStatus.ACTIVE, + model_config={}, + system_prompt="dev", + capabilities=[], + permissions={}, + metrics={}, + ) + db_session.add(agent) + await db_session.flush() + return agent + + +@pytest.mark.asyncio +async def test_create_supersedes_other_agents_active_session( + ws_setup: dict, db_session: AsyncSession +) -> None: + """A re-claim by a different agent abandons the stale session (no dup ACTIVE).""" + svc = ws_setup["svc"] + first = await svc.create(_payload(ws_setup)) # agent A + agent_b = await _second_agent(db_session) + second = await svc.create( + WorkSessionCreate( + project_id=ws_setup["project_id"], + task_id=ws_setup["task_id"], + agent_id=agent_b.id, + branch_name=f"feature/x-{uuid4().hex[:6]}", + base_branch="main", + target_branch="main", + ) + ) + # The prior holder's session is abandoned; exactly one ACTIVE remains. + refreshed_first = await svc.get(first.id) + assert refreshed_first is not None + assert refreshed_first.status == WorkSessionStatus.ABANDONED + active = await svc.get_active_for_task(ws_setup["task_id"]) + assert active is not None + assert active.id == second.id + + +@pytest.mark.asyncio +async def test_supersede_helper_keeps_named_agent(ws_setup: dict) -> None: + svc = ws_setup["svc"] + keep = await svc.create(_payload(ws_setup)) # agent A + n = await svc.supersede_active_sessions_for_task( + ws_setup["task_id"], keep_agent_id=ws_setup["agent_id"] + ) + assert n == 0 # nothing to supersede — A is the kept agent + refreshed = await svc.get(keep.id) + assert refreshed is not None + assert refreshed.status == WorkSessionStatus.ACTIVE + + +@pytest.mark.asyncio +async def test_partial_unique_index_blocks_two_active_for_task( + ws_setup: dict, db_session: AsyncSession +) -> None: + """The DB backstop: a second ACTIVE row for one task violates the index.""" + svc = ws_setup["svc"] + await svc.create(_payload(ws_setup)) # one ACTIVE session + agent_b = await _second_agent(db_session) + # Bypass the service supersede and force a raw duplicate ACTIVE row. + dup = WorkSessionTable( + project_id=ws_setup["project_id"], + task_id=ws_setup["task_id"], + agent_id=agent_b.id, + branch_name="feature/dup", + base_branch="main", + target_branch="main", + status=WorkSessionStatus.ACTIVE, + ) + db_session.add(dup) + with pytest.raises(IntegrityError): + await db_session.flush()