[chore] logical-gaps: route-layer force gate + privileged-field gate + pre-task audit attribution

tasks.py (5 gaps):
- _HATCH_OVERRIDE_STATES expanded to 7: a privileged PATCH INTO a gate
  state (completed/cancelled/awaiting_{qa,documentation,pr_review,
  pm_review,ceo_approval}) now requires explicit force — the panel hatch
  is no longer a quiet click that drops a task into/out of a human gate.
- _RESURRECT_SOURCE_STATES: a privileged PATCH OUT of a terminal status
  (completed/cancelled) resurrects finished work and likewise requires
  force, audited as an override.
- _PRIVILEGED_UPDATE_FIELDS gate: a bare task owner (UPDATE_OWN, no
  ASSIGN) cannot self-reassign / re-team / re-parent / re-depend /
  re-block / rewrite-plan / re-project its task — those structural fields
  are PM-gated; the REST surface must not bypass the verb-layer's
  reassign/delegate/triage gate. A 403 names the touched fields + the
  verb to use instead.
- pre-task create denial: a role that cannot create tasks is now logged
  via log_task_creation_denial (distinct task_creation target_type +
  attempted payload) instead of a 'N/A' task_id that coerced to NULL and
  left the role-escalation attempt unattributable.

audit.py:
- split log_task_action_denial (5-param, under PLR0913) from
  log_task_creation_denial (4-param) — the create path has no task_id;
  the non-UUID sentinel (N/A) is preserved in details[target_id_raw]
  rather than dropped to a NULL target_id indistinguishable from any
  other NULL-target denial.

tests:
- test_tasks_routes.py: parametrized admin-override gate (force
  required for gate + terminal states, force succeeds).
- test_tasks_route_privileged_fields.py: dev owner 403 on
  assigned_to/team/parent_task_id, 200 on dev-facing description.
- test_audit.py: pre-task attribution via log_task_creation_denial +
  non-UUID sentinel preservation.

roboco/api/routes/tasks.py

@@ -81,14 +81,30 @@ _logger = get_logger(__name__)
 
 # #13: lifecycle-bypass hatch states — a privileged PATCH into one of these is a
 # forced override that must carry the explicit ``force`` acknowledgement flag.
+# The set covers every gate / terminal state a panel drag could paste a task
+# into, bypassing the human gate that state represents: COMPLETED (the merge
+# decision), AWAITING_QA / AWAITING_DOCUMENTATION / AWAITING_PR_REVIEW /
+# AWAITING_PM_REVIEW / AWAITING_CEO_APPROVAL (the review/merge/CEO gates),
+# and CANCELLED (the terminal cancel). Without force these are refused so the
+# bypass is always an explicit, audited, acknowledged override — never a quiet
+# panel click that drops a task into (or out of) a gate.
 _HATCH_OVERRIDE_STATES = frozenset(
     {
         TaskStatus.COMPLETED,
+        TaskStatus.CANCELLED,
         TaskStatus.AWAITING_QA,
+        TaskStatus.AWAITING_DOCUMENTATION,
+        TaskStatus.AWAITING_PR_REVIEW,
         TaskStatus.AWAITING_PM_REVIEW,
+        TaskStatus.AWAITING_CEO_APPROVAL,
     }
 )
 
+# Terminal statuses — a privileged PATCH OUT of one of these resurrects
+# finished/cancelled work, which must also carry the explicit ``force``
+# acknowledgement (mirrors the escalate route's refusal to resurrect).
+_RESURRECT_SOURCE_STATES = frozenset({TaskStatus.COMPLETED, TaskStatus.CANCELLED})
+
 
 @dataclass(frozen=True, slots=True)
 class _StatusOverride:
@@ -127,6 +143,19 @@ async def _apply_forced_status_override(req: _StatusOverride) -> TaskTable:
                 '"force": true to acknowledge the forced override.'
             ),
         )
+    # Resurrecting a terminal task (completed / cancelled -> anything) is a
+    # bypass of the merge / cancel decision; it too requires the explicit force
+    # acknowledgement. The target-only hatch gate above misses this because the
+    # target (e.g. in_progress) is not itself a hatch state.
+    if req.task.status in _RESURRECT_SOURCE_STATES and not req.force:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=(
+                f"Task is in the terminal state {req.task.status.value};"
+                " resurrecting it past the lifecycle gate requires"
+                ' "force": true to acknowledge the override.'
+            ),
+        )
     task = await req.service.admin_set_status(
         req.task_id,
         req.new_status,
@@ -154,6 +183,25 @@ _NULLABLE_TASK_FIELDS: frozenset[str] = frozenset(
     {"assigned_to", "parent_task_id", "project_id"}
 )
 
+# Structural / ownership fields a bare task owner (UPDATE_OWN) must NOT
+# self-edit — they reassign the task, move it between teams, re-parent the task
+# tree, rewire the sequencing DAG, re-route it to another repo, or rewrite the
+# delegation plan. These are PM/ASSIGN-gated operations; the verb layer gates
+# them to PM roles (reassign/delegate/triage), so the REST PATCH surface must
+# not let an owner bypass that by setattr-ing them directly. Only a caller with
+# the higher ASSIGN permission may set them.
+_PRIVILEGED_UPDATE_FIELDS: frozenset[str] = frozenset(
+    {
+        "assigned_to",
+        "team",
+        "parent_task_id",
+        "dependency_ids",
+        "blocker_ids",
+        "plan",
+        "project_id",
+    }
+)
+
 
 def _translate_error(e: ServiceError) -> HTTPException:
     """Service errors → HTTP status. Kept at route layer; everything else moves."""
@@ -404,14 +452,24 @@ async def create_task(
     """Create a new task."""
     # Check create permission
     if not permissions.can_perform_task_action(agent, TaskAction.CREATE, data.team):
-        # Log the denial
+        # Log the denial. No task row exists yet, so record the attempted
+        # payload (title/team/type/project) under details with a distinct
+        # target_type — a "N/A" task_id would coerce to NULL and leave the
+        # denial unattributable, exactly where role-escalation attempts surface.
         audit = get_audit_service()
-        await audit.log_task_action_denial(
+        await audit.log_task_creation_denial(
             agent_id=agent.agent_id,
             agent_role=agent.role.value,
-            task_id="N/A",
             action="create",
-            reason="Role not permitted to create tasks",
+            details={
+                "reason": "Role not permitted to create tasks",
+                "attempted_title": getattr(data, "title", None),
+                "attempted_team": getattr(getattr(data, "team", None), "value", None),
+                "attempted_task_type": getattr(
+                    getattr(data, "task_type", None), "value", None
+                ),
+                "attempted_project_id": str(getattr(data, "project_id", None) or ""),
+            },
         )
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,
@@ -967,6 +1025,24 @@ async def update_task(
     # on the ORM object after the update returns.
     null_clears = _pop_null_clears(updates)
 
+    # A bare task owner (UPDATE_OWN) may edit dev-facing fields only. The
+    # structural / ownership fields are gated to ASSIGN/PM; an owner PATCHing
+    # any of them (set or explicitly nulled) without higher perms is refused —
+    # otherwise a dev self-reassigns / re-parents / re-routes their task past
+    # the verb layer's PM gate with no audited override.
+    touched_privileged = (
+        updates.keys() | null_clears.keys()
+    ) & _PRIVILEGED_UPDATE_FIELDS
+    if touched_privileged and not has_higher_perms:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=(
+                "Not authorized to set structural/ownership fields"
+                f" ({sorted(touched_privileged)}); reassign / re-parent /"
+                " re-route requires a PM role."
+            ),
+        )
+
     task = await service.update(task_id, **updates)
     if not task:
         raise HTTPException(

roboco/services/audit.py

@@ -122,8 +122,23 @@ class AuditService(SingletonService):
         ``agents.role`` at write time, not the caller-supplied param.
         Pre-fix the supplied param could disagree with the DB (verb's
         expected role vs caller's actual role); the DB is authoritative.
+
+        A non-UUID ``task_id`` sentinel (e.g. ``"N/A"``) is preserved in
+        ``details["target_id_raw"]`` rather than dropped silently to a NULL
+        target_id indistinguishable from any other NULL-target denial. For a
+        ``create`` denied before any task row exists, prefer
+        :meth:`log_task_creation_denial`, which records the attempted payload
+        under a distinct ``task_creation`` target_type.
         """
         actual_role = await self._resolve_actor_role_from_db(agent_id) or agent_role
+        details: dict[str, Any] = {
+            "agent_role": actual_role,
+            "action": action,
+            "reason": reason,
+        }
+        coerced = _coerce_uuid(task_id)
+        if coerced is None and task_id is not None:
+            details["target_id_raw"] = str(task_id)
         self.log.warning(
             "Task action denied",
             event_type=AuditEventType.TASK_ACTION_DENIED.value,
@@ -139,13 +154,50 @@ class AuditService(SingletonService):
                 event_type=AuditEventType.TASK_ACTION_DENIED.value,
                 agent_id=agent_id,
                 target_type="task",
-                target_id=task_id,
+                target_id=coerced,
                 severity="warning",
-                details={
+                details=details,
+            )
+        )
+
+    async def log_task_creation_denial(
+        self,
+        agent_id: str | UUID,
+        agent_role: str,
+        action: str,
+        details: dict[str, Any] | None = None,
+    ) -> None:
+        """Log a denial of task creation (no task row exists yet).
+
+        A ``create`` denial has no ``task_id``; a ``"N/A"`` placeholder would
+        coerce to a NULL ``target_id`` indistinguishable from any other
+        NULL-target denial, leaving the role-escalation attempt unattributable.
+        The attempted payload is recorded in ``details`` under a distinct
+        ``task_creation`` target_type so the Auditor can see what was tried.
+        """
+        actual_role = await self._resolve_actor_role_from_db(agent_id) or agent_role
+        merged: dict[str, Any] = {
             "agent_role": actual_role,
             "action": action,
-                    "reason": reason,
-                },
+        }
+        if details:
+            merged.update(details)
+        self.log.warning(
+            "Task creation denied",
+            event_type=AuditEventType.TASK_ACTION_DENIED.value,
+            agent_id=str(agent_id),
+            agent_role=actual_role,
+            action=action,
+            timestamp=datetime.now(UTC).isoformat(),
+        )
+        await self._persist(
+            _AuditEvent(
+                event_type=AuditEventType.TASK_ACTION_DENIED.value,
+                agent_id=agent_id,
+                target_type="task_creation",
+                target_id=None,
+                severity="warning",
+                details=merged,
             )
         )
 

tests/integration/test_tasks_route_privileged_fields.py

@@ -0,0 +1,136 @@
+"""Privileged-field gate on PATCH /api/tasks/{id} — the structural/ownership
+fields (assigned_to/team/parent_task_id/...) are PM-gated; a bare task owner
+(UPDATE_OWN) must not self-edit them past the verb layer."""
+
+from __future__ import annotations
+
+from http import HTTPStatus
+from typing import TYPE_CHECKING, cast
+from uuid import UUID, uuid4
+
+import pytest
+import pytest_asyncio
+from fastapi import FastAPI
+from httpx import ASGITransport, AsyncClient
+from roboco.api.deps import get_agent_context, get_db
+from roboco.api.routes.tasks import router as tasks_router
+from roboco.db.tables import AgentTable, ProjectTable, TaskTable
+from roboco.models import AgentRole, AgentStatus, Team
+from roboco.models.base import TaskNature, TaskStatus, TaskType
+from roboco.models.permissions import AgentContext
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+@pytest_asyncio.fixture
+async def dev_client(db_session: AsyncSession) -> AsyncIterator[dict]:
+    """A developer owner (UPDATE_OWN, no ASSIGN) acting on its own task."""
+    dev = AgentTable(
+        id=uuid4(),
+        name="Dev",
+        slug=f"dev-{uuid4().hex[:8]}",
+        role=AgentRole.DEVELOPER,
+        team=Team.BACKEND,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt="dev",
+        capabilities=[],
+        permissions={},
+        metrics={},
+    )
+    db_session.add(dev)
+    await db_session.flush()
+    project = ProjectTable(
+        id=uuid4(),
+        name="PF-Proj",
+        slug=f"pf-proj-{uuid4().hex[:6]}",
+        git_url="https://example.com/pf.git",
+        assigned_cell=Team.BACKEND,
+        created_by=dev.id,
+    )
+    db_session.add(project)
+    await db_session.flush()
+
+    app = FastAPI()
+    app.include_router(tasks_router, prefix="/api/tasks")
+
+    async def _override_db() -> AsyncIterator[AsyncSession]:
+        yield db_session
+
+    async def _override_agent() -> AgentContext:
+        return AgentContext(
+            agent_id=cast("UUID", dev.id), role=AgentRole.DEVELOPER, team=Team.BACKEND
+        )
+
+    app.dependency_overrides[get_db] = _override_db
+    app.dependency_overrides[get_agent_context] = _override_agent
+
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        yield {"client": client, "agent": dev, "project": project, "db": db_session}
+    app.dependency_overrides.clear()
+
+
+def _seed_owned(setup: dict, **kw) -> TaskTable:
+    task = TaskTable(
+        id=uuid4(),
+        title=kw.pop("title", "t"),
+        description=kw.pop("description", "d"),
+        acceptance_criteria=["ac"],
+        status=kw.pop("status", TaskStatus.IN_PROGRESS),
+        priority=2,
+        task_type=TaskType.CODE,
+        nature=TaskNature.TECHNICAL,
+        project_id=setup["project"].id,
+        created_by=setup["agent"].id,
+        assigned_to=setup["agent"].id,
+        team=Team.BACKEND,
+    )
+    setup["db"].add(task)
+    return task
+
+
+_HDR = {"X-Agent-ID": "ignored", "X-Agent-Role": "developer"}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("assigned_to", str(uuid4())),
+        ("team", "frontend"),
+        ("parent_task_id", str(uuid4())),
+    ],
+)
+async def test_owner_cannot_patch_privileged_fields(
+    dev_client: dict, field: str, value: object
+) -> None:
+    """A developer owner (UPDATE_OWN, no ASSIGN) cannot self-reassign / re-team
+    / re-parent their task — those are PM-gated; the REST surface must not
+    bypass the verb layer's reassign/delegate/triage gate."""
+    client = dev_client["client"]
+    task = _seed_owned(dev_client)
+    await dev_client["db"].flush()
+    response = await client.patch(
+        f"/api/tasks/{task.id}",
+        json={field: value},
+        headers=_HDR,
+    )
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+@pytest.mark.asyncio
+async def test_owner_can_patch_dev_facing_field(dev_client: dict) -> None:
+    """A developer owner may still edit dev-facing fields (description)."""
+    client = dev_client["client"]
+    task = _seed_owned(dev_client)
+    await dev_client["db"].flush()
+    response = await client.patch(
+        f"/api/tasks/{task.id}",
+        json={"description": "a long enough updated description for the schema"},
+        headers=_HDR,
+    )
+    assert response.status_code == HTTPStatus.OK

tests/integration/test_tasks_routes.py

@@ -328,6 +328,67 @@ async def test_update_task_status_override_non_hatch_needs_no_force(
     assert response.json()["status"] == "pending"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("hatch", ["awaiting_ceo_approval", "cancelled"])
+async def test_update_task_override_gate_states_require_force(
+    task_client: dict, hatch: str
+) -> None:
+    """The hatch set covers the CEO gate and the terminal cancel too (not just
+    completed/awaiting_qa/awaiting_pm_review): a privileged PATCH into either
+    without ``force`` is refused 400."""
+    client = task_client["client"]
+    task = _seed_task(task_client, status=TaskStatus.IN_PROGRESS)
+    await task_client["db"].flush()
+    response = await client.patch(
+        f"/api/tasks/{task.id}",
+        json={"status": hatch},
+        headers=_HDR,
+    )
+    assert response.status_code == HTTPStatus.BAD_REQUEST, hatch
+    assert "force" in response.json()["detail"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("hatch", ["awaiting_ceo_approval", "cancelled"])
+async def test_update_task_override_gate_states_with_force_succeeds(
+    task_client: dict, hatch: str
+) -> None:
+    client = task_client["client"]
+    task = _seed_task(task_client, status=TaskStatus.IN_PROGRESS)
+    await task_client["db"].flush()
+    response = await client.patch(
+        f"/api/tasks/{task.id}",
+        json={"status": hatch, "force": True},
+        headers=_HDR,
+    )
+    assert response.status_code == HTTPStatus.OK, hatch
+    assert response.json()["status"] == hatch
+
+
+@pytest.mark.asyncio
+async def test_update_task_resurrect_terminal_requires_force(task_client: dict) -> None:
+    """Resurrecting a COMPLETED task back to in_progress is a bypass of the merge
+    decision; the target (in_progress) is not itself a hatch state, so the
+    target-only gate would miss it — the source-terminal check requires force."""
+    client = task_client["client"]
+    task = _seed_task(task_client, status=TaskStatus.COMPLETED)
+    await task_client["db"].flush()
+    no_force = await client.patch(
+        f"/api/tasks/{task.id}",
+        json={"status": "in_progress"},
+        headers=_HDR,
+    )
+    assert no_force.status_code == HTTPStatus.BAD_REQUEST
+    assert "force" in no_force.json()["detail"]
+    with_force = await client.patch(
+        f"/api/tasks/{task.id}",
+        json={"status": "in_progress", "force": True},
+        headers=_HDR,
+    )
+    assert with_force.status_code == HTTPStatus.OK
+    assert with_force.json()["status"] == "in_progress"
+
+
 @pytest.mark.asyncio
 async def test_delete_task(task_client: dict) -> None:
     client = task_client["client"]

tests/unit/services/test_audit.py

@@ -144,3 +144,67 @@ async def test_resolve_actor_role_returns_none_when_db_unavailable(
     # still proceed with the caller-supplied role as fallback.
     result = await svc._resolve_actor_role_from_db(uuid4())
     assert result is None
+
+
+@pytest.mark.asyncio
+async def test_log_task_creation_denial_preserves_pre_task_attribution(
+    svc: AuditService,
+) -> None:
+    """A create denial has no task row yet; the attempted payload must land in
+    details with a distinct ``task_creation`` target_type, not as an anonymous
+    NULL-target row (a "N/A" task_id would coerce to NULL and leave the denial
+    unattributable — the exact hole where role-escalation attempts surface)."""
+    captured: list[_AuditEvent] = []
+    svc._resolve_actor_role_from_db = _AsyncNone()  # type: ignore[method-assign]
+    svc._persist = _Capture(captured)  # type: ignore[method-assign]
+
+    await svc.log_task_creation_denial(
+        agent_id=uuid4(),
+        agent_role="developer",
+        action="create",
+        details={
+            "attempted_title": "Steal the keys",
+            "attempted_team": "backend",
+        },
+    )
+    assert len(captured) == 1
+    event = captured[0]
+    assert event.target_type == "task_creation"
+    assert event.target_id is None
+    assert event.details["attempted_title"] == "Steal the keys"
+    assert event.details["attempted_team"] == "backend"
+    assert event.details["action"] == "create"
+
+
+@pytest.mark.asyncio
+async def test_log_task_action_denial_preserves_non_uuid_target_sentinel(
+    svc: AuditService,
+) -> None:
+    """A non-UUID task_id sentinel is preserved in details rather than dropped
+    silently to a NULL target_id (the dropped-identifier pattern)."""
+    captured: list[_AuditEvent] = []
+    svc._resolve_actor_role_from_db = _AsyncNone()  # type: ignore[method-assign]
+    svc._persist = _Capture(captured)  # type: ignore[method-assign]
+
+    await svc.log_task_action_denial(
+        agent_id=uuid4(),
+        agent_role="developer",
+        task_id="N/A",
+        action="claim",
+        reason="x",
+    )
+    assert captured[0].target_id is None
+    assert captured[0].details["target_id_raw"] == "N/A"
+
+
+class _AsyncNone:
+    async def __call__(self, *_a: object, **_k: object) -> None:
+        return None
+
+
+class _Capture:
+    def __init__(self, sink: list[_AuditEvent]) -> None:
+        self.sink = sink
+
+    async def __call__(self, event: _AuditEvent) -> None:
+        self.sink.append(event)