fix(gateway): heartbeat on content-write success, single-claimant gate, progress soft-warn

commit()/progress()/note() now refresh last_heartbeat_at on the success
path (best-effort, suppressed), not only on rejection — an actively
writing agent no longer looks idle to the reaper between verb successes.

commit()/progress() verify the caller holds the active claim
(active_claimant_id), not merely the historical assigned_to, so a reaped
or handed-off assignee can no longer write onto a freed task; a non-holder
gets a not_authorized envelope with a clear remediate.

progress() with no plan_step on a task that has steps is accepted (product
decision for narrative mid-step updates) and logs a soft warning instead of
rejecting.

roboco/services/gateway/content_actions.py

@@ -11,10 +11,13 @@ Pure orchestration; no DB writes outside what the underlying services do.
 
 from __future__ import annotations
 
+import contextlib
 import re
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, ClassVar
 
+import structlog
+
 from roboco.config import settings
 from roboco.exceptions import GitError
 from roboco.foundation.policy import communications as _comms
@@ -27,6 +30,9 @@ if TYPE_CHECKING:
     from uuid import UUID
 
 
+logger = structlog.get_logger()
+
+
 # Scope catalog is canonical in foundation.policy.journaling.
 # Derived here as a string frozenset for the existing call sites that
 # compare strings rather than the Scope enum.
@@ -195,6 +201,27 @@ def _ownership_violation(task_id: UUID) -> Envelope:
     )
 
 
+def _not_active_claimant(task_id: UUID) -> Envelope:
+    """Envelope for a caller who holds no active claim on ``task_id``.
+
+    The caller may still be the historical ``assigned_to`` (e.g. its claim
+    was reaped for going silent, or the task was handed to another agent),
+    but ``active_claimant_id`` no longer points at it. Writing would race the
+    real claimant, so the write is refused.
+    """
+    return Envelope.not_authorized(
+        message=(
+            f"you do not hold the active claim on {task_id}; "
+            "another agent owns it now or your claim was released"
+        ),
+        remediate=(
+            "call i_am_idle() and give_me_work() to pick up fresh work; "
+            "if you believe this is your task, re-claim it before writing"
+        ),
+        context_briefing={},
+    )
+
+
 @dataclass(frozen=True)
 class ContentActionsDeps:
     """Service deps for ContentActions; bundled to keep init signature flat."""
@@ -255,6 +282,38 @@ class ContentActions:
     def evidence_repo(self) -> Any:
         return self._deps.evidence_repo
 
+    async def _touch_heartbeat(self, task_id: UUID | None) -> None:
+        """Best-effort heartbeat refresh on a content-write success path.
+
+        Mirrors the choreographer's rejection-path heartbeat: an agent that
+        is actively committing / posting progress is alive, so refresh
+        ``last_heartbeat_at`` here too — otherwise the reaper sees the claim
+        as stale between verb successes. Wrapped in ``suppress`` so a
+        heartbeat write failure can never alter the response the agent gets.
+        """
+        if task_id is None:
+            return
+        with contextlib.suppress(Exception):
+            await self.task.heartbeat(task_id)
+
+    async def _active_claim_violation(
+        self, agent_id: UUID, task: Any
+    ) -> Envelope | None:
+        """Refuse a content write when the caller is not the active claimant.
+
+        ``assigned_to`` alone is insufficient: a reaped or handed-off agent
+        keeps ``assigned_to`` until reassignment, but ``active_claimant_id``
+        is cleared the moment its claim is released. Only the holder of the
+        active claim may write. A board co-reviewer on a coordination task is
+        exempt (it shares the task with the other board member by design).
+        """
+        claimant = getattr(task, "active_claimant_id", None)
+        if claimant == agent_id:
+            return None
+        if await self._board_may_co_review(agent_id, task):
+            return None
+        return _not_active_claimant(task.id)
+
     async def commit(
         self,
         *,
@@ -301,6 +360,8 @@ class ContentActions:
                 remediate="call give_me_work() first",
                 context_briefing={},
             )
+        if reject := await self._active_claim_violation(agent_id, t):
+            return reject
         canonical_prefix = f"[{str(t.id)[:8]}]"
         final_message = f"{canonical_prefix} {subject}"
         commit_result = await self.git.commit(
@@ -313,6 +374,7 @@ class ContentActions:
         await self.task.add_progress(
             t.id, agent_id, f"committed {sha[:8]}: {final_message}"
         )
+        await self._touch_heartbeat(t.id)
         return Envelope.ok(
             status=str(t.status),
             task_id=str(t.id),
@@ -434,6 +496,7 @@ class ContentActions:
             title=title,
             content=content,
         )
+        await self._touch_heartbeat(task_id)
         return Envelope.ok(
             status="noted",
             task_id=str(task_id) if task_id else None,
@@ -710,6 +773,36 @@ class ContentActions:
         {"cell_pm", "main_pm", "product_owner", "head_marketing", "ceo"}
     )
 
+    _PROGRESS_ACTIVE_STATUSES: ClassVar[frozenset[str]] = frozenset(
+        {"in_progress", "verifying", "awaiting_qa", "awaiting_documentation"}
+    )
+
+    async def _progress_precondition_reject(
+        self, agent_id: UUID, task: Any
+    ) -> Envelope | None:
+        """Ownership + active-claim + active-status gate for progress().
+
+        Returns the rejection envelope, or None when all preconditions hold.
+        Extracted so ``progress`` stays under the return-count bound.
+        """
+        if task.assigned_to != agent_id:
+            return _ownership_violation(task.id)
+        if reject := await self._active_claim_violation(agent_id, task):
+            return reject
+        if str(task.status) not in self._PROGRESS_ACTIVE_STATUSES:
+            return Envelope.invalid_state(
+                message=(
+                    f"task is in {task.status!r}; progress updates only valid "
+                    f"in active statuses ({sorted(self._PROGRESS_ACTIVE_STATUSES)})"
+                ),
+                remediate=(
+                    "use evidence(task_id) to re-read state; if you're past "
+                    "i_am_done, the run has moved on — call i_am_idle()"
+                ),
+                context_briefing={},
+            )
+        return None
+
     async def progress(
         self,
         *,
@@ -728,32 +821,20 @@ class ContentActions:
         mid-step documentation and carries the current derived %.
         ``percentage`` is only a fallback for tasks with no checklist.
 
-        Caller must be the task's assignee and the task must be in an
-        active status — same constraints as the pre-gateway handler.
+        Omitting ``plan_step`` on a task that *has* steps is accepted (a
+        product decision — narrative mid-step updates are valid) but logs a
+        soft warning so the gap is visible. It is never rejected.
+
+        Caller must be the active claimant and the task must be in an
+        active status — same constraints as the pre-gateway handler, plus
+        the single-claimant guard so a reaped/handed-off assignee cannot
+        keep writing.
         """
-        active = {
-            "in_progress",
-            "verifying",
-            "awaiting_qa",
-            "awaiting_documentation",
-        }
         t = await self.task.get(task_id)
         if t is None:
             return Envelope.not_found(message=f"task {task_id} not found")
-        if t.assigned_to != agent_id:
-            return _ownership_violation(task_id)
-        if str(t.status) not in active:
-            return Envelope.invalid_state(
-                message=(
-                    f"task is in {t.status!r}; progress updates only valid "
-                    f"in active statuses ({sorted(active)})"
-                ),
-                remediate=(
-                    "use evidence(task_id) to re-read state; if you're past "
-                    "i_am_done, the run has moved on — call i_am_idle()"
-                ),
-                context_briefing={},
-            )
+        if reject := await self._progress_precondition_reject(agent_id, t):
+            return reject
         result = await self.task.record_plan_progress(
             task_id=task_id,
             agent_id=agent_id,
@@ -773,6 +854,14 @@ class ContentActions:
                 ),
                 context_briefing={},
             )
+        if plan_step is None and result["valid_steps"]:
+            logger.warning(
+                "progress() called without plan_step on a stepped task",
+                task_id=str(task_id),
+                agent_id=str(agent_id),
+                valid_steps=result["valid_steps"],
+            )
+        await self._touch_heartbeat(task_id)
         return Envelope.ok(
             status=str(t.status),
             task_id=str(task_id),

tests/unit/gateway/test_commit_prefix.py

@@ -60,6 +60,7 @@ async def test_commit_prefixes_with_task_id_short() -> None:
     t = MagicMock(
         id=tid,
         assigned_to=aid,
+        active_claimant_id=aid,
         plan="x",
         status="in_progress",
         branch_name="feature/backend/abcd1234",
@@ -95,6 +96,7 @@ async def test_commit_strips_then_re_adds_prefix() -> None:
     t = MagicMock(
         id=tid,
         assigned_to=aid,
+        active_claimant_id=aid,
         plan="x",
         status="in_progress",
         branch_name="feature/backend/abcd1234",
@@ -127,6 +129,7 @@ async def test_commit_prefix_collapses_multiple_spaces() -> None:
     t = MagicMock(
         id=tid,
         assigned_to=aid,
+        active_claimant_id=aid,
         plan="x",
         status="in_progress",
         branch_name="feature/backend/abcd1234",

tests/unit/gateway/test_content_actions.py

@@ -81,7 +81,10 @@ async def test_commit_descriptive_with_active_task_succeeds() -> None:
     agent_id = uuid4()
     task_id = uuid4()
     task_obj = MagicMock(
-        id=task_id, status="in_progress", branch_name="feature/backend/abc"
+        id=task_id,
+        status="in_progress",
+        branch_name="feature/backend/abc",
+        active_claimant_id=agent_id,
     )
     task_svc = AsyncMock()
     task_svc.get_active_task_for_agent.return_value = task_obj
@@ -133,7 +136,10 @@ async def test_commit_strips_existing_task_prefix() -> None:
     task_id = uuid4()
     expected_prefix = f"[{str(task_id)[:8]}]"
     task_obj = MagicMock(
-        id=task_id, status="in_progress", branch_name="feature/backend/abc"
+        id=task_id,
+        status="in_progress",
+        branch_name="feature/backend/abc",
+        active_claimant_id=agent_id,
     )
     task_svc = AsyncMock()
     task_svc.get_active_task_for_agent.return_value = task_obj
@@ -214,7 +220,10 @@ async def test_commit_allows_documenter_role() -> None:
     agent_id = uuid4()
     task_id = uuid4()
     task_obj = MagicMock(
-        id=task_id, status="awaiting_documentation", branch_name="feature/backend/abc"
+        id=task_id,
+        status="awaiting_documentation",
+        branch_name="feature/backend/abc",
+        active_claimant_id=agent_id,
     )
     task_svc = AsyncMock()
     task_svc.agent_for.return_value = MagicMock(role="documenter")
@@ -747,7 +756,12 @@ async def test_reflect_thin_note_records_without_rejection() -> None:
 
 
 def _active_task(agent_id: object) -> MagicMock:
-    return MagicMock(id=uuid4(), assigned_to=agent_id, status="in_progress")
+    return MagicMock(
+        id=uuid4(),
+        assigned_to=agent_id,
+        active_claimant_id=agent_id,
+        status="in_progress",
+    )
 
 
 @pytest.mark.asyncio

tests/unit/gateway/test_content_actions_ownership.py

@@ -78,6 +78,7 @@ async def test_commit_active_task_returned_must_be_caller_owned() -> None:
         id=task_id,
         status="in_progress",
         assigned_to=agent_id,
+        active_claimant_id=agent_id,
         branch_name="feature/backend/abc",
     )
     task_svc = AsyncMock()

tests/unit/gateway/test_content_actions_runkiller.py

@@ -0,0 +1,282 @@
+"""Real-DB tests for content-action run-killers.
+
+Three contracts, all exercised against the live test Postgres so the bug
+boundaries (heartbeat write, claimant gate, plan-step soft-warn) are real:
+
+1. HEARTBEAT-ON-SUCCESS — commit() must refresh ``last_heartbeat_at`` on the
+   success path, not only on rejection. Without it an actively-committing
+   agent looks idle to the reaper between verb successes.
+2. CLAIM-OWNERSHIP — commit()/progress() must verify the caller is the active
+   claimant (``active_claimant_id``), not merely the historical ``assigned_to``.
+   A reaped/stale assignee whose claim was released must not be able to write.
+3. PROGRESS SOFT-WARN — progress() with no ``plan_step`` on a stepped task is
+   accepted (product decision) but emits a warning.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import TYPE_CHECKING
+from unittest.mock import AsyncMock
+from uuid import UUID, uuid4
+
+import pytest
+import structlog
+from roboco.db.tables import AgentTable, ProjectTable, TaskTable
+from roboco.models.base import (
+    AgentRole,
+    AgentStatus,
+    Complexity,
+    TaskNature,
+    TaskStatus,
+    TaskType,
+    Team,
+)
+from roboco.services.gateway.content_actions import ContentActions, ContentActionsDeps
+from roboco.services.task import TaskService
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def _seed_dev_agent(session: AsyncSession, slug_prefix: str) -> UUID:
+    agent = AgentTable(
+        id=uuid4(),
+        name="Backend Dev",
+        slug=f"{slug_prefix}-{uuid4().hex[:8]}",
+        role=AgentRole.DEVELOPER,
+        team=Team.BACKEND,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt="dev",
+        capabilities=["python"],
+        permissions={},
+        metrics={},
+    )
+    session.add(agent)
+    await session.flush()
+    return UUID(str(agent.id))
+
+
+async def _seed_claimed_task(
+    session: AsyncSession,
+    *,
+    assigned_to: UUID,
+    active_claimant_id: UUID | None,
+    plan: dict | None,
+    status: TaskStatus = TaskStatus.IN_PROGRESS,
+) -> UUID:
+    """Seed a project + system creator + an in-progress task and return its id."""
+    system_agent = AgentTable(
+        id=uuid4(),
+        name="System",
+        slug=f"system-{uuid4().hex[:8]}",
+        role=AgentRole.SYSTEM,
+        team=None,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt="system",
+        capabilities=[],
+        permissions={},
+        metrics={},
+    )
+    session.add(system_agent)
+    await session.flush()
+
+    project = ProjectTable(
+        id=uuid4(),
+        name="Run-killer Test Project",
+        slug=f"runkiller-{uuid4().hex[:8]}",
+        git_url="https://github.com/example/runkiller.git",
+        default_branch="main",
+        protected_branches=["main"],
+        assigned_cell=Team.BACKEND,
+        created_by=system_agent.id,
+        is_active=True,
+    )
+    session.add(project)
+    await session.flush()
+
+    task = TaskTable(
+        id=uuid4(),
+        title="Run-killer target task",
+        description="Synthetic task for content-action run-killer tests.",
+        acceptance_criteria=["content actions behave"],
+        status=status,
+        priority=2,
+        task_type=TaskType.CODE,
+        nature=TaskNature.TECHNICAL,
+        project_id=project.id,
+        branch_name="feature/backend/RUNKILL1",
+        created_by=system_agent.id,
+        assigned_to=assigned_to,
+        active_claimant_id=active_claimant_id,
+        team=Team.BACKEND,
+        dependency_ids=[],
+        blocker_ids=[],
+        sequence=0,
+        plan=plan,
+        estimated_complexity=Complexity.LOW,
+        checkpoints=[],
+        progress_updates=[],
+        commits=[],
+        documents=[],
+        last_heartbeat_at=None,
+    )
+    session.add(task)
+    await session.commit()
+    return UUID(str(task.id))
+
+
+def _content_actions(session: AsyncSession) -> ContentActions:
+    """ContentActions backed by a real TaskService; only git is faked.
+
+    The bug boundaries (heartbeat write, claimant gate) live in
+    ContentActions + TaskService against the DB — git is intentionally the
+    one faked dependency since these tests don't exercise real git work.
+    """
+    git = AsyncMock()
+    git.commit.return_value = {"sha": "abc12345def"}
+    return ContentActions(
+        ContentActionsDeps(
+            task=TaskService(session),
+            git=git,
+            messaging=AsyncMock(),
+            a2a=AsyncMock(),
+            journal=AsyncMock(),
+            workspace=AsyncMock(),
+            notifications=AsyncMock(),
+            notification_delivery=AsyncMock(),
+            evidence_repo=AsyncMock(),
+        )
+    )
+
+
+@pytest.mark.asyncio
+async def test_commit_refreshes_heartbeat_on_success(db_session: AsyncSession) -> None:
+    """commit() success must advance last_heartbeat_at past the claim time.
+
+    Regression: the success path returned ok() without touching the
+    heartbeat, so an actively-committing agent looked idle to the reaper.
+    """
+    agent_id = await _seed_dev_agent(db_session, "be-dev")
+    task_id = await _seed_claimed_task(
+        db_session,
+        assigned_to=agent_id,
+        active_claimant_id=agent_id,
+        plan={"steps": ["build"]},
+    )
+    svc = TaskService(db_session)
+    # Stamp an explicitly-old heartbeat so the success-path refresh is
+    # unambiguous (no reliance on sub-millisecond clock resolution).
+    stale = datetime.now(UTC) - timedelta(minutes=5)
+    row = await svc.get(task_id)
+    assert row is not None
+    row.last_heartbeat_at = stale
+    await db_session.commit()
+
+    ca = _content_actions(db_session)
+    env = await ca.commit(
+        agent_id=agent_id,
+        message="feat(api): add /healthz endpoint for liveness checks",
+    )
+    await db_session.commit()
+
+    assert env.as_dict()["error"] is None
+    refreshed = await svc.get(task_id)
+    assert refreshed is not None
+    assert refreshed.last_heartbeat_at is not None
+    assert refreshed.last_heartbeat_at > stale, (
+        "commit() must refresh the claimant heartbeat on success, "
+        "not only on the rejection path"
+    )
+
+
+@pytest.mark.asyncio
+async def test_commit_rejected_when_claim_released(db_session: AsyncSession) -> None:
+    """A stale/reaped assignee whose active claim was cleared cannot commit.
+
+    assigned_to still points at the old agent, but active_claimant_id is
+    NULL (claim released by the reaper). The historical assignee must be
+    refused with not_authorized rather than writing onto a freed task.
+    """
+    agent_id = await _seed_dev_agent(db_session, "be-dev")
+    task_id = await _seed_claimed_task(
+        db_session,
+        assigned_to=agent_id,
+        active_claimant_id=None,
+        plan={"steps": ["build"]},
+    )
+
+    ca = _content_actions(db_session)
+    env = await ca.commit(
+        agent_id=agent_id,
+        message="feat(api): add /healthz endpoint for liveness checks",
+    )
+    body = env.as_dict()
+
+    assert body["error"] == "not_authorized", (
+        "an assignee whose active claim was released must not commit"
+    )
+    _ = task_id
+
+
+@pytest.mark.asyncio
+async def test_commit_rejected_when_another_agent_holds_claim(
+    db_session: AsyncSession,
+) -> None:
+    """Another agent holding the active claim blocks the historical assignee."""
+    old_agent = await _seed_dev_agent(db_session, "be-dev")
+    new_agent = await _seed_dev_agent(db_session, "be-dev")
+    await _seed_claimed_task(
+        db_session,
+        assigned_to=old_agent,
+        active_claimant_id=new_agent,
+        plan={"steps": ["build"]},
+    )
+
+    ca = _content_actions(db_session)
+    env = await ca.commit(
+        agent_id=old_agent,
+        message="feat(api): add /healthz endpoint for liveness checks",
+    )
+    body = env.as_dict()
+
+    assert body["error"] == "not_authorized"
+
+
+@pytest.mark.asyncio
+async def test_progress_no_plan_step_on_stepped_task_is_accepted(
+    db_session: AsyncSession,
+) -> None:
+    """progress() with no plan_step on a stepped task is accepted + warns.
+
+    Product decision: a narrative progress entry without a plan_step on a
+    task that has plan sub_tasks is allowed (not rejected), but emits a
+    soft warning so the gap is visible.
+    """
+    agent_id = await _seed_dev_agent(db_session, "be-dev")
+    task_id = await _seed_claimed_task(
+        db_session,
+        assigned_to=agent_id,
+        active_claimant_id=agent_id,
+        plan={"sub_tasks": [{"id": "s1", "title": "build"}]},
+    )
+
+    ca = _content_actions(db_session)
+    with structlog.testing.capture_logs() as logs:
+        env = await ca.progress(
+            agent_id=agent_id,
+            task_id=task_id,
+            message="made some mid-step progress without finishing a step",
+        )
+    await db_session.commit()
+
+    body = env.as_dict()
+    assert body["error"] is None, "missing plan_step on a stepped task must be accepted"
+    assert body["task_id"] == str(task_id)
+    assert any(
+        entry.get("log_level") == "warning"
+        and "plan_step" in str(entry.get("event", ""))
+        for entry in logs
+    ), "a soft warning must be emitted when plan_step is omitted on a stepped task"