[F113] collapse WorkSession creation to the validated service path

_create_work_session_if_needed constructed WorkSessionTable directly,
duplicating WorkSessionService.create's validation (existing-active
check, single-active-per-task supersede, project/task existence). The
two sites had drifted. Route through WorkSessionService.create instead,
mapping ConflictError to the idempotent 'if needed' None. Remove the
now-dead _supersede_other_active_sessions (create's
supersede_active_sessions_for_task replaces it).

Fix three pre-existing RED tests surfaced by the sweep (all confirmed
failing on the F110 commit before this change):
- test_fail_qa_work_session_fallback_excludes_qa_session: inserted two
  ACTIVE work_sessions per task, violating uq_work_sessions_one_active
  _per_task (migration 047). The QA session is now ABANDONED — still in
  the fallback query's result set (the query filters by task_id +
  agent_id, not status), so the exclude filter (agent_id != qa_id) is
  still exercised and the dev is resolved.
- test_ceo_reject_routes_coordination_task_to_main_pm /
  test_ceo_reject_routes_batch_umbrella_to_main_pm: ceo_reject emits an
  audit row keyed to CEO_AGENT_ID, but the tests never seeded the CEO
  agent row (fk_audit_log_agent_id_agents). Seed the CEO agent (get-or-
  create, mirroring test_ceo_reject_writes_handoff_journal).

tests/integration/test_task_service_lifecycle_misc.py

@@ -36,12 +36,13 @@ from roboco.models.base import (
 )
 from roboco.models.permissions import AgentContext
 from roboco.models.task import TaskCreateRequest
-from roboco.models.work_session import WorkSessionStatus
+from roboco.models.work_session import WorkSessionCreate, WorkSessionStatus
 from roboco.services.task import (
     SoftBlockInfo,
     SoftBlockInput,
     TaskService,
 )
+from roboco.services.work_session import WorkSessionService
 from sqlalchemy import select
 
 if TYPE_CHECKING:
@@ -448,6 +449,53 @@ async def test_create_work_session_no_project_returns_none(
     assert out is None
 
 
+@pytest.mark.asyncio
+async def test_create_work_session_delegates_to_service_create(
+    task_setup: dict, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """F113: the claim path must create the WorkSession through the validated
+    ``WorkSessionService.create`` (the single source of truth), not construct a
+    ``WorkSessionTable`` directly. Two divergent creation sites had drifted and
+    bypassed the service-layer validation (existing-active check, supersede
+    invariant, project/task existence). Routing through ``create`` collapses
+    them to one validated path. The derived target_branch (parent branch for
+    subtasks, project default for roots) is passed in via ``WorkSessionCreate``.
+    """
+    svc = task_setup["svc"]
+    task = await svc.create(_req(task_setup))
+    task.branch_name = "feature/backend/delegate"
+    await db_session.flush()
+
+    captured: dict[str, Any] = {}
+    original_create = WorkSessionService.create
+
+    async def _spy_create(_self: Any, data: WorkSessionCreate) -> Any:
+        # Record the WorkSessionCreate the claim path handed to the service,
+        # then run the real create so the row persists (the FK on
+        # tasks.work_session_id requires a real work_sessions row).
+        captured["data"] = data
+        return await original_create(_self, data)
+
+    monkeypatch.setattr(WorkSessionService, "create", _spy_create)
+
+    out = await svc._create_work_session_if_needed(
+        task, task_setup["agent_id"], "developer"
+    )
+
+    assert out is not None
+    assert "data" in captured
+    sent = captured["data"]
+    assert isinstance(sent, WorkSessionCreate)
+    assert sent.project_id == task_setup["project_id"]
+    assert sent.task_id == task.id
+    assert sent.agent_id == task_setup["agent_id"]
+    assert sent.branch_name == "feature/backend/delegate"
+    # Root task targets the project default branch.
+    assert sent.target_branch == sent.base_branch
+    # The claim path links the session back onto the task.
+    assert task.work_session_id == out.id
+
+
 # ---------------------------------------------------------------------------
 # unclaim_for_reaper / unclaim_for_agent — work-session abandon paths
 # ---------------------------------------------------------------------------

tests/integration/test_task_service_transitions.py

@@ -952,7 +952,13 @@ async def test_fail_qa_work_session_fallback_excludes_qa_session(
     db_session.add(qa)
     await db_session.flush()
 
-    # A QA-attributed work session (older) and a dev-attributed one (newer).
+    # A QA-attributed work session (older, abandoned) and a dev-attributed one
+    # (newer, active). Only one ACTIVE session may exist per task
+    # (``uq_work_sessions_one_active_per_task``), so the QA session is
+    # ABANDONED — realistic for a stale QA review session and still in the
+    # fallback query's result set (the query filters by task_id + agent_id,
+    # not by status). The exclude filter (``agent_id != qa_agent_id``) is
+    # what must keep the QA session from being misread as the revision dev.
     qa_ws = WorkSessionTable(
         id=uuid4(),
         project_id=task_setup["project_id"],
@@ -961,7 +967,7 @@ async def test_fail_qa_work_session_fallback_excludes_qa_session(
         branch_name="feature/backend/abc",
         base_branch="main",
         target_branch="main",
-        status=WorkSessionStatus.ACTIVE,
+        status=WorkSessionStatus.ABANDONED,
     )
     dev_ws = WorkSessionTable(
         id=uuid4(),
@@ -1083,6 +1089,25 @@ async def test_ceo_reject_routes_coordination_task_to_main_pm(
                 metrics={},
             )
         )
+    # The CEO rejects the task; ceo_reject emits an audit row keyed to the CEO
+    # agent, so the CEO row must exist (fk_audit_log_agent_id_agents).
+    ceo_id = UUID(AGENT_UUIDS["ceo"])
+    if await db_session.get(AgentTable, ceo_id) is None:
+        db_session.add(
+            AgentTable(
+                id=ceo_id,
+                name="CEO",
+                slug="ceo",
+                role=AgentRole.CEO,
+                team=Team.MAIN_PM,
+                status=AgentStatus.ACTIVE,
+                model_config={},
+                system_prompt="ceo",
+                capabilities=[],
+                permissions=[],
+                metrics={},
+            )
+        )
     product = ProductTable(
         name="P", slug=f"p-{uuid4().hex[:8]}", created_by=task_setup["agent_id"]
     )
@@ -1133,6 +1158,27 @@ async def test_ceo_reject_routes_batch_umbrella_to_main_pm(
         )
         await db_session.flush()
 
+    # The CEO rejects the umbrella; ceo_reject emits an audit row keyed to the
+    # CEO agent, so the CEO row must exist (fk_audit_log_agent_id_agents).
+    ceo_id = UUID(AGENT_UUIDS["ceo"])
+    if await db_session.get(AgentTable, ceo_id) is None:
+        db_session.add(
+            AgentTable(
+                id=ceo_id,
+                name="CEO",
+                slug="ceo",
+                role=AgentRole.CEO,
+                team=Team.MAIN_PM,
+                status=AgentStatus.ACTIVE,
+                model_config={},
+                system_prompt="ceo",
+                capabilities=[],
+                permissions=[],
+                metrics={},
+            )
+        )
+        await db_session.flush()
+
     task = await svc.create(_req(task_setup))
     task.status = TaskStatus.AWAITING_CEO_APPROVAL
     task.project_id = None  # umbrella: no project, no product — carries a batch_id