[chore] work-session-routes: ownership check on mutating routes + stamp merge_pr merged_by from auth (#158 #271)

Every mutating work-session route keyed off session_id alone after the
role gate, so any developer could commit into / abandon / complete a
peer's active session (breaking the single-active-WorkSession invariant
and stranding that task) and any PM could merge any cell's PR — the REST
surface bypassed the verb layer's active-claimant gate entirely. Add a
shared _assert_ownership guard: dev ops require session.agent_id to be
the caller; PM merge_pr requires a cell PM to own the session's task cell
(main PM / CEO / board coordinate every cell), 404 for a missing session.

merge_pr took merged_by from the request body, so any PM could record a
PR merge under another agent's id, corrupting the merge audit trail the
completion/CEO-approval chain and metrics rely on. Drop the body param
and stamp the authenticated caller's agent_id as merged_by (the
MergePRRequest schema is gone with it).

Tests: a second dev's token hitting a peer's /commits and /abandon -> 403
(session left active); a foreign-cell PM -> 403, same-cell PM -> 200; a
spoofed body merged_by is ignored and the persisted row records the PM.

roboco/api/routes/work_session.py

@@ -19,7 +19,6 @@ from roboco.api.schemas.work_session import (
     AddCommitRequest,
     AddFilesRequest,
     CreatePRRequest,
-    MergePRRequest,
     UpdatePRStatusRequest,
     WorkSessionCreateRequest,
     WorkSessionResponse,
@@ -27,12 +26,60 @@ from roboco.api.schemas.work_session import (
     session_to_response,
     session_to_summary,
 )
+from roboco.models import AgentRole
+from roboco.models.permissions import AgentContext
 from roboco.models.work_session import WorkSessionCreate, WorkSessionStatus
-from roboco.services.work_session import get_work_session_service
+from roboco.services.work_session import WorkSessionService, get_work_session_service
 
 router = APIRouter()
 
 
+# =============================================================================
+# OWNERSHIP GUARD
+#
+# Every mutating route keys off session_id alone, so without a re-check any
+# developer could mutate a peer's session and any PM could merge any cell's PR
+# — bypassing the verb layer's active-claimant gate. Re-assert the caller owns
+# the session (dev ops) or owns the session's task cell (PM ops) before the
+# service call (#158).
+# =============================================================================
+
+
+async def _assert_ownership(
+    service: WorkSessionService,
+    session_id: UUID,
+    agent: AgentContext,
+    *,
+    pm_op: bool,
+) -> None:
+    """Fetch the session and verify the caller may mutate it.
+
+    Raises 404 for a missing session, 403 for a wrong-owner / wrong-cell caller.
+    Dev ops require the caller to BE the session's agent. PM ops (merge_pr)
+    require a cell PM to own the session's task cell; main PM / CEO / board
+    coordinate every cell and are admitted by the role gate alone.
+    """
+    session = await service.get(session_id)
+    if not session:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Work session not found: {session_id}",
+        )
+    if pm_op:
+        if agent.role == AgentRole.CELL_PM:
+            team = await service.task_team_for_session(session_id)
+            if agent.team is None or team is None or team != agent.team:
+                raise HTTPException(
+                    status_code=status.HTTP_403_FORBIDDEN,
+                    detail="cell PM does not own this session's task cell",
+                )
+    elif session.agent_id != agent.agent_id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="not the owner of this work session",
+        )
+
+
 # =============================================================================
 # LIST & GET ENDPOINTS
 # =============================================================================
@@ -166,6 +213,7 @@ async def add_commit(
     require_developer_or_above(agent.role, "add commits")
 
     service = get_work_session_service(db)
+    await _assert_ownership(service, session_id, agent, pm_op=False)
 
     session = await service.add_commit(session_id, data.commit_sha)
     await db.commit()
@@ -190,6 +238,7 @@ async def add_files_modified(
     require_developer_or_above(agent.role, "add files")
 
     service = get_work_session_service(db)
+    await _assert_ownership(service, session_id, agent, pm_op=False)
 
     session = await service.add_files_modified(session_id, data.file_paths)
     await db.commit()
@@ -219,6 +268,7 @@ async def create_pr(
     require_developer_or_above(agent.role, "create PRs")
 
     service = get_work_session_service(db)
+    await _assert_ownership(service, session_id, agent, pm_op=False)
 
     session = await service.create_pr(session_id, data.pr_number, data.pr_url)
     await db.commit()
@@ -243,6 +293,7 @@ async def update_pr_status(
     require_developer_or_above(agent.role, "update PR status")
 
     service = get_work_session_service(db)
+    await _assert_ownership(service, session_id, agent, pm_op=False)
 
     session = await service.update_pr_status(session_id, data.pr_status)
     await db.commit()
@@ -259,16 +310,22 @@ async def update_pr_status(
 @router.post("/{session_id}/pr/merge", response_model=WorkSessionResponse)
 async def merge_pr(
     session_id: UUID,
-    data: MergePRRequest,
     db: DbSession,
     agent: CurrentAgentContext,
 ) -> WorkSessionResponse:
-    """Record PR merge and complete the session (PM only)."""
+    """Record PR merge and complete the session (PM only).
+
+    The merger is the AUTHENTICATED caller — never a client-supplied body value,
+    which any PM could spoof to record the merge under another agent's id,
+    corrupting the merge audit trail the completion/CEO-approval chain and
+    metrics rely on (#271).
+    """
     require_pm_or_above(agent.role, "merge PRs")
 
     service = get_work_session_service(db)
+    await _assert_ownership(service, session_id, agent, pm_op=True)
 
-    session = await service.merge_pr(session_id, data.merged_by)
+    session = await service.merge_pr(session_id, agent.agent_id)
     await db.commit()
 
     if not session:
@@ -295,6 +352,7 @@ async def complete_session(
     require_developer_or_above(agent.role, "complete sessions")
 
     service = get_work_session_service(db)
+    await _assert_ownership(service, session_id, agent, pm_op=False)
 
     session = await service.complete(session_id)
     await db.commit()
@@ -319,6 +377,7 @@ async def abandon_session(
     require_developer_or_above(agent.role, "abandon sessions")
 
     service = get_work_session_service(db)
+    await _assert_ownership(service, session_id, agent, pm_op=False)
 
     session = await service.abandon(session_id, reason=reason)
     await db.commit()

roboco/api/schemas/work_session.py

@@ -114,12 +114,6 @@ class UpdatePRStatusRequest(BaseModel):
     pr_status: str  # open, merged, closed
 
 
-class MergePRRequest(BaseModel):
-    """Request to record PR merge."""
-
-    merged_by: UUID
-
-
 # =============================================================================
 # CONVERTERS
 # =============================================================================

roboco/services/work_session.py

@@ -13,6 +13,7 @@ from sqlalchemy import and_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from roboco.db.tables import ProjectTable, TaskTable, WorkSessionTable
+from roboco.models import Team
 from roboco.models.work_session import (
     WorkSessionCreate,
     WorkSessionStatus,
@@ -143,6 +144,20 @@ class WorkSessionService(BaseService):
             raise NotFoundError("WorkSession", str(session_id))
         return work_session
 
+    async def task_team_for_session(self, session_id: UUID) -> Team | None:
+        """Return the team (cell) of the task a session belongs to.
+
+        Used by the route layer's PM cell-ownership check: a cell PM may only
+        merge the PR of a session whose task lives in their own cell. None when
+        the session or its task is missing.
+        """
+        result = await self.session.execute(
+            select(TaskTable.team)
+            .join(WorkSessionTable, WorkSessionTable.task_id == TaskTable.id)
+            .where(WorkSessionTable.id == session_id)
+        )
+        return result.scalar_one_or_none()
+
     async def update(
         self,
         session_id: UUID,

tests/integration/test_work_session_routes.py

@@ -22,6 +22,7 @@ from roboco.models.base import (
 from roboco.models.permissions import AgentContext
 from roboco.models.work_session import WorkSessionStatus
 from roboco.services.base import ValidationError
+from roboco.services.work_session import get_work_session_service
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator
@@ -867,3 +868,312 @@ async def test_create_session_unknown_project_reraises(ws_client: dict) -> None:
             },
             headers=_HDR,
         )
+
+
+# ---------------------------------------------------------------------------
+# #158: mutating routes must check session ownership — a second developer
+# cannot commit into / abandon a peer's session, and a foreign-cell PM cannot
+# merge another cell's PR. #271: merge_pr stamps the AUTHENTICATED caller as
+# merged_by, ignoring any client-supplied body value.
+# ---------------------------------------------------------------------------
+
+
+def _build_app(db_session: AsyncSession, ctx: AgentContext) -> FastAPI:
+    """Build a WS-router app whose agent context is ``ctx`` (for ownership tests
+    that need a caller other than the fixture's default dev)."""
+
+    app = FastAPI()
+    app.include_router(ws_router, prefix="/api/work-sessions")
+
+    async def _override_db() -> AsyncIterator[AsyncSession]:
+        yield db_session
+
+    async def _override_agent() -> AgentContext:
+        return ctx
+
+    app.dependency_overrides[get_db] = _override_db
+    app.dependency_overrides[get_agent_context] = _override_agent
+    return app
+
+
+async def _seed_agent(
+    db_session: AsyncSession,
+    *,
+    role: AgentRole,
+    team: Team | None,
+    name: str,
+) -> AgentTable:
+    agent = AgentTable(
+        id=uuid4(),
+        name=name,
+        slug=f"{name.lower()}-{uuid4().hex[:6]}",
+        role=role,
+        team=team,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt=name,
+        capabilities=[],
+        permissions={},
+        metrics={},
+    )
+    db_session.add(agent)
+    await db_session.flush()
+    return agent
+
+
+@pytest.mark.asyncio
+async def test_add_commit_other_dev_forbidden(
+    ws_client: dict, db_session: AsyncSession
+) -> None:
+    """A second developer cannot add commits to a peer's session (#158)."""
+    ws = _seed_ws(ws_client, branch_name="feature/owner")
+    db_session.add(ws)
+    await db_session.flush()
+
+    other = await _seed_agent(
+        db_session, role=AgentRole.DEVELOPER, team=Team.BACKEND, name="Other"
+    )
+    app = _build_app(
+        db_session,
+        AgentContext(
+            agent_id=cast("UUID", other.id), role=AgentRole.DEVELOPER, team=Team.BACKEND
+        ),
+    )
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.post(
+            f"/api/work-sessions/{ws.id}/commits",
+            json={"commit_sha": "abc123"},
+            headers=_HDR,
+        )
+    app.dependency_overrides.clear()
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+@pytest.mark.asyncio
+async def test_abandon_other_dev_forbidden(
+    ws_client: dict, db_session: AsyncSession
+) -> None:
+    """A second developer cannot abandon a peer's active session (#158)."""
+    ws = _seed_ws(ws_client, branch_name="feature/abandon-owner")
+    db_session.add(ws)
+    await db_session.flush()
+
+    other = await _seed_agent(
+        db_session, role=AgentRole.DEVELOPER, team=Team.BACKEND, name="Other2"
+    )
+    app = _build_app(
+        db_session,
+        AgentContext(
+            agent_id=cast("UUID", other.id), role=AgentRole.DEVELOPER, team=Team.BACKEND
+        ),
+    )
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.post(
+            f"/api/work-sessions/{ws.id}/abandon",
+            params={"reason": "hijack"},
+            headers=_HDR,
+        )
+    app.dependency_overrides.clear()
+    assert response.status_code == HTTPStatus.FORBIDDEN
+    # The session is still active — the peer's session was not abandoned.
+    assert ws.status == WorkSessionStatus.ACTIVE
+
+
+@pytest.mark.asyncio
+async def test_merge_pr_foreign_cell_pm_forbidden(db_session: AsyncSession) -> None:
+    """A cell PM cannot merge the PR of a session whose task is in another
+    cell (#158)."""
+    pm = await _seed_agent(
+        db_session, role=AgentRole.CELL_PM, team=Team.BACKEND, name="BePM"
+    )
+    project = ProjectTable(
+        id=uuid4(),
+        name="FE-Proj",
+        slug=f"fe-proj-{uuid4().hex[:6]}",
+        git_url="https://example.com/r.git",
+        assigned_cell=Team.FRONTEND,
+        created_by=pm.id,
+    )
+    db_session.add(project)
+    await db_session.flush()
+    task = TaskTable(
+        id=uuid4(),
+        title="t",
+        description="d",
+        acceptance_criteria=["ac"],
+        status=TaskStatus.PENDING,
+        priority=2,
+        task_type=TaskType.CODE,
+        nature=TaskNature.TECHNICAL,
+        project_id=project.id,
+        created_by=pm.id,
+        team=Team.FRONTEND,  # frontend task — the backend PM does not own it
+    )
+    db_session.add(task)
+    await db_session.flush()
+    ws = WorkSessionTable(
+        id=uuid4(),
+        project_id=project.id,
+        task_id=task.id,
+        agent_id=pm.id,
+        branch_name="feature/fe-merge",
+        base_branch="main",
+        target_branch="main",
+        status=WorkSessionStatus.ACTIVE,
+        pr_number=11,
+        pr_status="open",
+    )
+    db_session.add(ws)
+    await db_session.flush()
+
+    app = _build_app(
+        db_session,
+        AgentContext(
+            agent_id=cast("UUID", pm.id), role=AgentRole.CELL_PM, team=Team.BACKEND
+        ),
+    )
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.post(
+            f"/api/work-sessions/{ws.id}/pr/merge",
+            headers=_HDR,
+        )
+    app.dependency_overrides.clear()
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+@pytest.mark.asyncio
+async def test_merge_pr_same_cell_pm_succeeds(db_session: AsyncSession) -> None:
+    """A cell PM may merge the PR of a session whose task is in their own cell."""
+    pm = await _seed_agent(
+        db_session, role=AgentRole.CELL_PM, team=Team.BACKEND, name="BePM2"
+    )
+    project = ProjectTable(
+        id=uuid4(),
+        name="BE-Proj",
+        slug=f"be-proj-{uuid4().hex[:6]}",
+        git_url="https://example.com/r.git",
+        assigned_cell=Team.BACKEND,
+        created_by=pm.id,
+    )
+    db_session.add(project)
+    await db_session.flush()
+    task = TaskTable(
+        id=uuid4(),
+        title="t",
+        description="d",
+        acceptance_criteria=["ac"],
+        status=TaskStatus.PENDING,
+        priority=2,
+        task_type=TaskType.CODE,
+        nature=TaskNature.TECHNICAL,
+        project_id=project.id,
+        created_by=pm.id,
+        team=Team.BACKEND,
+    )
+    db_session.add(task)
+    await db_session.flush()
+    ws = WorkSessionTable(
+        id=uuid4(),
+        project_id=project.id,
+        task_id=task.id,
+        agent_id=pm.id,
+        branch_name="feature/be-merge",
+        base_branch="main",
+        target_branch="main",
+        status=WorkSessionStatus.ACTIVE,
+        pr_number=12,
+        pr_status="open",
+    )
+    db_session.add(ws)
+    await db_session.flush()
+
+    app = _build_app(
+        db_session,
+        AgentContext(
+            agent_id=cast("UUID", pm.id), role=AgentRole.CELL_PM, team=Team.BACKEND
+        ),
+    )
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.post(
+            f"/api/work-sessions/{ws.id}/pr/merge",
+            headers=_HDR,
+        )
+    app.dependency_overrides.clear()
+    assert response.status_code == HTTPStatus.OK
+
+
+@pytest.mark.asyncio
+async def test_merge_pr_stamps_authenticated_identity_not_body(
+    db_session: AsyncSession,
+) -> None:
+    """merge_pr records the authenticated caller as merged_by — a spoofed body
+    merged_by is ignored (#271)."""
+    pm = await _seed_agent(
+        db_session, role=AgentRole.MAIN_PM, team=None, name="MainPMMerge"
+    )
+    decoy = await _seed_agent(
+        db_session, role=AgentRole.DEVELOPER, team=Team.BACKEND, name="Decoy"
+    )
+    project = ProjectTable(
+        id=uuid4(),
+        name="Stamp-Proj",
+        slug=f"stamp-proj-{uuid4().hex[:6]}",
+        git_url="https://example.com/r.git",
+        assigned_cell=Team.BACKEND,
+        created_by=pm.id,
+    )
+    db_session.add(project)
+    await db_session.flush()
+    task = TaskTable(
+        id=uuid4(),
+        title="t",
+        description="d",
+        acceptance_criteria=["ac"],
+        status=TaskStatus.PENDING,
+        priority=2,
+        task_type=TaskType.CODE,
+        nature=TaskNature.TECHNICAL,
+        project_id=project.id,
+        created_by=pm.id,
+        team=Team.BACKEND,
+    )
+    db_session.add(task)
+    await db_session.flush()
+    ws = WorkSessionTable(
+        id=uuid4(),
+        project_id=project.id,
+        task_id=task.id,
+        agent_id=decoy.id,
+        branch_name="feature/stamp",
+        base_branch="main",
+        target_branch="main",
+        status=WorkSessionStatus.ACTIVE,
+        pr_number=21,
+        pr_status="open",
+    )
+    db_session.add(ws)
+    await db_session.flush()
+
+    app = _build_app(
+        db_session,
+        AgentContext(agent_id=cast("UUID", pm.id), role=AgentRole.MAIN_PM, team=None),
+    )
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        # Body carries a SPOOFED merged_by (the decoy dev) — must be ignored.
+        response = await client.post(
+            f"/api/work-sessions/{ws.id}/pr/merge",
+            json={"merged_by": str(decoy.id)},
+            headers=_HDR,
+        )
+    app.dependency_overrides.clear()
+    assert response.status_code == HTTPStatus.OK
+    merged = await get_work_session_service(db_session).get(ws.id)
+    assert merged is not None
+    # The recorded merger is the authenticated PM, NOT the spoofed decoy.
+    assert merged.merged_by == pm.id
+    assert merged.merged_by != decoy.id