fix(release): close the 0.19.0 scan findings — sandbox mongo tag, flow-verb timeout walls, video hardening (#329)

- mongo:8-alpine → mongo:8 (tag never existed; a mongo-opted project could spawn no agents) + a Docker Hub tag-existence e2e guard for every sandbox engine
- flow-verb timeouts at both walls: shared SLOW_VERBS policy (i_am_done / submit_up / submit_root / open_pr / i_will_work_on get the 900s server budget); the MCP client now outlasts the server budget (+10s headroom, orchestrator-injected env) so agents receive the middleware's clean 504 envelope instead of dying at the old flat 30s client timeout
- cancellation safety: the quality gate kills+reaps its child on CancelledError; create_pr records the PR via a shield-with-wait-out helper so the write can neither be skipped nor race get_db's rollback
- video engine: renderer sidecar isolated on a render-only network, 2g/2cpu caps, 570s render watchdog with exit-on-hang, 512MB tar decompression cap, CEO notification on terminal render failure, reject under the approve mutex (fail-closed on Redis-down)
- dead python-jose dependency removed (drops ecdsa and its unfixable Minerva advisory PYSEC-2026-1325); panel --font-mono now a real monospace stack

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

tests/e2e_smoke/test_flow_verb_timeout.py

@@ -212,9 +212,11 @@ def test_flow_verb_holds_lock_when_timeout_disarmed(
     _patch_hang_in_set_plan(monkeypatch)
 
     # Prime the per-agent flow_server reload with a no-op verb (i_am_idle
-    # touches no task). The reload resets module globals (_TIMEOUT=30), so a
-    # pre-call patch would be clobbered; after this call the module is
-    # pinned to this agent and the patch below survives.
+    # touches no task). The reload resets module globals (the env-derived
+    # _TIMEOUT), so a pre-call patch would be clobbered; after this call the
+    # module is pinned to this agent and the patch below survives.
+    # i_will_plan is a default-budget verb (not in SLOW_VERBS), so the
+    # client selects _TIMEOUT for it.
     main_pm.flow("i_am_idle")
     monkeypatch.setattr(flow_server, "_TIMEOUT", _MCP_CLIENT_TIMEOUT_SECONDS)
 

tests/e2e_smoke/test_sandbox_image_tags.py

@@ -0,0 +1,48 @@
+"""Regression guard for the 2026-07-08 ``mongo:8-alpine`` ghost-tag bug.
+
+``roboco/models/sandbox.py`` pinned ``_MongoEngine.image = "mongo:8-alpine"``,
+a tag that has never existed on Docker Hub (MongoDB ships no Alpine variant).
+Every unit test mocks the docker CLI, so none of them ever touch a real
+registry and none caught it — the bug only surfaces the moment a real
+``docker run`` pulls the image. This test queries the Docker Hub registry API
+for every ``SANDBOX_ENGINES`` entry's pinned ``image:tag`` and fails if the
+tag does not actually exist, which is the check that would have caught it.
+
+Network-dependent by design; skips cleanly when the registry is unreachable
+rather than failing (mirrors ``test_background_engines.py``'s local-Redis
+reachability skip).
+"""
+
+from __future__ import annotations
+
+import httpx
+import pytest
+from roboco.models.sandbox import SANDBOX_ENGINES
+
+_REGISTRY_URL = (
+    "https://registry.hub.docker.com/v2/repositories/library/{name}/tags/{tag}"
+)
+_TIMEOUT_SECONDS = 10.0
+_HTTP_OK = 200
+
+
+def _split_image(image: str) -> tuple[str, str]:
+    name, _, tag = image.partition(":")
+    return name, tag or "latest"
+
+
+@pytest.mark.parametrize("engine_name", sorted(SANDBOX_ENGINES))
+def test_sandbox_engine_image_tag_exists_on_docker_hub(engine_name: str) -> None:
+    image = SANDBOX_ENGINES[engine_name].image
+    name, tag = _split_image(image)
+    url = _REGISTRY_URL.format(name=name, tag=tag)
+
+    try:
+        resp = httpx.get(url, timeout=_TIMEOUT_SECONDS)
+    except httpx.TransportError:
+        pytest.skip(f"Docker Hub registry unreachable, cannot verify {image!r}")
+
+    assert resp.status_code == _HTTP_OK, (
+        f"{engine_name}: pinned image {image!r} not found on Docker Hub "
+        f"(library/{name}, tag {tag!r}, status {resp.status_code}) — {url}"
+    )

tests/integration/test_video_routes.py

@@ -460,9 +460,10 @@ async def test_reject_cancels_and_records_reason(
     db_session: AsyncSession, ceo_client: AsyncClient
 ) -> None:
     task = await _seed_draft(db_session)
-    resp = await ceo_client.post(
-        f"/api/video/posts/{task.id}/reject", json={"reason": "Not our voice"}
-    )
+    with _LOCKED[0], _LOCKED[1]:
+        resp = await ceo_client.post(
+            f"/api/video/posts/{task.id}/reject", json={"reason": "Not our voice"}
+        )
     assert resp.status_code == HTTPStatus.OK
     assert resp.json()["reject_reason"] == "Not our voice"
     refreshed = await db_session.get(TaskTable, task.id)

tests/unit/api/test_middleware.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from http import HTTPStatus
 from typing import Any
 
@@ -18,6 +19,7 @@ from roboco.api.middleware import (
     get_status_code,
     setup_middleware,
 )
+from roboco.config import settings
 from roboco.exceptions import (
     AuthenticationError,
     InvalidStateError,
@@ -417,3 +419,53 @@ def test_request_validation_handler_log_preserves_non_secret_fields() -> None:
     assert logged_body["title"] == "visible-title"  # non-secret preserved
     assert logged_body["git_token"] == "***REDACTED***"  # secret redacted
     assert "ghp_secret_xyz" not in str(logged_body)
+
+
+# ---------------------------------------------------------------------------
+# FlowVerbTimeoutMiddleware — per-verb budget selection
+# ---------------------------------------------------------------------------
+
+
+def _make_flow_app() -> FastAPI:
+    """Two /api/v1/flow/* routes that each sleep past the fast budget but
+    under the slow one, so the picked timeout is observable by outcome."""
+    app = FastAPI()
+
+    @app.post("/api/v1/flow/developer/give_me_work")
+    async def _normal_verb() -> Any:
+        await asyncio.sleep(0.3)
+        return {"status": "ok"}
+
+    @app.post("/api/v1/flow/developer/i_am_done")
+    async def _slow_verb() -> Any:
+        await asyncio.sleep(0.3)
+        return {"status": "ok"}
+
+    setup_middleware(app)
+    return app
+
+
+def test_flow_verb_timeout_normal_verb_uses_default_budget(
+    monkeypatch: Any,
+) -> None:
+    """A verb outside _SLOW_VERBS keeps the short default budget — a 0.3s
+    handler exceeds a 0.05s budget and comes back as a 504."""
+    monkeypatch.setattr(settings, "flow_verb_timeout_seconds", 0.05)
+    monkeypatch.setattr(settings, "flow_verb_slow_timeout_seconds", 5)
+
+    client = TestClient(_make_flow_app())
+    response = client.post("/api/v1/flow/developer/give_me_work")
+    assert response.status_code == HTTPStatus.GATEWAY_TIMEOUT
+    assert response.json()["error"] == "gateway_timeout"
+
+
+def test_flow_verb_timeout_slow_verb_uses_slow_budget(monkeypatch: Any) -> None:
+    """A _SLOW_VERBS verb gets the longer budget — the same 0.3s handler that
+    times out on the default budget completes fine under the slow one."""
+    monkeypatch.setattr(settings, "flow_verb_timeout_seconds", 0.05)
+    monkeypatch.setattr(settings, "flow_verb_slow_timeout_seconds", 5)
+
+    client = TestClient(_make_flow_app())
+    response = client.post("/api/v1/flow/developer/i_am_done")
+    assert response.status_code == HTTPStatus.OK
+    assert response.json() == {"status": "ok"}

tests/unit/gateway/test_quality_gate.py

@@ -5,9 +5,10 @@ i_am_done, blocking a red submit before it reaches QA. Full tests stay on CI.
 from __future__ import annotations
 
 import asyncio
+import os
 from types import SimpleNamespace
-from typing import TYPE_CHECKING
-from unittest.mock import AsyncMock, MagicMock
+from typing import TYPE_CHECKING, Any
+from unittest.mock import AsyncMock, MagicMock, patch
 
 if TYPE_CHECKING:
     import pathlib
@@ -128,6 +129,40 @@ async def test_run_one_reaps_killed_timeout_process(
     fake_proc.wait.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_run_one_kills_child_on_outer_cancellation(
+    tmp_path: pathlib.Path,
+) -> None:
+    """An outer cancellation (e.g. FlowVerbTimeoutMiddleware's own
+    asyncio.timeout firing around the whole i_am_done submit) throws
+    CancelledError into the wait_for, bypassing the TimeoutError handler
+    above. Without a dedicated handler the child is orphaned and keeps
+    running past the cancelled request; _run_one must kill + reap it and
+    re-raise.
+    """
+    real_create_subprocess_shell = asyncio.create_subprocess_shell
+    spawned: dict[str, asyncio.subprocess.Process] = {}
+
+    async def _capturing_create(*args: Any, **kwargs: Any) -> Any:
+        proc = await real_create_subprocess_shell(*args, **kwargs)
+        spawned["proc"] = proc
+        return proc
+
+    with patch.object(asyncio, "create_subprocess_shell", _capturing_create):
+        task = asyncio.ensure_future(quality_gate._run_one(tmp_path, "sleep 30"))
+        while "proc" not in spawned:
+            await asyncio.sleep(0.01)
+        await asyncio.sleep(0.1)  # let the shell actually exec sleep
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    proc = spawned["proc"]
+    assert proc.returncode is not None, "child was not reaped after cancellation"
+    with pytest.raises(ProcessLookupError):
+        os.kill(proc.pid, 0)
+
+
 # --- GitService command selection -------------------------------------------
 
 

tests/unit/mcp_servers/test_do_server.py

@@ -52,13 +52,18 @@ def test_commit_posts_message_and_files(do_module: Any) -> None:
     fake_response.json.return_value = {"status": "in_progress", "task_id": "x"}
     fake_client.post.return_value = fake_response
 
-    with patch("httpx.Client", return_value=fake_client):
+    with patch("httpx.Client", return_value=fake_client) as client_cls:
         result = do_module.commit("feat(api): add /healthz", files=["foo.py"])
 
     assert result["status"] == "in_progress"
     args, kwargs = fake_client.post.call_args
     assert "/api/v1/do/commit" in args[0]
     assert kwargs["json"] == {"message": "feat(api): add /healthz", "files": ["foo.py"]}
+    # commit stages+commits a large changeset server-side (up to
+    # git_commit_timeout_seconds, default 180s) — the shared _TIMEOUT (30s)
+    # is tuned for fast content-tool calls and would give up first.
+    assert client_cls.call_args.kwargs["timeout"] == do_module._COMMIT_TIMEOUT
+    assert do_module._COMMIT_TIMEOUT > do_module._TIMEOUT
 
 
 def test_note_default_scope_note(do_module: Any) -> None:

tests/unit/mcp_servers/test_flow_server.py

@@ -504,6 +504,80 @@ def test_escalate_up_passes_reason(monkeypatch: pytest.MonkeyPatch) -> None:
     }
 
 
+# ---------------------------------------------------------------------------
+# Client-side timeout selection — must always outlast the matching server
+# wall (FlowVerbTimeoutMiddleware) so the agent sees the clean 504 envelope
+# instead of a raw httpx timeout. See roboco.foundation.policy.flow_timeouts.
+# ---------------------------------------------------------------------------
+
+
+def test_client_timeout_normal_verb_is_default_plus_headroom(
+    flow_module: types.ModuleType,
+) -> None:
+    assert flow_module._TIMEOUT == (
+        flow_module._SERVER_TIMEOUT_SECONDS + flow_module.CLIENT_HEADROOM_SECONDS
+    )
+    assert flow_module._client_timeout_for("give_me_work") == flow_module._TIMEOUT
+
+
+def test_client_timeout_slow_verb_is_slow_budget_plus_headroom(
+    flow_module: types.ModuleType,
+) -> None:
+    assert flow_module._SLOW_TIMEOUT == (
+        flow_module._SERVER_SLOW_TIMEOUT_SECONDS + flow_module.CLIENT_HEADROOM_SECONDS
+    )
+    assert flow_module._client_timeout_for("i_am_done") == flow_module._SLOW_TIMEOUT
+    # Every SLOW_VERBS member routes through the same slow budget.
+    for verb in flow_module.SLOW_VERBS:
+        assert flow_module._client_timeout_for(verb) == flow_module._SLOW_TIMEOUT
+
+
+def test_client_timeout_env_override_respected(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    manifest_path = tmp_path / "tool-manifest.json"
+    manifest_path.write_text(json.dumps(_FULL_MANIFEST))
+    monkeypatch.setenv("ROBOCO_AGENT_ID", "00000000-0000-0000-0000-000000000001")
+    monkeypatch.setenv("ROBOCO_AGENT_ROLE", "developer")
+    monkeypatch.setenv("ROBOCO_ORCHESTRATOR_URL", "http://test-orchestrator:8000")
+    monkeypatch.setenv("ROBOCO_TOOL_MANIFEST_PATH", str(manifest_path))
+    monkeypatch.setenv("ROBOCO_FLOW_VERB_TIMEOUT_SECONDS", "45")
+    monkeypatch.setenv("ROBOCO_FLOW_VERB_SLOW_TIMEOUT_SECONDS", "600")
+
+    import roboco.mcp.flow_server as srv
+
+    importlib.reload(srv)
+    try:
+        assert srv._TIMEOUT == 45 + srv.CLIENT_HEADROOM_SECONDS
+        assert srv._SLOW_TIMEOUT == 600 + srv.CLIENT_HEADROOM_SECONDS
+        assert srv._client_timeout_for("give_me_work") == srv._TIMEOUT
+        assert srv._client_timeout_for("i_am_done") == srv._SLOW_TIMEOUT
+    finally:
+        importlib.reload(srv)  # restore module state for later tests
+
+
+def test_post_opens_httpx_client_with_slow_timeout_for_slow_verb(
+    flow_module: types.ModuleType,
+) -> None:
+    fake_client = _make_fake_client({"status": "awaiting_qa"})
+
+    with patch("httpx.Client", return_value=fake_client) as client_cls:
+        flow_module.i_am_done("task-abc", notes="done")
+
+    assert client_cls.call_args.kwargs["timeout"] == flow_module._SLOW_TIMEOUT
+
+
+def test_post_opens_httpx_client_with_default_timeout_for_normal_verb(
+    flow_module: types.ModuleType,
+) -> None:
+    fake_client = _make_fake_client({"status": "idle"})
+
+    with patch("httpx.Client", return_value=fake_client) as client_cls:
+        flow_module.give_me_work()
+
+    assert client_cls.call_args.kwargs["timeout"] == flow_module._TIMEOUT
+
+
 def test_escalate_to_ceo_passes_reason(monkeypatch: pytest.MonkeyPatch) -> None:
     """Board / Main PM verb forwards to /api/v1/flow/<role>/escalate_to_ceo."""
     srv = _reload_for_role(

tests/unit/runtime/test_sandbox_provisioner.py

@@ -128,7 +128,7 @@ async def test_provision_mongo_engine() -> None:
     assert mongo.user == "sandbox"
     assert mongo.database == "admin"
     run_call = next(c for c in runner.calls if c[0] == "run")
-    assert "mongo:8-alpine" in run_call
+    assert "mongo:8" in run_call
     # MONGO_INITDB_ROOT_PASSWORD env is baked into the run.
     assert any(a.startswith("MONGO_INITDB_ROOT_PASSWORD=") for a in run_call)
     # /data/db tmpfs mount for the engine.

tests/unit/runtime/test_spawn_strict_mcp.py

@@ -14,6 +14,7 @@ from pathlib import Path
 from unittest.mock import patch
 
 import pytest
+from roboco.config import settings
 from roboco.models.runtime import OrchestratorAgentConfig, SpawnGitContext
 from roboco.runtime.orchestrator import AgentOrchestrator
 
@@ -102,3 +103,19 @@ class TestMcpConfigPinsBakedVenv:
                 f"a drifted workspace-clone lock can't trigger a resync stall; "
                 f"got args={spec['args']}"
             )
+
+    @pytest.mark.asyncio
+    async def test_mcp_env_mirrors_flow_verb_timeout_settings(self) -> None:
+        """flow_server.py (a subprocess, can't read Settings) mirrors the two
+        server-side flow-verb timeout budgets via env so its client timeout
+        stays coherent with operator tuning of either setting."""
+        orch = AgentOrchestrator.__new__(AgentOrchestrator)
+        config_path = await orch._generate_mcp_config("be-dev-1")
+        config = json.loads(Path(config_path).read_text())
+        env = config["mcpServers"]["roboco-flow"]["env"]
+        assert env["ROBOCO_FLOW_VERB_TIMEOUT_SECONDS"] == str(
+            settings.flow_verb_timeout_seconds
+        )
+        assert env["ROBOCO_FLOW_VERB_SLOW_TIMEOUT_SECONDS"] == str(
+            settings.flow_verb_slow_timeout_seconds
+        )

tests/unit/runtime/test_video_render_loop.py

@@ -431,7 +431,15 @@ async def test_render_video_task_terminal_after_max_attempts(
     workspace = _fake_workspace()
     orch = _orch()
     p1, p2 = _render_patches(renderer, workspace)
-    with p1, p2:
+    notify_svc = AsyncMock()
+    with (
+        p1,
+        p2,
+        patch(
+            "roboco.services.notification.NotificationService",
+            return_value=notify_svc,
+        ),
+    ):
         await orch._render_video_task(db_session, task)  # tips to terminal
         calls_at_terminal = len(renderer.calls)
         await orch._render_video_task(db_session, task)  # now a no-op
@@ -443,3 +451,47 @@ async def test_render_video_task_terminal_after_max_attempts(
     assert len(renderer.calls) == calls_at_terminal  # not retried after terminal
     posts = await get_task_service(db_session).list_open_video_posts()
     assert posts == []
+    # Exactly one CEO alert — the second (no-op) call must not re-notify.
+    notify_svc.send_ack_notification.assert_awaited_once()
+    notify_kwargs = notify_svc.send_ack_notification.await_args.kwargs
+    assert notify_kwargs["to_agent"] == "ceo"
+    assert task.title in notify_kwargs["body"]
+    assert "render blew up" in notify_kwargs["body"]
+    assert notify_kwargs["task_id"] == task.id
+
+
+@pytest.mark.asyncio
+async def test_render_video_task_notify_failure_does_not_raise(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A broken notification path (e.g. the second DB connection is down)
+    must not surface out of the render loop — best-effort, like the
+    strategy-engine failure notifier."""
+    await _seed(db_session)
+    _enable(monkeypatch)
+    task = await _make_completed_video_task(
+        db_session, occasion="notify-fails", composition_id="Intro"
+    )
+    seeded = markers.get_video_draft(task) or {}
+    markers.set_video_draft(
+        task, {**seeded, "render_attempts": _MAX_VIDEO_RENDER_ATTEMPTS - 1}
+    )
+    await db_session.flush()
+
+    renderer = _FakeRenderer(fail=True)
+    workspace = _fake_workspace()
+    orch = _orch()
+    p1, p2 = _render_patches(renderer, workspace)
+    with (
+        p1,
+        p2,
+        patch(
+            "roboco.services.notification.NotificationService",
+            side_effect=RuntimeError("notification DB unreachable"),
+        ),
+    ):
+        await orch._render_video_task(db_session, task)  # must not raise
+
+    draft = markers.get_video_draft(task)
+    assert draft is not None
+    assert draft["render_status"] == "failed"

tests/unit/services/test_git.py

@@ -7,6 +7,7 @@ mock the network and filesystem boundaries.
 
 from __future__ import annotations
 
+import asyncio
 from pathlib import Path
 from typing import TYPE_CHECKING
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -512,6 +513,122 @@ async def test_create_pr_returns_pr_dict() -> None:
     assert out["is_root_pr"] is True
 
 
+@pytest.mark.asyncio
+async def test_create_pr_records_pr_despite_cancellation_after_post() -> None:
+    """A cancellation landing after the GitHub POST succeeds but before the
+    local record commits must not lose the record: asyncio.shield lets
+    _record_pr_atomically finish, the cancellation still propagates."""
+    project_id = uuid4()
+    fake_task = MagicMock(
+        id=uuid4(),
+        project_id=project_id,
+        assigned_to=uuid4(),
+        title="Add login",
+        description="A short description",
+    )
+    fake_project = MagicMock(slug="roboco")
+    svc = _service()
+    _bind(svc, "_task_for_branch", AsyncMock(return_value=fake_task))
+    _bind(svc, "_workspace_for_branch", AsyncMock(return_value=Path("/tmp/ws")))
+    _bind(svc, "_get_project_token_or_raise", AsyncMock(return_value="tok"))
+    _bind(svc, "_parse_github_remote", MagicMock(return_value=("acme", "repo")))
+    _bind(svc, "_project_default_branch", AsyncMock(return_value="master"))
+
+    recorded = {"done": False}
+
+    async def _slow_record(*_args: object, **_kwargs: object) -> None:
+        await asyncio.sleep(0.05)
+        recorded["done"] = True
+
+    _bind(svc, "_record_pr_atomically", _slow_record)
+
+    fake_resp = MagicMock()
+    fake_resp.is_success = True
+    fake_resp.status_code = 201
+    fake_resp.json.return_value = {
+        "number": _EXPECTED_PR_NUMBER,
+        "html_url": f"https://github.com/acme/repo/pull/{_EXPECTED_PR_NUMBER}",
+    }
+    _bind(svc, "_post_pr", AsyncMock(return_value=fake_resp))
+
+    with _patch_project_service(fake_project):
+        task = asyncio.ensure_future(
+            svc.create_pr("feature/backend/abc12345", parent="master", is_root_pr=True)
+        )
+        await asyncio.sleep(0.01)  # let create_pr reach the shielded await
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    # _await_shielded waits the in-flight record out BEFORE re-raising, so
+    # by the time `await task` raised, the record had already completed.
+    assert recorded["done"] is True, (
+        "shield must let the record finish despite cancellation"
+    )
+
+
+@pytest.mark.asyncio
+async def test_create_pr_cancellation_waits_out_record_before_reraising() -> None:
+    """Ordering guard for _await_shielded: on cancellation the in-flight
+    _record_pr_atomically must run to COMPLETION before CancelledError
+    re-raises to the caller. A bare asyncio.shield detaches the write and
+    re-raises immediately — the write then races get_db's rollback on the
+    same AsyncSession and asyncpg raises InterfaceError ('another operation
+    is in progress'), escaping as a 500 instead of the middleware's 504."""
+    project_id = uuid4()
+    fake_task = MagicMock(
+        id=uuid4(),
+        project_id=project_id,
+        assigned_to=uuid4(),
+        title="Add login",
+        description="A short description",
+    )
+    fake_project = MagicMock(slug="roboco")
+    svc = _service()
+    _bind(svc, "_task_for_branch", AsyncMock(return_value=fake_task))
+    _bind(svc, "_workspace_for_branch", AsyncMock(return_value=Path("/tmp/ws")))
+    _bind(svc, "_get_project_token_or_raise", AsyncMock(return_value="tok"))
+    _bind(svc, "_parse_github_remote", MagicMock(return_value=("acme", "repo")))
+    _bind(svc, "_project_default_branch", AsyncMock(return_value="master"))
+
+    order: list[str] = []
+
+    async def _slow_record(*_args: object, **_kwargs: object) -> None:
+        await asyncio.sleep(0.05)
+        order.append("record_done")
+
+    _bind(svc, "_record_pr_atomically", _slow_record)
+
+    fake_resp = MagicMock()
+    fake_resp.is_success = True
+    fake_resp.status_code = 201
+    fake_resp.json.return_value = {
+        "number": _EXPECTED_PR_NUMBER,
+        "html_url": f"https://github.com/acme/repo/pull/{_EXPECTED_PR_NUMBER}",
+    }
+    _bind(svc, "_post_pr", AsyncMock(return_value=fake_resp))
+
+    with _patch_project_service(fake_project):
+        task = asyncio.ensure_future(
+            svc.create_pr("feature/backend/abc12345", parent="master", is_root_pr=True)
+        )
+        await asyncio.sleep(0.01)  # mid-record: cancellation lands in the shield
+        task.cancel()
+        # pytest.raises re-raises any OTHER exception type (e.g. the
+        # InterfaceError a racing rollback would surface), failing the test —
+        # that is assertion (a): CancelledError and nothing else propagates.
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        # Appended synchronously right after propagation: no other coroutine
+        # can run in between, so this marker coming SECOND proves the record
+        # coroutine had already completed before the cancellation re-raised.
+        order.append("cancel_raised")
+
+    assert order == ["record_done", "cancel_raised"], (
+        f"record must complete BEFORE the cancellation re-raises; got {order}"
+    )
+
+
 @pytest.mark.asyncio
 async def test_create_pr_raises_when_branch_not_found() -> None:
     svc = _service()

tests/unit/services/test_video_post_service.py

@@ -500,14 +500,68 @@ async def test_approve_unknown_task_returns_none(db_session: AsyncSession) -> No
 @pytest.mark.asyncio
 async def test_reject_records_reason_and_cancels(db_session: AsyncSession) -> None:
     task = await _seed_video_post(db_session)
-    updated = await _svc(
-        db_session, x_poster=_StubXPoster(), tiktok_poster=_StubTikTokPoster()
-    ).reject(_id(task), "Doesn't match the release")
+    with _LOCKED[0], _LOCKED[1]:
+        updated = await _svc(
+            db_session, x_poster=_StubXPoster(), tiktok_poster=_StubTikTokPoster()
+        ).reject(_id(task), "Doesn't match the release")
     assert updated is not None
     assert updated.status == TS.CANCELLED
     assert markers.get_video_reject_reason(updated) == "Doesn't match the release"
 
 
+@pytest.mark.asyncio
+async def test_reject_takes_the_same_lock_approve_holds(
+    db_session: AsyncSession,
+) -> None:
+    """A reject under the real acquire/release path still lands (mirrors the
+    approve happy-path locking) — proves the mutex round-trip, not just the
+    mutation."""
+    task = await _seed_video_post(db_session)
+    with _LOCKED[0], _LOCKED[1]:
+        updated = await _svc(
+            db_session, x_poster=_StubXPoster(), tiktok_poster=_StubTikTokPoster()
+        ).reject(_id(task), "Doesn't match the release")
+    assert updated is not None
+    assert updated.status == TS.CANCELLED
+    _LOCKED[0].new.assert_awaited()
+    _LOCKED[1].new.assert_awaited()
+
+
+@pytest.mark.asyncio
+async def test_reject_concurrent_lock_held_refuses(db_session: AsyncSession) -> None:
+    """A reject arriving while a concurrent approve holds the lock must not
+    cancel a draft that may be mid-post — same refusal as approve's own
+    already-in-progress case."""
+    task = await _seed_video_post(db_session)
+    with patch.object(HeartbeatMutex, "acquire", AsyncMock(return_value=None)):
+        result = await _svc(
+            db_session, x_poster=_StubXPoster(), tiktok_poster=_StubTikTokPoster()
+        ).reject(_id(task), "Doesn't match the release")
+    assert result is None
+    await db_session.refresh(task)
+    assert task.status == TS.PENDING  # never cancelled while the lock was held
+
+
+@pytest.mark.asyncio
+async def test_reject_redis_unavailable_refuses(db_session: AsyncSession) -> None:
+    """Reject fails CLOSED when Redis is unreachable, mirroring approve: an
+    approve that took the lock while Redis was up stays authoritative through
+    the heartbeat grace window after Redis drops, so an unlocked reject could
+    CANCEL a draft that approve is mid-posting. The CEO retries once Redis is
+    back."""
+    task = await _seed_video_post(db_session)
+    broken = MagicMock()
+    broken.set = AsyncMock(side_effect=ConnectionError("redis down"))
+    broken.aclose = AsyncMock()
+    with patch("roboco.services.heartbeat_mutex.redis.from_url", return_value=broken):
+        result = await _svc(
+            db_session, x_poster=_StubXPoster(), tiktok_poster=_StubTikTokPoster()
+        ).reject(_id(task), "Doesn't match the release")
+    assert result is None
+    await db_session.refresh(task)
+    assert task.status == TS.PENDING  # never cancelled without the mutex
+
+
 @pytest.mark.asyncio
 async def test_list_held_video_posts_excludes_terminal(
     db_session: AsyncSession,
@@ -515,7 +569,8 @@ async def test_list_held_video_posts_excludes_terminal(
     open_task = await _seed_video_post(db_session)
     rejected_task = await _seed_video_post(db_session)
     svc = _svc(db_session, x_poster=_StubXPoster(), tiktok_poster=_StubTikTokPoster())
-    await svc.reject(_id(rejected_task), "not relevant")
+    with _LOCKED[0], _LOCKED[1]:
+        await svc.reject(_id(rejected_task), "not relevant")
     held = await svc.list_held_video_posts()
     ids = {t.id for t in held}
     assert open_task.id in ids