diff --git a/.roboco/conventions.yml b/.roboco/conventions.yml
index 171509aa..f684c7fa 100644
--- a/.roboco/conventions.yml
+++ b/.roboco/conventions.yml
@@ -99,4 +99,11 @@ rules:
   # routes, no routes in services, no models/routes in panel components, etc.
 
 custom: []
-waivers: []
+waivers:
+  - path: panel/src/hooks/__tests__/use-tasks-null-guards.test.tsx
+    rule: no_components_in_hooks
+    reason: >-
+      `wrapper` is a QueryClientProvider test fixture local to this test file,
+      not a production component — the same pattern already used by the
+      pre-existing use-agents.test.tsx / use-observability.test.tsx hook
+      tests colocated under hooks/__tests__/.
diff --git a/docs/frontend/api-rate-limiting.md b/docs/frontend/api-rate-limiting.md
new file mode 100644
index 00000000..870c6b7c
--- /dev/null
+++ b/docs/frontend/api-rate-limiting.md
@@ -0,0 +1,44 @@
+# API rate limiting and retry behavior
+
+The API client (`panel/src/lib/api/client.ts`) gates retries on HTTP 429 (rate limit) responses by HTTP method to prevent accidental duplicate side effects from replayed requests.
+
+## Automatic retries (safe methods)
+
+**GET** and **PUT** requests automatically retry up to 3 times on a 429 response:
+
+- **GET** has no side effect — retrying is always safe.
+- **PUT** is a full-resource replace — replaying it is a no-op past the first apply (idempotent).
+
+When a retry succeeds, the user sees a toast: `"Rate limited by <provider>. The system has paused operations and will resume automatically in ~Ns."`
+
+When retries are exhausted, the same toast appears, but the operation completes (after backoff delay).
+
+## Manual retries (stateful methods)
+
+**POST**, **PATCH**, and **DELETE** requests do NOT auto-retry on a 429, because:
+
+- **POST** can create a duplicate resource if replayed.
+- **PATCH** can double-apply a partial update.
+- **DELETE** can be replayed, causing confusion about the resource's state.
+
+A stateful request that hits a 429 fails immediately with a toast: `"Rate limited by <provider>. This action was not automatically retried to avoid duplicating it — please try again in ~Ns."`
+
+## Idempotency-key override (client-side gate only, not a working feature yet)
+
+`isRetrySafe` also accepts an `X-Idempotency-Key` header as an override: a POST/PATCH/DELETE request carrying that header is treated as retry-safe and auto-retries on a 429 the same as GET/PUT.
+
+This is client-side scaffolding, not a live contract today:
+
+- No call site in the panel sets `X-Idempotency-Key` — every POST/PATCH/DELETE in the app currently takes the no-retry path above.
+- There is no backend support for the header at all. The API does not store or check idempotency keys, so nothing would deduplicate a replayed request even if one were retried this way.
+
+Idempotency-key retry — the client attaching a key and the backend storing it to return a cached result on a repeat — is a **possible future enhancement**, not something implemented server-side. Don't set this header expecting deduplication; until the backend catches up, it would only unlock a client-side retry with no safety net behind it.
+
+## Implementation notes
+
+The retry decision is made by the exported `isRetrySafe(config)` function, which checks:
+
+1. The HTTP method (case-insensitive).
+2. For POST/PATCH/DELETE, the presence of the `X-Idempotency-Key` header.
+
+This pattern is tested in `panel/src/lib/__tests__/client.test.ts` and enforced at the request-interceptor level in `client.ts`.
diff --git a/docs/frontend/hooks.md b/docs/frontend/hooks.md
index 0ebe7955..8480e499 100644
--- a/docs/frontend/hooks.md
+++ b/docs/frontend/hooks.md
@@ -105,3 +105,36 @@ Returns a `PageRefreshState` object:
 
 - `panel/src/components/providers.tsx` was renamed to `panel/src/components/app-providers.tsx` so that `@/components/providers` could be used as a barrel export for `PageRefreshProvider`. Update any direct import of the root providers component from `@/components/providers` to `@/components/app-providers`.
 - The earlier scope-keyed provider files (`panel/src/components/page-refresh-provider.tsx` and `panel/src/store/page-refresh-context.ts`) were deleted. The current implementation lives in `panel/src/components/providers/page-refresh-provider.tsx` and is consumed through `usePageRefresh` from `@/hooks`.
+
+## Data-hook null-guard audit
+
+Every useQuery hook in `panel/src/hooks/` has been audited for missing `enabled` guards on undefined/null IDs, staleTime mismatches, and refetchInterval leaks on unmount.
+
+### Audit results
+
+All hooks carrying id-driven queries (`useTask`, `useSubtasks`, `useBoardReview`, `useTaskFindings`, `useTaskCollisionMap`, `useProject`, `useWorkSession`, `useWorkSessionForTask`, `useAgentStatus`, `useAgentDefinition`, `useJournalByAgent`, `useJournalEntry`, `useNotification`, `useGitStatus`, `useGitLog`, `useGitBranches`, `useGitDiff`, `useGitFile`, `useMemberScorecard`, and others) already carry correct `enabled: !!id` guards preventing undefined/null IDs from reaching the API.
+
+**Special case: board-review polls.** `useTask` includes a conditional `refetchInterval` when the task belongs to the Board team and `board_review_complete` is still `false`. The interval is correctly wired to self-disable via a selector function — once the backend reports `board_review_complete: true`, the refetchInterval gate closes and no further polls are scheduled. TanStack Query's `Observer` already tears down the interval timer on unmount, so there is no lifecycle leak.
+
+No code changes were required. A regression test suite (`panel/src/hooks/__tests__/use-tasks-null-guards.test.tsx`) verifies the enabled guards and the board-review poll behavior with fake timers.
+
+### Using these hooks safely
+
+When calling any id-driven hook, always pass the id from a verified source:
+
+```tsx
+import { useTask } from "@/hooks";
+
+export function TaskDetail({ taskId }: { taskId: string | undefined }) {
+  // The hook's `enabled` guard ensures no API call occurs when taskId is empty
+  const { data, isLoading, error } = useTask(taskId);
+
+  if (!taskId) return <p>No task selected</p>;
+  if (isLoading) return <p>Loading...</p>;
+  if (error) return <p>Error: {error.message}</p>;
+
+  return <div>{data?.title}</div>;
+}
+```
+
+No manual guard is needed before calling the hook — the `enabled: !!taskId` guard is built in and prevents wasted API calls and race conditions.
diff --git a/docs/rag/lifecycle/status-transitions.md b/docs/rag/lifecycle/status-transitions.md
index d7f39fdf..083614ab 100644
--- a/docs/rag/lifecycle/status-transitions.md
+++ b/docs/rag/lifecycle/status-transitions.md
@@ -11,6 +11,7 @@
 | awaiting_documentation | claimed | claim | documenter |
 | awaiting_pm_review | awaiting_ceo_approval | escalate_to_ceo | head_marketing, main_pm, product_owner |
 | awaiting_pm_review | cancelled | cancel | cell_pm, ceo, main_pm |
+| awaiting_pm_review | claimed | claim | cell_pm, main_pm |
 | awaiting_pm_review | completed | complete | cell_pm, main_pm |
 | awaiting_pm_review | needs_revision | request_changes | cell_pm, main_pm |
 | awaiting_pr_review | awaiting_pm_review | pr_pass | pr_reviewer |
diff --git a/panel/docs/frontend/hooks.md b/panel/docs/frontend/hooks.md
new file mode 100644
index 00000000..98583d4d
--- /dev/null
+++ b/panel/docs/frontend/hooks.md
@@ -0,0 +1,432 @@
+# Frontend Hooks Reference
+
+This document covers the reusable React hooks exported from `@/hooks` (principally `panel/src/hooks/use-websocket.ts`), with emphasis on the WebSocket message stream patterns and reconnect handling.
+
+## Overview
+
+The panel's real-time coordination streams are built on shared, ref-counted WebSocket connections that fan messages to multiple subscribers. This architecture eliminates duplicate connections to the same endpoint and ensures consistent state across different components that consume the same stream.
+
+### Connection Architecture
+
+- **Shared connections**: Two consumers of the same endpoint (e.g., A2A live view + rate-limit banner both reading `/ws/system`) now share a single WebSocket connection instead of each opening their own.
+- **Message fanning**: The shared connection broadcasts incoming messages to all registered subscribers.
+- **State syncing**: New subscribers are immediately replayed the connection's current state (e.g., a component mounting mid-reconnection sees `connecting` instead of stale `disconnected`).
+
+### Message Loss & Reconnect Handling
+
+The WebSocket connection at `panel/src/lib/websocket/connection.ts` (lines 91–119) has **no server-side message buffering or replay**. When the socket drops and reconnects:
+
+- Any frame published while disconnected is lost from the WebSocket stream forever.
+- Specialized hooks like `useNotificationStream` implement **REST catch-up** to fill the gap: they fetch unread notifications at REST the moment the socket recovers, so no message is silently lost.
+
+This is a point-in-time catch-up strategy, not a byte-for-byte replay — a deliberate trade-off documented in the task acceptance criteria.
+
+---
+
+## `useWebSocket<T>(endpoint, queryParams?, enabled?)`
+
+The foundation hook for subscribing to any WebSocket endpoint. All other hooks (`useNotificationStream`, `useAgentStream`, `useA2ALiveStream`) build on top of it.
+
+### Parameters
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `endpoint` | string | — | Path after `/ws/`, e.g., `/notifications/{agentId}` or `/system` |
+| `queryParams` | `Record<string, string>` | `undefined` | Optional query string as an object, e.g., `{ viewer_id: "..." }` |
+| `enabled` | boolean | `true` | Enable/disable the connection (useful for conditional subscriptions) |
+
+### Return Value
+
+```typescript
+{
+  state: ConnectionState;           // "disconnected", "connecting", "reconnecting", "connected"
+  lastMessage: T | null;            // The most recent message
+  messages: T[];                    // Ring buffer of ≤100 messages (STREAM_MAX_MESSAGES)
+  disconnect: () => void;           // Manually tear down the connection
+  clearMessages: () => void;        // Clear the buffer
+  isConnected: boolean;             // Shorthand for state === "connected"
+  isConnecting: boolean;            // Shorthand for state === "connecting" || "reconnecting"
+}
+```
+
+### Example: Raw WebSocket Consumption
+
+```tsx
+import { useWebSocket } from "@/hooks";
+
+export function MyAgentMonitor({ agentId }: { agentId: string }) {
+  const { state, lastMessage, messages, isConnected } = useWebSocket(
+    `/agents/${agentId}`,
+    { viewer_id: CEO_AGENT_ID },
+    !!agentId  // disable if agentId is falsy
+  );
+
+  return (
+    <>
+      <p>Connection: {state}</p>
+      {isConnected && <p>Last update: {lastMessage?.timestamp}</p>}
+      <ul>
+        {messages.map((msg, i) => (
+          <li key={i}>{JSON.stringify(msg)}</li>
+        ))}
+      </ul>
+    </>
+  );
+}
+```
+
+---
+
+## `useNotificationStream()`
+
+Subscribes to **CEO notifications** via `/ws/notifications/{CEO_AGENT_ID}`. This is the only subscription actively using the REST catch-up fallback to guarantee no notification is lost during a reconnect.
+
+### Return Value
+
+```typescript
+{
+  state: ConnectionState;
+  lastMessage: NotificationMessage | null;
+  notifications: NotificationMessage[];    // Deduped by notification_id
+  allMessages: NotificationMessage[];      // All messages from WS (raw)
+  clearMessages: () => void;               // Clears both notifications AND cached REST catch-up batch
+  isConnected: boolean;
+  isConnecting: boolean;
+}
+```
+
+### Reconnect Behavior (Key Change)
+
+When the WebSocket reconnects (transitions from `disconnected` / `reconnecting` → `connected`):
+
+1. **On initial connect**: No REST fetch occurs.
+2. **On a real reconnect** (socket was up before, dropped, and recovered):
+   - A `GET /api/notifications?unread_only=true` fetch fires immediately.
+   - Unread notifications from this fetch are transformed into notification frames and held in local state.
+   - These cached frames are merged with live frames **before** dedup.
+   - The dedup logic ensures no notification appears twice (caught-up notification wins if also delivered live).
+
+3. **Fetch failure**: Silently tolerated — live WS delivery resumes regardless. The catchup is best-effort.
+
+### Deduplication Strategy
+
+The `notifications` array is deduped by `notification_id` using a newest→oldest walk:
+
+- Walk backward through `[...cachedCatchup, ...liveMessages]`.
+- Keep only the first occurrence of each unique `notification_id` (newest wins).
+- Restore arrival order.
+- Notifications without an id (older frames) are always kept.
+
+The cached catch-up batch is placed **ahead** of live frames so live delivery can never be shadowed by an older cached copy.
+
+### Clearing Notifications
+
+Calling `clearMessages()` clears:
+- The live message buffer.
+- The cached catch-up batch.
+
+This prevents an immediate repopulation from cached state after a user clears the notification badge.
+
+### Example
+
+```tsx
+import { useNotificationStream } from "@/hooks";
+
+export function NotificationBell() {
+  const { notifications, isConnected, clearMessages } = useNotificationStream();
+
+  return (
+    <>
+      <button onClick={clearMessages}>
+        Clear ({notifications.length})
+      </button>
+      {!isConnected && <span className="dot" title="offline" />}
+    </>
+  );
+}
+```
+
+---
+
+## `useAgentStream(agentId)`
+
+Subscribes to live agent output (streaming work-in-progress) via `/ws/agents/{agentId}`.
+
+### Parameters
+
+| Parameter | Type | Description |
+|-----------|------|-------------|
+| `agentId` | string \| null | The agent UUID. Pass `null` to disable. |
+
+### Return Value
+
+```typescript
+{
+  state: ConnectionState;
+  lastMessage: AgentStreamMessage | null;
+  messages: AgentStreamMessage[];           // Raw frames
+  streamChunks: string[];                   // Extracted chunk strings
+  streamOutput: string;                     // All chunks concatenated
+  clearMessages: () => void;
+  isConnected: boolean;
+  isConnecting: boolean;
+}
+```
+
+### Message Type
+
+```typescript
+interface AgentStreamMessage {
+  type: "connected" | "agent.stream";
+  agent_id?: string;
+  chunk?: string;                           // Output fragment
+  watcher_count?: number;                   // Live viewer count
+  timestamp?: string;
+}
+```
+
+### Example
+
+```tsx
+import { useAgentStream } from "@/hooks";
+
+export function AgentOutputPanel({ agentId }: { agentId: string }) {
+  const { streamOutput, isConnecting, messages } = useAgentStream(agentId);
+
+  return (
+    <div className="output">
+      {isConnecting && <em>Connecting…</em>}
+      <code>{streamOutput}</code>
+      <p className="meta">
+        {messages.length} frames • {streamOutput.length} chars
+      </p>
+    </div>
+  );
+}
+```
+
+---
+
+## `useA2ALiveStream()`
+
+Subscribes to live **agent-to-agent** (A2A) messages via `/ws/system`. The system stream also carries rate-limit and usage events; this hook filters to `a2a.message` frames only.
+
+### Return Value
+
+```typescript
+{
+  state: ConnectionState;
+  lastMessage: A2ASystemMessage | null;
+  a2aMessages: A2ASystemMessage[];         // Filtered to type === "a2a.message"
+  allMessages: A2ASystemMessage[];         // All frames (includes usage/rate-limit)
+  clearMessages: () => void;
+  isConnected: boolean;
+  isConnecting: boolean;
+}
+```
+
+### Message Type
+
+```typescript
+interface A2ASystemMessage {
+  type: "connected" | "a2a.message";
+  conversation_id?: string;
+  message_id?: string;
+  task_id?: string;
+  from_agent?: string;
+  to_agent?: string;
+  skill?: string | null;
+  body_excerpt?: string;                   // Capped; fetch full body via REST
+  timestamp?: string;
+}
+```
+
+### Important Note: Excerpt-Only Delivery
+
+A2A frames carry a **capped excerpt**, not the full message body. Consumers must:
+
+1. Listen to `a2a.message` frames.
+2. Invalidate their **REST query** for the full message (e.g., `GET /api/a2a/conversations/{id}`).
+3. Fetch the full body from the REST endpoint.
+
+Do not render the `body_excerpt` as the message — it is metadata only.
+
+### Reconnect Fallback
+
+The A2A hook already has a working reconnect fallback at the consumer level: `panel/src/components/a2a/a2a-view.tsx` invalidates the entire a2a query family both per-frame (lines 185–194, on every `a2a.message` frame) and on the disconnected→connected edge (lines 212–218, gated on a `prevConnected` ref so it never fires on initial mount), ensuring no message is lost. No change was needed for this hook.
+
+### Example
+
+```tsx
+import { useA2ALiveStream } from "@/hooks";
+import { useQuery } from "@tanstack/react-query";
+
+export function A2ALiveView() {
+  const { a2aMessages, isConnected } = useA2ALiveStream();
+  const { data: conversations } = useQuery({
+    queryKey: ["a2a", "conversations"],
+    // Auto-fetched when a2aMessages changes (invalidation on frame)
+  });
+
+  return (
+    <div>
+      <p>
+        {isConnected ? "Connected" : "Offline"}
+        {a2aMessages.length > 0 && " (live updates)"}
+      </p>
+      {/* Render conversations with full bodies from REST */}
+    </div>
+  );
+}
+```
+
+---
+
+## `useConnectionStatus()`
+
+Tracks the connection state of **all active subscriptions** in a single hook. Useful for global connection indicators.
+
+### Return Value
+
+```typescript
+{
+  connections: Record<string, ConnectionState>;  // { [endpoint]: state }
+  updateConnection: (id: string, state: ConnectionState) => void;
+  removeConnection: (id: string) => void;
+  hasActiveConnections: boolean;                 // Any connection is up/ing
+  allConnected: boolean;                         // Every connection is connected
+}
+```
+
+### Example: Global Status Indicator
+
+```tsx
+import { useConnectionStatus } from "@/hooks";
+
+export function GlobalConnectionStatus() {
+  const { hasActiveConnections, allConnected } = useConnectionStatus();
+
+  return (
+    <div className="status-badge">
+      {allConnected && <span className="icon-check">Connected</span>}
+      {hasActiveConnections && !allConnected && (
+        <span className="icon-sync">Reconnecting…</span>
+      )}
+      {!hasActiveConnections && <span className="icon-offline">Offline</span>}
+    </div>
+  );
+}
+```
+
+---
+
+## Testing
+
+The hooks come with comprehensive test coverage in `panel/src/hooks/__tests__/`:
+
+- **`use-websocket.test.tsx`**: Core hook mechanics (shared connection, fan-out, state replay).
+- **`use-notification-stream.test.tsx`**: REST catch-up verification (new; covers the reconnect fallback):
+  - No fetch on initial connect.
+  - Fetch + fold-in on a real reconnect.
+  - Dedup prevents double-counting when the same notification arrives both via catch-up and live.
+  - `clearMessages()` drops the cached catch-up batch.
+
+Run tests with:
+
+```bash
+pnpm test
+```
+
+---
+
+## Audited Hooks: Reconnect Coverage
+
+Three hooks were audited for reconnect message-loss risk:
+
+| Hook | Connection Type | Fallback | Status |
+|------|-----------------|----------|--------|
+| `useNotificationStream` | CEO notification stream | REST catch-up (`GET /notifications?unread_only=true`) | ✅ Hardened |
+| `useA2ALiveStream` | A2A + system stream | REST query invalidation (`a2a-view.tsx`) | ✅ Verified |
+| Rate-limit consumers | System stream (`/ws/system`) | REST resync (rate-limit-banner.tsx, usage-overview-panel.tsx) | ✅ Verified |
+
+No defects were found in `useA2ALiveStream` or rate-limit consumption; the REST polling fallbacks were already in place and tested.
+
+### Adding a New Reconnect Fallback
+
+Picking a fallback strategy for a new WS-consuming hook comes down to what kind of data it carries:
+
+1. **Event data** (can only happen once, e.g. a notification) — implement REST catch-up: fetch unread/pending items over REST the moment the socket recovers, fold them into local state, and dedup against live frames (see `useNotificationStream` above).
+2. **State data** (always available via REST) — implement query invalidation: invalidate the relevant React Query cache keys on the disconnected→connected edge and let components refetch (see `useA2ALiveStream` above).
+3. **Always** add a regression test that simulates a disconnect/reconnect cycle with fake timers.
+
+---
+
+## Best Practices
+
+1. **Always disable on falsy keys**: Pass a conditional `enabled` flag (third param) if your hook depends on a variable parameter. This prevents spurious connections and ensures cleanup.
+
+   ```tsx
+   const { messages } = useWebSocket(
+     `/agents/${agentId}`,
+     undefined,
+     !!agentId  // disable if agentId is null/undefined
+   );
+   ```
+
+2. **Invalidate REST queries on frame**: When receiving a frame (especially A2A excerpts), trigger a React Query invalidation to fetch fresh data:
+
+   ```tsx
+   const queryClient = useQueryClient();
+   useEffect(() => {
+     if (a2aMessages.length > 0) {
+       queryClient.invalidateQueryData({ queryKey: ["a2a"] });
+     }
+   }, [a2aMessages, queryClient]);
+   ```
+
+3. **Don't hold onto stale messages**: The message buffer is bounded to 100 frames. Don't assume it's a complete history — treat it as a stream.
+
+4. **Catch fetch failures gracefully**: All REST fallbacks (like the notification catch-up) are best-effort and fail silently. Live WS delivery is not guaranteed to block on fetch completion.
+
+5. **Use `isConnecting` for UI feedback**: Show a loading state when `isConnecting` is true, not just when `!isConnected`.
+
+   ```tsx
+   {isConnecting && <Spinner />}
+   {isConnected && <CheckMark />}
+   ```
+
+---
+
+## Connection Types & States
+
+### ConnectionState
+
+```typescript
+type ConnectionState = 
+  | "disconnected"  // Not connected; attempting to reconnect or no connection attempt yet
+  | "connecting"    // Initial connection attempt
+  | "reconnecting"  // Reconnection after a close (watchdog/network failure)
+  | "connected"     // Stable connection, messages flowing
+```
+
+### State Transitions
+
+```
+disconnected ──> connecting ──> connected
+                                    ^
+                                    │
+                              (watchdog fires)
+                                    │
+                          reconnecting ──┘
+```
+
+On a reconnect transition (`reconnecting` → `connected`), hooks like `useNotificationStream` fire their REST catch-up fetch.
+
+---
+
+## Further Reading
+
+- **Control panel README**: `panel/README.md`
+- **WebSocket connection implementation**: `panel/src/lib/websocket/connection.ts`
+- **API client**: `panel/src/lib/api/`
+- **Notification types & components**: `panel/src/app/(dashboard)/notifications/`
diff --git a/panel/lib/lifecycle.json b/panel/lib/lifecycle.json
index c673a92e..8d0aa4c3 100644
--- a/panel/lib/lifecycle.json
+++ b/panel/lib/lifecycle.json
@@ -2,6 +2,7 @@
   "claim_rules": {
     "auditor": [],
     "cell_pm": [
+      "awaiting_pm_review",
       "needs_revision",
       "pending"
     ],
@@ -16,6 +17,7 @@
     ],
     "head_marketing": [],
     "main_pm": [
+      "awaiting_pm_review",
       "needs_revision",
       "pending"
     ],
@@ -529,6 +531,15 @@
       "source": "awaiting_pm_review",
       "target": "cancelled"
     },
+    {
+      "action": "claim",
+      "roles": [
+        "cell_pm",
+        "main_pm"
+      ],
+      "source": "awaiting_pm_review",
+      "target": "claimed"
+    },
     {
       "action": "complete",
       "roles": [
diff --git a/panel/src/components/dashboard/__tests__/release-proposal-card-status-feedback.test.tsx b/panel/src/components/dashboard/__tests__/release-proposal-card-status-feedback.test.tsx
new file mode 100644
index 00000000..89fb64ea
--- /dev/null
+++ b/panel/src/components/dashboard/__tests__/release-proposal-card-status-feedback.test.tsx
@@ -0,0 +1,160 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { fireEvent, render, screen, waitFor } from "@testing-library/react";
+import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
+import type { ReactNode } from "react";
+import type { ReleaseProposal } from "@/lib/api/release";
+import { PageRefreshProvider } from "@/components/providers";
+
+// Unlike release-proposal-card.test.tsx (which stubs useMutation entirely to
+// test the query-failure/execute-status surfacing paths), these tests
+// exercise the REAL useMutation onSuccess handler — mirrors the
+// x-post-queue/video-post-queue/roadmap-review-queue test pattern — so
+// approve()'s per-status toast copy is actually asserted.
+const { resolveApproveRef } = vi.hoisted(() => ({
+  resolveApproveRef: { current: null as null | ((v: unknown) => void) },
+}));
+
+const { getProposal, approve, reject } = vi.hoisted(() => ({
+  getProposal: vi.fn(
+    async (): Promise<ReleaseProposal> => ({
+      task_id: "t1",
+      title: "Cut v0.14.0",
+      status: "awaiting_ceo_approval",
+      required_changes: null,
+      report: {
+        proposed_version: "0.14.0",
+        bump_kind: "minor",
+        change_summary: ["feat: metrics"],
+        drafted_changelog: "## 0.14.0\n- metrics",
+        version_bump_plan: ["pyproject.toml"],
+        gaps: [],
+        migration_notes: [],
+        gate_state: "green",
+      },
+    }),
+  ),
+  // Deferred so the test can freeze the approve mid-flight.
+  approve: vi.fn(
+    () =>
+      new Promise((r) => {
+        resolveApproveRef.current = r as (v: unknown) => void;
+      }),
+  ),
+  reject: vi.fn(async () => ({})),
+}));
+
+vi.mock("@/lib/api", () => ({
+  releaseApi: { getProposal, approve, reject },
+}));
+
+const { toast } = vi.hoisted(() => ({
+  toast: { success: vi.fn(), warning: vi.fn(), info: vi.fn(), error: vi.fn() },
+}));
+vi.mock("sonner", () => ({ toast }));
+
+import { ReleaseProposalCard } from "../release-proposal-card";
+
+function withProviders(ui: ReactNode) {
+  const client = new QueryClient({
+    defaultOptions: { queries: { retry: false }, mutations: { retry: false } },
+  });
+  return (
+    <QueryClientProvider client={client}>
+      <PageRefreshProvider>{ui}</PageRefreshProvider>
+    </QueryClientProvider>
+  );
+}
+
+async function clickApprove() {
+  render(withProviders(<ReleaseProposalCard />));
+  fireEvent.click(
+    await screen.findByRole("button", { name: /Approve & publish/i }),
+  );
+  // Radix marks the rest of the page inert/aria-hidden once the dialog opens
+  // — the main card's own button drops out of the accessible tree, so this
+  // now uniquely matches the dialog's confirm button.
+  await screen.findByRole("dialog");
+  fireEvent.click(
+    await screen.findByRole("button", { name: /Approve & publish/i }),
+  );
+  await waitFor(() => expect(approve).toHaveBeenCalled());
+}
+
+describe("ReleaseProposalCard — status feedback (silent-bug-sweep #c)", () => {
+  beforeEach(() => {
+    getProposal.mockClear();
+    approve.mockClear();
+    reject.mockClear();
+    toast.success.mockClear();
+    toast.warning.mockClear();
+    toast.info.mockClear();
+    toast.error.mockClear();
+    resolveApproveRef.current = null;
+  });
+  afterEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it.each([
+    [
+      "already_in_progress",
+      "A release execute is already in progress for this proposal.",
+    ],
+    [
+      "redis_unavailable",
+      "Redis is unavailable — can't acquire the release mutex.",
+    ],
+    ["lock_lost", "The release lock was lost mid-execute — retry the approve."],
+    ["gate_failed", "Release halted (gate_failed): the gate is red"],
+  ])("shows distinct feedback for the %s status", async (status, message) => {
+    await clickApprove();
+
+    resolveApproveRef.current?.({
+      status,
+      version: "0.14.0",
+      files_changed: [],
+      commit_sha: null,
+      release_url: null,
+      detail: "the gate is red",
+    });
+
+    await waitFor(() => expect(toast.warning).toHaveBeenCalledWith(message));
+    expect(toast.success).not.toHaveBeenCalled();
+  });
+
+  it("shows a success toast for the published status", async () => {
+    await clickApprove();
+
+    resolveApproveRef.current?.({
+      status: "published",
+      version: "0.14.0",
+      files_changed: ["pyproject.toml"],
+      commit_sha: "abc123",
+      release_url: "https://github.com/example/example/releases/tag/v0.14.0",
+      detail: "published",
+    });
+
+    await waitFor(() =>
+      expect(toast.success).toHaveBeenCalledWith("Published v0.14.0"),
+    );
+  });
+
+  it("shows an info toast for the accepted (background-dispatched) status", async () => {
+    await clickApprove();
+
+    resolveApproveRef.current?.({
+      status: "accepted",
+      version: "0.14.0",
+      files_changed: [],
+      commit_sha: null,
+      release_url: null,
+      detail: "dispatched",
+    });
+
+    await waitFor(() =>
+      expect(toast.info).toHaveBeenCalledWith(
+        "Release execute dispatched — running in the background. This card updates as it progresses.",
+      ),
+    );
+  });
+});
diff --git a/panel/src/components/dashboard/__tests__/roadmap-review-queue.test.tsx b/panel/src/components/dashboard/__tests__/roadmap-review-queue.test.tsx
index 7c6a4da0..04041d34 100644
--- a/panel/src/components/dashboard/__tests__/roadmap-review-queue.test.tsx
+++ b/panel/src/components/dashboard/__tests__/roadmap-review-queue.test.tsx
@@ -58,6 +58,11 @@ vi.mock("@/lib/api", () => ({
   roadmapApi: { listCycles, approveItem, rejectItem },
 }));
 
+const { toast } = vi.hoisted(() => ({
+  toast: { success: vi.fn(), warning: vi.fn(), error: vi.fn() },
+}));
+vi.mock("sonner", () => ({ toast }));
+
 import { RoadmapReviewQueue } from "../roadmap-review-queue";
 
 function withQueryClient(ui: ReactNode) {
@@ -72,6 +77,9 @@ describe("RoadmapReviewQueue", () => {
     listCycles.mockClear();
     approveItem.mockClear();
     rejectItem.mockClear();
+    toast.success.mockClear();
+    toast.warning.mockClear();
+    toast.error.mockClear();
     resolveApproveRef.current = null;
   });
   afterEach(() => {
@@ -140,4 +148,48 @@ describe("RoadmapReviewQueue", () => {
     await waitFor(() => expect(listCycles).toHaveBeenCalled());
     expect(container).toBeEmptyDOMElement();
   });
+
+  // F(silent-bug-sweep #c): RoadmapService uses its own disjoint status
+  // vocabulary (approved/already_approved/invalid_state/rejected/
+  // already_rejected) — none of the 7 named cross-queue statuses
+  // (already_in_progress, redis_unavailable, lock_lost, post_failed,
+  // posted_partial, no_platforms, no_credentials) ever reach this queue, so
+  // this locks in distinct feedback for roadmap's own statuses instead.
+  it.each([
+    [
+      "already_approved",
+      "this item was already approved",
+      "Item approved — added to the backlog",
+    ],
+    [
+      "invalid_state",
+      "item is 'rejected', not proposed — cannot approve",
+      "item is 'rejected', not proposed — cannot approve",
+    ],
+  ])(
+    "shows distinct feedback for the %s status",
+    async (status, detail, message) => {
+      render(withQueryClient(<RoadmapReviewQueue />));
+      const approveButtons = await screen.findAllByRole("button", {
+        name: /Approve/,
+      });
+      fireEvent.click(approveButtons[0]);
+      await waitFor(() => expect(approveItem).toHaveBeenCalled());
+
+      resolveApproveRef.current?.({
+        status,
+        item_id: "item-0",
+        materialized_task_id: null,
+        detail,
+      });
+
+      await waitFor(() => {
+        if (status === "already_approved") {
+          expect(toast.success).toHaveBeenCalledWith(message);
+        } else {
+          expect(toast.warning).toHaveBeenCalledWith(message);
+        }
+      });
+    },
+  );
 });
diff --git a/panel/src/components/dashboard/__tests__/video-post-queue.test.tsx b/panel/src/components/dashboard/__tests__/video-post-queue.test.tsx
index fe619ab3..23f5287f 100644
--- a/panel/src/components/dashboard/__tests__/video-post-queue.test.tsx
+++ b/panel/src/components/dashboard/__tests__/video-post-queue.test.tsx
@@ -102,6 +102,11 @@ vi.mock("@/components/projects/project-selector", () => ({
   ),
 }));
 
+const { toast } = vi.hoisted(() => ({
+  toast: { success: vi.fn(), warning: vi.fn(), error: vi.fn() },
+}));
+vi.mock("sonner", () => ({ toast }));
+
 import { VideoPostQueue } from "../video-post-queue";
 
 function withQueryClient(ui: ReactNode) {
@@ -119,6 +124,9 @@ describe("VideoPostQueue", () => {
     reject.mockClear();
     requestVideo.mockClear();
     getMediaBlob.mockClear();
+    toast.success.mockClear();
+    toast.warning.mockClear();
+    toast.error.mockClear();
     resolveApproveRef.current = null;
     // jsdom has no Blob URL implementation. Distinct URLs per call so a
     // revoke can be asserted against the specific (stale) one it replaced.
@@ -558,4 +566,68 @@ describe("VideoPostQueue", () => {
     await screen.findByText("release");
     expect(document.querySelector("iframe")).not.toBeInTheDocument();
   });
+
+  // F(silent-bug-sweep #c): every VideoPostService.approve status must
+  // render a distinct, non-swallowed toast — regression guard, no code gap
+  // was found here (describeExecuteResult already branches every one of
+  // these).
+  it.each([
+    [
+      "posted_partial",
+      { status: "posted_partial", posted: { x: "1" }, detail: "tiktok: down" },
+      "Posted to some platforms — tiktok: down",
+    ],
+    [
+      "post_failed",
+      { status: "post_failed", posted: {}, detail: "both platforms down" },
+      "Posting failed: both platforms down",
+    ],
+    [
+      "already_in_progress",
+      { status: "already_in_progress", posted: {}, detail: "" },
+      "A post is already in progress for this draft.",
+    ],
+    [
+      "no_platforms",
+      { status: "no_platforms", posted: {}, detail: "" },
+      "This draft has no target platforms.",
+    ],
+    [
+      "lock_lost",
+      { status: "lock_lost", posted: {}, detail: "" },
+      "The post lock was lost mid-upload — retry the approve.",
+    ],
+    [
+      "redis_unavailable",
+      { status: "redis_unavailable", posted: {}, detail: "" },
+      "Redis is unavailable — can't acquire the post lock.",
+    ],
+  ])("shows distinct feedback for the %s status", async (_, result, message) => {
+    render(withQueryClient(<VideoPostQueue />));
+    await screen.findByText("release");
+    fireEvent.click(screen.getByRole("button", { name: /Approve/ }));
+    await waitFor(() => expect(approve).toHaveBeenCalled());
+
+    resolveApproveRef.current?.(result);
+
+    await waitFor(() => expect(toast.warning).toHaveBeenCalledWith(message));
+    expect(toast.success).not.toHaveBeenCalled();
+  });
+
+  it("shows a success toast for the posted status", async () => {
+    render(withQueryClient(<VideoPostQueue />));
+    await screen.findByText("release");
+    fireEvent.click(screen.getByRole("button", { name: /Approve/ }));
+    await waitFor(() => expect(approve).toHaveBeenCalled());
+
+    resolveApproveRef.current?.({
+      status: "posted",
+      posted: { x: "1", tiktok: "2" },
+      detail: "posted to all platforms",
+    });
+
+    await waitFor(() =>
+      expect(toast.success).toHaveBeenCalledWith("Posted to all platforms."),
+    );
+  });
 });
diff --git a/panel/src/components/dashboard/__tests__/x-post-queue.test.tsx b/panel/src/components/dashboard/__tests__/x-post-queue.test.tsx
index 373976ac..3c997f00 100644
--- a/panel/src/components/dashboard/__tests__/x-post-queue.test.tsx
+++ b/panel/src/components/dashboard/__tests__/x-post-queue.test.tsx
@@ -44,6 +44,11 @@ const { listPosts, approve, reject } = vi.hoisted(() => ({
 
 vi.mock("@/lib/api", () => ({ xApi: { listPosts, approve, reject } }));
 
+const { toast } = vi.hoisted(() => ({
+  toast: { success: vi.fn(), warning: vi.fn(), error: vi.fn() },
+}));
+vi.mock("sonner", () => ({ toast }));
+
 import { XPostQueue } from "../x-post-queue";
 
 function withQueryClient(ui: ReactNode) {
@@ -58,6 +63,9 @@ describe("XPostQueue", () => {
     listPosts.mockClear();
     approve.mockClear();
     reject.mockClear();
+    toast.success.mockClear();
+    toast.warning.mockClear();
+    toast.error.mockClear();
     resolveApproveRef.current = null;
   });
   afterEach(() => {
@@ -178,4 +186,55 @@ describe("XPostQueue", () => {
     await waitFor(() => expect(listPosts).toHaveBeenCalled());
     expect(container).toBeEmptyDOMElement();
   });
+
+  // F(silent-bug-sweep #c): every XPostService.approve status must render a
+  // distinct, non-swallowed toast — not a blanket success/failure.
+  it.each([
+    ["already_in_progress", "A post is already in progress for this draft."],
+    [
+      "no_credentials",
+      "No X credentials configured — set them below first.",
+    ],
+    ["post_failed", "Posting failed: the X API rejected the tweet"],
+    [
+      "redis_unavailable",
+      "Redis is unavailable — can't acquire the post lock.",
+    ],
+    ["already_posted", "Already posted — no-op."],
+  ])("shows distinct feedback for the %s status", async (status, message) => {
+    render(withQueryClient(<XPostQueue />));
+    const approveButtons = await screen.findAllByRole("button", {
+      name: /Approve/,
+    });
+    fireEvent.click(approveButtons[0]);
+    await waitFor(() => expect(approve).toHaveBeenCalled());
+
+    resolveApproveRef.current?.({
+      status,
+      tweet_id: null,
+      detail: "the X API rejected the tweet",
+    });
+
+    await waitFor(() => expect(toast.warning).toHaveBeenCalledWith(message));
+    expect(toast.success).not.toHaveBeenCalled();
+  });
+
+  it("shows a success toast for the posted status", async () => {
+    render(withQueryClient(<XPostQueue />));
+    const approveButtons = await screen.findAllByRole("button", {
+      name: /Approve/,
+    });
+    fireEvent.click(approveButtons[0]);
+    await waitFor(() => expect(approve).toHaveBeenCalled());
+
+    resolveApproveRef.current?.({
+      status: "posted",
+      tweet_id: "1",
+      detail: "ok",
+    });
+
+    await waitFor(() =>
+      expect(toast.success).toHaveBeenCalledWith("Posted to X."),
+    );
+  });
 });
diff --git a/panel/src/components/dashboard/release-proposal-card.tsx b/panel/src/components/dashboard/release-proposal-card.tsx
index 2225eb4a..ab4d7ba7 100644
--- a/panel/src/components/dashboard/release-proposal-card.tsx
+++ b/panel/src/components/dashboard/release-proposal-card.tsx
@@ -38,6 +38,20 @@ function gateBadgeVariant(
   return "secondary";
 }
 
+// Distinct copy for the concurrency/infra statuses a release execute can
+// come back with (already_in_progress / redis_unavailable / lock_lost) —
+// the rest (gate_failed, ci_failed, ...) fall through to the generic
+// "Release halted (status): detail" message, itself still status-specific.
+function describeHaltedStatus(result: ReleaseExecuteResult): string {
+  if (result.status === "already_in_progress")
+    return "A release execute is already in progress for this proposal.";
+  if (result.status === "redis_unavailable")
+    return "Redis is unavailable — can't acquire the release mutex.";
+  if (result.status === "lock_lost")
+    return "The release lock was lost mid-execute — retry the approve.";
+  return `Release halted (${result.status}): ${result.detail}`;
+}
+
 // A red gate / open gaps make publishing risky — the CEO should resolve them
 // first. Approval still runs the fail-closed executor, so it can't ship a bad
 // release; this only steers the CEO.
@@ -86,7 +100,7 @@ export function ReleaseProposalCard({ className }: { className?: string }) {
           "Release execute dispatched — running in the background. This card updates as it progresses.",
         );
       } else {
-        toast.warning(`Release halted (${result.status}): ${result.detail}`);
+        toast.warning(describeHaltedStatus(result));
       }
       closeDialog();
     },
diff --git a/panel/src/components/dashboard/x-post-queue.tsx b/panel/src/components/dashboard/x-post-queue.tsx
index a334fb00..99af024b 100644
--- a/panel/src/components/dashboard/x-post-queue.tsx
+++ b/panel/src/components/dashboard/x-post-queue.tsx
@@ -77,6 +77,10 @@ function describeExecuteResult(result: XPostExecuteResult): string {
     return "A post is already in progress for this draft.";
   if (result.status === "no_credentials")
     return "No X credentials configured — set them below first.";
+  if (result.status === "post_failed")
+    return `Posting failed: ${result.detail}`;
+  if (result.status === "redis_unavailable")
+    return "Redis is unavailable — can't acquire the post lock.";
   return `${result.status}: ${result.detail}`;
 }
 
diff --git a/panel/src/hooks/__tests__/use-notification-stream.test.tsx b/panel/src/hooks/__tests__/use-notification-stream.test.tsx
new file mode 100644
index 00000000..19adc11c
--- /dev/null
+++ b/panel/src/hooks/__tests__/use-notification-stream.test.tsx
@@ -0,0 +1,222 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { render, act, waitFor } from "@testing-library/react";
+import { useEffect } from "react";
+import type {
+  ConnectionState,
+  WebSocketOptions,
+} from "@/lib/websocket/connection";
+
+// connection.ts has no message buffering/replay across a reconnect (audited
+// lines 91-119): drive the WS state machine from the test via a mocked
+// connection and assert useNotificationStream's REST catch-up covers the gap
+// instead of silently losing whatever was published while disconnected.
+const hoisted = vi.hoisted(() => {
+  const instances: MockConnection[] = [];
+  class MockConnection {
+    url: string;
+    onMessage?: (data: unknown) => void;
+    onStateChange?: (state: ConnectionState) => void;
+    constructor(opts: WebSocketOptions) {
+      this.url = opts.url;
+      this.onMessage = opts.onMessage;
+      this.onStateChange = opts.onStateChange;
+      instances.push(this);
+    }
+    connect() {
+      this.onStateChange?.("connecting");
+      this.onStateChange?.("connected");
+    }
+    disconnect() {
+      this.onStateChange?.("disconnected");
+    }
+    getState() {
+      return "connected";
+    }
+    getLastPongAt() {
+      return Date.now();
+    }
+    checkPong() {}
+  }
+  return { instances, MockConnection };
+});
+
+vi.mock("@/lib/websocket/connection", () => ({
+  getWebSocketUrl: () => "ws://test/ws",
+  WebSocketConnection: hoisted.MockConnection,
+}));
+
+vi.mock("@/lib/constants", () => ({
+  CEO_AGENT_ID: "00000000-0000-0000-0000-000000000001",
+  STREAM_MAX_MESSAGES: 100,
+}));
+
+const listMock = vi.fn();
+vi.mock("@/lib/api/notifications", () => ({
+  notificationsApi: { list: (...args: unknown[]) => listMock(...args) },
+}));
+
+import {
+  useNotificationStream,
+  _resetSharedSocketsForTest,
+} from "../use-websocket";
+
+const resultRef: {
+  current: ReturnType<typeof useNotificationStream> | null;
+} = { current: null };
+
+function Harness() {
+  const stream = useNotificationStream();
+  useEffect(() => {
+    resultRef.current = stream;
+  });
+  return null;
+}
+
+function emptyList() {
+  return { items: [], total: 0, unread_count: 0, pending_ack_count: 0 };
+}
+
+describe("useNotificationStream — REST catch-up on reconnect", () => {
+  beforeEach(() => {
+    hoisted.instances.length = 0;
+    resultRef.current = null;
+    listMock.mockReset();
+    listMock.mockResolvedValue(emptyList());
+    _resetSharedSocketsForTest();
+  });
+  afterEach(() => {
+    vi.clearAllMocks();
+    _resetSharedSocketsForTest();
+  });
+
+  it("does not fetch a catch-up batch on the initial connect", async () => {
+    render(<Harness />);
+    await act(async () => {});
+    expect(listMock).not.toHaveBeenCalled();
+  });
+
+  it("fetches unread notifications on reconnect and folds them in", async () => {
+    listMock.mockResolvedValue({
+      items: [
+        {
+          id: "n1",
+          type: "task_update",
+          priority: "normal",
+          subject: "Missed while offline",
+          timestamp: "2026-07-21T00:00:00Z",
+        },
+      ],
+      total: 1,
+      unread_count: 1,
+      pending_ack_count: 0,
+    });
+
+    render(<Harness />);
+    const conn = hoisted.instances[0];
+    await act(async () => {});
+    expect(listMock).not.toHaveBeenCalled();
+
+    // Drop, then recover — the real sequence connection.ts drives on a
+    // watchdog/close event followed by scheduleReconnect().
+    act(() => {
+      conn.onStateChange?.("reconnecting");
+    });
+    act(() => {
+      conn.onStateChange?.("connecting");
+    });
+    act(() => {
+      conn.onStateChange?.("connected");
+    });
+
+    await waitFor(() =>
+      expect(listMock).toHaveBeenCalledWith({ unread_only: true }),
+    );
+    await waitFor(() =>
+      expect(resultRef.current?.notifications).toHaveLength(1),
+    );
+    expect(resultRef.current?.notifications[0].notification_id).toBe("n1");
+  });
+
+  it("does not surface the catch-up copy twice when the same notification also arrives live", async () => {
+    listMock.mockResolvedValue({
+      items: [
+        {
+          id: "n1",
+          type: "task_update",
+          priority: "normal",
+          subject: "Missed",
+          timestamp: "2026-07-21T00:00:00Z",
+        },
+      ],
+      total: 1,
+      unread_count: 1,
+      pending_ack_count: 0,
+    });
+    render(<Harness />);
+    const conn = hoisted.instances[0];
+    await act(async () => {});
+
+    act(() => {
+      conn.onStateChange?.("reconnecting");
+    });
+    act(() => {
+      conn.onStateChange?.("connecting");
+    });
+    act(() => {
+      conn.onStateChange?.("connected");
+    });
+    await waitFor(() => expect(listMock).toHaveBeenCalled());
+    await waitFor(() =>
+      expect(resultRef.current?.notifications).toHaveLength(1),
+    );
+
+    // The live frame for the same notification arrives right after reconnect.
+    act(() => {
+      conn.onMessage?.({
+        type: "notification",
+        notification_id: "n1",
+        subject: "Missed",
+        priority: "normal",
+      });
+    });
+
+    expect(resultRef.current?.notifications).toHaveLength(1);
+  });
+
+  it("clearMessages also drops the held catch-up batch (no repopulate-after-clear)", async () => {
+    listMock.mockResolvedValue({
+      items: [
+        {
+          id: "n1",
+          type: "task_update",
+          priority: "normal",
+          subject: "Missed",
+          timestamp: "2026-07-21T00:00:00Z",
+        },
+      ],
+      total: 1,
+      unread_count: 1,
+      pending_ack_count: 0,
+    });
+    render(<Harness />);
+    const conn = hoisted.instances[0];
+    await act(async () => {});
+    act(() => {
+      conn.onStateChange?.("reconnecting");
+    });
+    act(() => {
+      conn.onStateChange?.("connecting");
+    });
+    act(() => {
+      conn.onStateChange?.("connected");
+    });
+    await waitFor(() =>
+      expect(resultRef.current?.notifications).toHaveLength(1),
+    );
+
+    act(() => {
+      resultRef.current?.clearMessages();
+    });
+    expect(resultRef.current?.notifications).toHaveLength(0);
+  });
+});
diff --git a/panel/src/hooks/__tests__/use-tasks-null-guards.test.tsx b/panel/src/hooks/__tests__/use-tasks-null-guards.test.tsx
new file mode 100644
index 00000000..5dfd8f71
--- /dev/null
+++ b/panel/src/hooks/__tests__/use-tasks-null-guards.test.tsx
@@ -0,0 +1,110 @@
+import { describe, expect, it, vi, beforeEach, afterEach } from "vitest";
+import { renderHook, waitFor } from "@testing-library/react";
+import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
+import { type ReactNode } from "react";
+
+// Regression coverage for the data-hook null-guard audit: undefined/empty ids
+// must never reach the API (the `enabled` guard), and useTask's board-review
+// poll must stop the moment board_review_complete flips — the two things
+// called out as "known suspects" in the audit task. TanStack Query already
+// tears the poll's internal timer down on unmount (its Observer
+// unsubscribes), so the real regression risk is the poll condition itself,
+// not a missing cleanup — this exercises the condition end-to-end with fake
+// timers rather than re-deriving it in the test.
+
+const { get, getSubtasks } = vi.hoisted(() => ({
+  get: vi.fn(),
+  getSubtasks: vi.fn(),
+}));
+
+vi.mock("@/lib/api/tasks", async () => {
+  const actual =
+    await vi.importActual<typeof import("@/lib/api/tasks")>(
+      "@/lib/api/tasks",
+    );
+  return {
+    ...actual,
+    tasksApi: { ...actual.tasksApi, get, getSubtasks },
+  };
+});
+
+import { useTask, useSubtasks } from "@/hooks/use-tasks";
+import { Team } from "@/types";
+import type { Task } from "@/types";
+
+function wrapper({ children }: { children: ReactNode }) {
+  const client = new QueryClient({
+    defaultOptions: { queries: { retry: false } },
+  });
+  return <QueryClientProvider client={client}>{children}</QueryClientProvider>;
+}
+
+describe("data-hook null-guard audit", () => {
+  beforeEach(() => {
+    get.mockReset();
+    getSubtasks.mockReset();
+  });
+
+  it("useTask never calls the API when taskId is empty", () => {
+    renderHook(() => useTask(""), { wrapper });
+    expect(get).not.toHaveBeenCalled();
+  });
+
+  it("useSubtasks never calls the API when parentTaskId is empty", () => {
+    renderHook(() => useSubtasks(""), { wrapper });
+    expect(getSubtasks).not.toHaveBeenCalled();
+  });
+
+  it("useTask fetches once a real id is supplied", async () => {
+    get.mockResolvedValue({ id: "t1", team: Team.BACKEND } as Task);
+    renderHook(() => useTask("t1"), { wrapper });
+    await waitFor(() => expect(get).toHaveBeenCalledWith("t1"));
+  });
+
+  describe("board-review poll stops itself", () => {
+    beforeEach(() => {
+      vi.useFakeTimers({ shouldAdvanceTime: true });
+    });
+
+    afterEach(() => {
+      vi.useRealTimers();
+    });
+
+    it("polls a board task every 4s until board_review_complete flips true", async () => {
+      get.mockResolvedValue({
+        id: "board-task",
+        team: Team.BOARD,
+        board_review_complete: false,
+      } as Task);
+
+      renderHook(() => useTask("board-task"), { wrapper });
+      await vi.waitFor(() => expect(get).toHaveBeenCalledTimes(1));
+
+      // Still incomplete — the poll must fire again after 4s.
+      await vi.advanceTimersByTimeAsync(4000);
+      await vi.waitFor(() => expect(get).toHaveBeenCalledTimes(2));
+
+      // Board finishes reviewing — the next poll response reports completion.
+      get.mockResolvedValue({
+        id: "board-task",
+        team: Team.BOARD,
+        board_review_complete: true,
+      } as Task);
+      await vi.advanceTimersByTimeAsync(4000);
+      await vi.waitFor(() => expect(get).toHaveBeenCalledTimes(3));
+
+      // No further poll should be scheduled once complete.
+      await vi.advanceTimersByTimeAsync(10000);
+      expect(get).toHaveBeenCalledTimes(3);
+    });
+
+    it("never polls a non-board task", async () => {
+      get.mockResolvedValue({ id: "t1", team: Team.BACKEND } as Task);
+      renderHook(() => useTask("t1"), { wrapper });
+      await vi.waitFor(() => expect(get).toHaveBeenCalledTimes(1));
+
+      await vi.advanceTimersByTimeAsync(10000);
+      expect(get).toHaveBeenCalledTimes(1);
+    });
+  });
+});
diff --git a/panel/src/hooks/__tests__/use-websocket.test.tsx b/panel/src/hooks/__tests__/use-websocket.test.tsx
index b4fe3da2..067fc205 100644
--- a/panel/src/hooks/__tests__/use-websocket.test.tsx
+++ b/panel/src/hooks/__tests__/use-websocket.test.tsx
@@ -57,6 +57,13 @@ vi.mock("@/lib/constants", () => ({
   STREAM_MAX_MESSAGES: 100,
 }));
 
+// use-websocket.ts imports notificationsApi (for useNotificationStream's
+// reconnect catch-up) which transitively pulls in the API client — stub it so
+// this file's unrelated hook tests don't need a real API_URL/axios setup.
+vi.mock("@/lib/api/notifications", () => ({
+  notificationsApi: { list: vi.fn().mockResolvedValue({ items: [] }) },
+}));
+
 import { useWebSocket, _resetSharedSocketsForTest } from "../use-websocket";
 
 interface Frame {
diff --git a/panel/src/hooks/use-websocket.ts b/panel/src/hooks/use-websocket.ts
index 9166aec5..94d86ab3 100644
--- a/panel/src/hooks/use-websocket.ts
+++ b/panel/src/hooks/use-websocket.ts
@@ -6,6 +6,7 @@ import {
   getWebSocketUrl,
 } from "@/lib/websocket/connection";
 import { CEO_AGENT_ID, STREAM_MAX_MESSAGES } from "@/lib/constants";
+import { notificationsApi } from "@/lib/api/notifications";
 
 // Re-export ConnectionState type
 export type { ConnectionState } from "@/lib/websocket/connection";
@@ -251,16 +252,57 @@ export function useNotificationStream() {
     true,
   );
 
+  // connection.ts has no message buffering/replay: a notification published
+  // while the socket is down (disconnected/reconnecting) is gone from the WS
+  // stream forever, not merely delayed. Catch up over REST the moment the
+  // stream recovers from a real gap (not the initial mount's first connect)
+  // so a replay-suppressed dedup below can't hide something the bell never
+  // actually received.
+  const [catchup, setCatchup] = useState<NotificationMessage[]>([]);
+  const everConnectedRef = useRef(false);
+  const hadGapRef = useRef(false);
+  useEffect(() => {
+    if (state === "reconnecting" || state === "disconnected") {
+      hadGapRef.current = true;
+      return;
+    }
+    if (state !== "connected") return;
+    if (everConnectedRef.current && hadGapRef.current) {
+      hadGapRef.current = false;
+      notificationsApi
+        .list({ unread_only: true })
+        .then((res) => {
+          setCatchup(
+            res.items.map((n) => ({
+              type: "notification" as const,
+              notification_id: n.id,
+              notification_type: n.type,
+              subject: n.subject,
+              priority: n.priority,
+              timestamp: n.timestamp,
+            })),
+          );
+        })
+        .catch(() => {
+          // Best-effort: live WS delivery resumes regardless of catch-up
+          // success, so a fetch failure here isn't fatal.
+        });
+    }
+    everConnectedRef.current = true;
+  }, [state]);
+
   // Filter to notification events, de-duplicated by notification_id so a
   // stream replay (e.g. after a websocket reconnect) does not surface — or
   // count — the same notification twice. Walk newest→oldest keeping the most
   // recent copy of each id, then restore arrival order. Events without an id
-  // (older payloads) are always kept.
+  // (older payloads) are always kept. The REST catch-up batch is folded in
+  // ahead of the live WS messages so it can't shadow anything delivered live.
   const notifications = useMemo(() => {
+    const combined = [...catchup, ...messages];
     const seen = new Set<string>();
     const deduped: NotificationMessage[] = [];
-    for (let i = messages.length - 1; i >= 0; i--) {
-      const m = messages[i];
+    for (let i = combined.length - 1; i >= 0; i--) {
+      const m = combined[i];
       if (m.type !== "notification") continue;
       const id = m.notification_id;
       if (id) {
@@ -271,14 +313,21 @@ export function useNotificationStream() {
     }
     deduped.reverse();
     return deduped;
-  }, [messages]);
+  }, [messages, catchup]);
+
+  // Clear must drop the REST catch-up batch too — otherwise a cleared badge
+  // would immediately repopulate from the still-held catchup state.
+  const clearAll = useCallback(() => {
+    setCatchup([]);
+    clearMessages();
+  }, [clearMessages]);
 
   return {
     state,
     lastMessage,
     notifications,
     allMessages: messages,
-    clearMessages,
+    clearMessages: clearAll,
     isConnected,
     isConnecting,
   };
diff --git a/panel/src/lib/__tests__/client.test.ts b/panel/src/lib/__tests__/client.test.ts
index 40aaf73e..8e324de4 100644
--- a/panel/src/lib/__tests__/client.test.ts
+++ b/panel/src/lib/__tests__/client.test.ts
@@ -30,7 +30,8 @@ vi.mock("@/store/rate-limit-store", () => ({
 // ---------------------------------------------------------------------------
 // Import the function under test AFTER mocks are in place
 // ---------------------------------------------------------------------------
-import { getErrorMessage, isTgSurfacePath } from "@/lib/api/client";
+import { getErrorMessage, isTgSurfacePath, isRetrySafe } from "@/lib/api/client";
+import { AxiosHeaders } from "axios";
 
 // ---------------------------------------------------------------------------
 // Helpers
@@ -226,3 +227,37 @@ describe("isTgSurfacePath — the /tg login-redirect exemption", () => {
     expect(isTgSurfacePath("/")).toBe(false);
   });
 });
+
+// ---------------------------------------------------------------------------
+// isRetrySafe — the 429 retry-by-method gate
+// ---------------------------------------------------------------------------
+
+describe("isRetrySafe — 429 retry gated by HTTP method", () => {
+  function config(method: string, headers = new AxiosHeaders()) {
+    return { method, headers } as never;
+  }
+
+  it("retries GET and PUT unconditionally — the explicit PO safe-method carve-out", () => {
+    expect(isRetrySafe(config("get"))).toBe(true);
+    expect(isRetrySafe(config("GET"))).toBe(true);
+    expect(isRetrySafe(config("put"))).toBe(true);
+  });
+
+  it("does not retry POST/PATCH/DELETE without an idempotency key", () => {
+    expect(isRetrySafe(config("post"))).toBe(false);
+    expect(isRetrySafe(config("patch"))).toBe(false);
+    expect(isRetrySafe(config("delete"))).toBe(false);
+  });
+
+  it("retries POST/PATCH/DELETE when the caller attached an idempotency key", () => {
+    const headers = new AxiosHeaders();
+    headers.set("X-Idempotency-Key", "abc123");
+    expect(isRetrySafe(config("post", headers))).toBe(true);
+    expect(isRetrySafe(config("patch", headers))).toBe(true);
+    expect(isRetrySafe(config("delete", headers))).toBe(true);
+  });
+
+  it("returns false when config is missing", () => {
+    expect(isRetrySafe(undefined)).toBe(false);
+  });
+});
diff --git a/panel/src/lib/api/client.ts b/panel/src/lib/api/client.ts
index 7ab64868..dcbd6927 100644
--- a/panel/src/lib/api/client.ts
+++ b/panel/src/lib/api/client.ts
@@ -13,6 +13,24 @@ declare module "axios" {
 
 const RATE_LIMIT_MAX_RETRIES = 3;
 
+// GET/PUT are safe to auto-retry on a 429 — GET has no side effect and PUT is
+// a full-resource replace, so replaying it is a no-op past the first apply.
+// POST/PATCH/DELETE are NOT safe by default (a replayed POST can create a
+// duplicate task, a replayed DELETE/PATCH can double-apply a partial update)
+// — they only retry when the caller attached an idempotency key the backend
+// can dedupe on. Header name matches what a future idempotency-key-bearing
+// caller would set; no call site sets it yet, so these methods currently
+// skip the auto-retry entirely rather than risking a duplicate side effect.
+const RETRY_SAFE_METHODS = new Set(["get", "put"]);
+const IDEMPOTENCY_KEY_HEADER = "X-Idempotency-Key";
+
+export function isRetrySafe(config: AxiosError["config"]): boolean {
+  if (!config) return false;
+  const method = (config.method ?? "get").toLowerCase();
+  if (RETRY_SAFE_METHODS.has(method)) return true;
+  return Boolean(config.headers?.has?.(IDEMPOTENCY_KEY_HEADER));
+}
+
 // Create axios instance with default config
 // axios ^1.16.0 audit (2026-07-09): every call in panel/src rides this
 // browser instance (no proxy option, no maxRedirects/adapter override, no
@@ -146,22 +164,30 @@ api.interceptors.response.use(
       };
       useRateLimitStore.getState().hitRateLimit(hitEvent);
 
-      // Track retry count; retry the request (after backoff delay) until exhausted, then toast
-      const retryCount = (error.config?._retryCount ?? 0) + 1;
-      if (error.config) {
-        error.config._retryCount = retryCount;
-        if (retryCount < RATE_LIMIT_MAX_RETRIES) {
-          // Wait retryAfterSeconds before retrying — interceptor re-runs on each subsequent 429
-          const delayMs = safeRetryAfter * 1000;
-          return new Promise<void>((resolve) =>
-            setTimeout(resolve, delayMs),
-          ).then(() => api(error.config!));
+      if (isRetrySafe(error.config)) {
+        // Track retry count; retry the request (after backoff delay) until exhausted, then toast
+        const retryCount = (error.config?._retryCount ?? 0) + 1;
+        if (error.config) {
+          error.config._retryCount = retryCount;
+          if (retryCount < RATE_LIMIT_MAX_RETRIES) {
+            // Wait retryAfterSeconds before retrying — interceptor re-runs on each subsequent 429
+            const delayMs = safeRetryAfter * 1000;
+            return new Promise<void>((resolve) =>
+              setTimeout(resolve, delayMs),
+            ).then(() => api(error.config!));
+          }
         }
+        // Retries exhausted — notify the user via Sonner toast
+        toast.warning(
+          `Rate limited by ${provider}. The system has paused operations and will resume automatically in ~${safeRetryAfter}s.`,
+        );
+      } else {
+        // A non-idempotent write (POST/PATCH/DELETE) never auto-retries —
+        // replaying it could double-apply the action. Surface it once instead.
+        toast.warning(
+          `Rate limited by ${provider}. This action was not automatically retried to avoid duplicating it — please try again in ~${safeRetryAfter}s.`,
+        );
       }
-      // Retries exhausted — notify the user via Sonner toast
-      toast.warning(
-        `Rate limited by ${provider}. The system has paused operations and will resume automatically in ~${safeRetryAfter}s.`,
-      );
     }
 
     // Log comprehensive error info
diff --git a/roboco/api/routes/release.py b/roboco/api/routes/release.py
index c3e93373..78752e61 100644
--- a/roboco/api/routes/release.py
+++ b/roboco/api/routes/release.py
@@ -149,9 +149,15 @@ async def reject_release_proposal(
             status_code=status.HTTP_404_NOT_FOUND, detail="No open release proposal"
         )
     revised = await svc.reject(cast("UUID", task.id), data.required_changes)
-    if revised is None:  # pragma: no cover - open_proposal already guaranteed it
+    if revised is None:
+        # A concurrent approve is mid-execute (holds the release mutex) or
+        # Redis is unreachable — reject fails closed rather than racing it.
         raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND, detail="No open release proposal"
+            status_code=status.HTTP_409_CONFLICT,
+            detail=(
+                "Reject refused: a release approve is in progress or Redis is"
+                " unavailable. Retry once it clears."
+            ),
         )
     await db.commit()
     return _to_response(revised)
diff --git a/roboco/foundation/policy/lifecycle.py b/roboco/foundation/policy/lifecycle.py
index 376c5ecf..380c3bf0 100644
--- a/roboco/foundation/policy/lifecycle.py
+++ b/roboco/foundation/policy/lifecycle.py
@@ -245,6 +245,14 @@ _STATUS_TRANSITIONS: tuple[StatusTransition, ...] = (
         frozenset({Role.DOCUMENTER}),
     ),
     StatusTransition(Status.NEEDS_REVISION, Status.CLAIMED, "claim", None),
+    # A PM re-claims an awaiting_pm_review task it already owns (CLAIM_RULES
+    # grants this to CELL_PM/MAIN_PM) — see the "claim" ActionSpec comment.
+    StatusTransition(
+        Status.AWAITING_PM_REVIEW,
+        Status.CLAIMED,
+        "claim",
+        frozenset({Role.CELL_PM, Role.MAIN_PM}),
+    ),
     # Start
     StatusTransition(Status.CLAIMED, Status.IN_PROGRESS, "start", None),
     # Block / pause / resume
@@ -446,6 +454,14 @@ _ATOMIC_ACTIONS: dict[str, ActionSpec] = {
                 Status.AWAITING_QA,
                 Status.AWAITING_DOCUMENTATION,
                 Status.AWAITING_PR_REVIEW,
+                # A PM re-claims a review/queue task it already owns (e.g. after
+                # a respawn) via i_will_plan — CLAIM_RULES[CELL_PM/MAIN_PM]
+                # grants AWAITING_PM_REVIEW. Without it here, can_invoke_intent
+                # rejected i_will_plan on an awaiting_pm_review task with
+                # invalid_state even though CLAIM_RULES said it was allowed.
+                # test_claim_rules_match_pre_gateway_table keeps this source set
+                # and CLAIM_RULES in sync.
+                Status.AWAITING_PM_REVIEW,
             }
         ),
         target_status=Status.CLAIMED,
@@ -732,8 +748,22 @@ CLAIM_RULES: dict[Role, frozenset[Status]] = {
     # that scopes a developer's leaf-revision: give_me_work only ever offers an
     # agent its own assigned tasks. (A per-instance ownership gate at the gateway
     # would diverge from this spec — the parity invariant forbids that.)
-    Role.CELL_PM: frozenset({Status.PENDING, Status.NEEDS_REVISION}),
-    Role.MAIN_PM: frozenset({Status.PENDING, Status.NEEDS_REVISION}),
+    #
+    # AWAITING_PM_REVIEW: a PM re-claims its own review-queue task (e.g. after
+    # a respawn) via i_will_plan. roboco/services/task.py's
+    # _ROLE_CLAIM_STATUSES already granted this to "cell_pm"/"main_pm" on the
+    # stated belief that "the spec (lifecycle.CLAIM_RULES) grants it" — but
+    # CLAIM_RULES never actually did, so can_invoke_intent silently rejected
+    # every i_will_plan attempt on an awaiting_pm_review task with
+    # invalid_state regardless of what the service layer allowed. Added here
+    # to match; test_claim_rules_match_pre_gateway_table asserts CLAIM_RULES
+    # and the service table stay in sync so they can't diverge again.
+    Role.CELL_PM: frozenset(
+        {Status.PENDING, Status.NEEDS_REVISION, Status.AWAITING_PM_REVIEW}
+    ),
+    Role.MAIN_PM: frozenset(
+        {Status.PENDING, Status.NEEDS_REVISION, Status.AWAITING_PM_REVIEW}
+    ),
     Role.PRODUCT_OWNER: frozenset(),
     Role.HEAD_MARKETING: frozenset(),
     Role.AUDITOR: frozenset(),
diff --git a/roboco/services/release_executor.py b/roboco/services/release_executor.py
index 2b51ef71..bce58c0b 100644
--- a/roboco/services/release_executor.py
+++ b/roboco/services/release_executor.py
@@ -254,6 +254,15 @@ async def _await_proc(
             proc.kill()  # already-exited between the timeout and the kill is fine
         await proc.wait()
         return _TIMEOUT_RC, f"subprocess timed out after {int(timeout)}s"
+    except asyncio.CancelledError:
+        # An outer cancellation (e.g. the release loop's own task being
+        # cancelled mid-op) throws in here instead of the TimeoutError above —
+        # same orphaned child + leaked FDs if left unkilled. Mirrors
+        # quality_gate.py's ``_run_one`` handler for the identical shape.
+        with contextlib.suppress(ProcessLookupError):
+            proc.kill()
+        await proc.wait()
+        raise
     return proc.returncode or 0, out.decode("utf-8", "replace")
 
 
diff --git a/roboco/services/release_proposal.py b/roboco/services/release_proposal.py
index ecc10ff7..a18f1852 100644
--- a/roboco/services/release_proposal.py
+++ b/roboco/services/release_proposal.py
@@ -439,6 +439,14 @@ class ReleaseProposalService(BaseService):
         published (COMPLETED) — an approve may have shipped it after a stale
         reject button/request was queued; cancelling a published release
         would lie about the release's real, already-public state.
+
+        Acquires the same release mutex ``approve()`` holds (same key, same
+        non-blocking acquire style) so a reject can't interleave with a
+        concurrent in-flight approve — an unguarded write here used to be
+        able to land on the proposal while an approve was mid-execute (up to
+        ~40 min), racing the approve's own post-lock write. Fails CLOSED like
+        approve, both when the lock is held and when Redis is unreachable —
+        the CEO retries the reject once it clears.
         """
         task = await get_task_service(self.session).get(task_id)
         if task is None or task.source != RELEASE_MANAGER_SOURCE:
@@ -448,10 +456,34 @@ class ReleaseProposalService(BaseService):
                 f"release proposal {task_id} already published (COMPLETED);"
                 " cannot be rejected"
             )
-        markers.set_release_required_changes(task, required_changes)
-        task.status = TaskStatus.CANCELLED
-        await self.session.flush()
-        return task
+
+        lock_key = f"{_RELEASE_LOCK_PREFIX}{task_id}"
+        try:
+            lock_token = await self._acquire_release_lock(lock_key)
+        except ReleaseLockUnavailable as exc:
+            logger.error("release reject lock unavailable (redis down): %s", exc)
+            return None
+        if lock_token is None:
+            return None  # a concurrent approve is mid-execute; refuse the reject
+        try:
+            # Re-read under the lock: a concurrent approve may have committed
+            # COMPLETED between the pre-lock check and here.
+            self.session.expire(task)
+            locked = await get_task_service(self.session).get(task_id)
+            if locked is None:
+                return None
+            if locked.status == TaskStatus.COMPLETED:
+                raise TaskAlreadyCompletedError(
+                    f"release proposal {task_id} already published (COMPLETED);"
+                    " cannot be rejected"
+                )
+            markers.set_release_required_changes(locked, required_changes)
+            locked.status = TaskStatus.CANCELLED
+            await self.session.flush()
+            return locked
+        finally:
+            await self._release_release_lock(lock_key, lock_token)
+            await self._close_redis()
 
 
 def get_release_proposal_service(session: AsyncSession) -> ReleaseProposalService:
diff --git a/roboco/services/sequencing.py b/roboco/services/sequencing.py
index 123538d6..75fc6fa6 100644
--- a/roboco/services/sequencing.py
+++ b/roboco/services/sequencing.py
@@ -282,6 +282,47 @@ def _same_assignee_lane_edges(siblings: list) -> list[tuple[object, object]]:
     return fallback
 
 
+def _reaches(precedes: dict, src: object, dst: object) -> bool:
+    """True if ``src`` can reach ``dst`` following precedence edges, where
+    ``precedes[a]`` is the set of nodes that must run after ``a``. Used to drop
+    a lane-fallback edge that would close a cycle against the analyzer edges."""
+    seen: set = set()
+    stack = [src]
+    while stack:
+        node = stack.pop()
+        if node == dst:
+            return True
+        for nxt in precedes.get(node, ()):
+            if nxt not in seen:
+                seen.add(nxt)
+                stack.append(nxt)
+    return False
+
+
+def _extend_acyclic(
+    edges: list[tuple[object, object]],
+    candidates: list[tuple[object, object]],
+) -> list[tuple[object, object]]:
+    """Append ``(depends_on, task)`` candidate edges to ``edges``, skipping any
+    that duplicate an already-ordered pair (either direction) or would close a
+    cycle. ``edges`` are treated as authoritative and internally acyclic, so
+    the returned list is always acyclic — safe for ``TaskService.add_dependency``.
+    """
+    result = list(edges)
+    covered = {frozenset((a, b)) for a, b in edges}
+    precedes: dict[object, set] = defaultdict(set)
+    for dep_on, task in edges:
+        precedes[dep_on].add(task)
+    for dep_on, task in candidates:
+        pair = frozenset((dep_on, task))
+        if pair in covered or _reaches(precedes, task, dep_on):
+            continue
+        result.append((dep_on, task))
+        precedes[dep_on].add(task)
+        covered.add(pair)
+    return result
+
+
 def dev_task_collision_edges(siblings: list) -> list[tuple[object, object]]:
     """Wire the dev-task collision DAG for a parent's surfaced siblings.
 
@@ -306,6 +347,18 @@ def dev_task_collision_edges(siblings: list) -> list[tuple[object, object]]:
     waves are edge-consistent, so an edge-wired pair can never invert its
     sort). ``add_dependency`` dedupes, so repeated wiring is a no-op on
     already-wired pairs.
+
+    The same-assignee-lane fallback (see ``_same_assignee_lane_edges``) always
+    runs alongside the collision edges above, not only when the analyzer found
+    none: a collision between ONE pair of surfaced siblings must not silently
+    drop the lane-ordering of every OTHER same-assignee pair the analyzer never
+    saw (an unsurfaced sibling contributes no analyzer edge at all — it isn't
+    even in ``surfaced``). The analyzer edges are authoritative: a fallback edge
+    is kept only when it can't close a cycle against the edges already accepted
+    — both the direct same-pair conflict and the transitive one (an analyzer
+    edge that inverts priority order, plus a fallback chain through an
+    unsurfaced middle sibling that contradicts it), so the returned DAG is
+    always acyclic and never poisons ``add_dependency``.
     """
     # Collision edges from DECLARED surfaces. Fewer than two surfaced siblings
     # -> no collision path (edges stays empty); the undeclared-surface fallback
@@ -337,13 +390,14 @@ def dev_task_collision_edges(siblings: list) -> list[tuple[object, object]]:
         # any warning attributable. Empty capacity -> no warnings emitted.
         plan = SequencingService().analyze(surfaces, lambda _idx: "", {})
         edges = [(surfaced[a].id, surfaced[b].id) for a, b in plan.edges]
-    if edges:
-        return edges
 
-    # Undeclared-surface fallback (zero collision edges): chain same-assignee
-    # same-repo lanes so they don't merge out-of-order. See
-    # ``_same_assignee_lane_edges`` for the rationale + re-run idempotency.
-    return _same_assignee_lane_edges(siblings)
+    # Same-assignee-lane fallback: runs over EVERY sibling (surfaced or not),
+    # regardless of whether the collision analyzer above produced edges
+    # elsewhere in the batch. Analyzer edges are authoritative (and internally
+    # acyclic); extend them with the fallback, dropping any fallback edge that
+    # duplicates an already-ordered pair or would close a cycle (see
+    # _extend_acyclic — the transitive-contradiction guard).
+    return _extend_acyclic(edges, _same_assignee_lane_edges(siblings))
 
 
 # ---------------------------------------------------------------------------
diff --git a/roboco/services/task.py b/roboco/services/task.py
index 49c65ce7..dae22643 100644
--- a/roboco/services/task.py
+++ b/roboco/services/task.py
@@ -5484,6 +5484,17 @@ class TaskService(BaseService):
             )
             owning_pm = await self._resolve_pm_for_review(task)
             task.assigned_to = cast("Any", owning_pm) if owning_pm else None
+            # The documenter's claim ends here — unlike submit_for_qa/pass_qa/
+            # fail_qa (which all clear claimed_by + active_claimant_id on their
+            # own review-queue entry), this path pre-assigns a SPECIFIC owning
+            # PM rather than leaving assigned_to null. Without clearing the
+            # stale documenter claimant_id too, `_active_claim_violation`
+            # (content_actions.py) wrongly refuses the newly-assigned PM's own
+            # note()/commit() calls against this task_id before it formally
+            # claims — `assigned_to == agent_id` routes into the active-claim
+            # check, which still sees the documenter as claimant.
+            task.claimed_by = cast("Any", owning_pm) if owning_pm else None
+            task.active_claimant_id = cast("Any", owning_pm) if owning_pm else None
             self.log.info(
                 "Documentation complete, awaiting PM review",
                 task_id=str(task_id),
diff --git a/roboco/services/video_engine.py b/roboco/services/video_engine.py
index 272d9ee3..57e4f194 100644
--- a/roboco/services/video_engine.py
+++ b/roboco/services/video_engine.py
@@ -29,6 +29,7 @@ from roboco.foundation.policy.content import markers
 from roboco.models.base import Complexity, TaskNature, TaskStatus, TaskType, Team
 from roboco.services.base import BaseService
 from roboco.services.company_goals import get_company_goals_service
+from roboco.services.heartbeat_mutex import HeartbeatLockUnavailable, HeartbeatMutex
 from roboco.services.notification_delivery import get_notification_delivery_service
 from roboco.services.project import get_project_service
 from roboco.services.task import (
@@ -73,6 +74,19 @@ _CHAT_TIMEOUT_SECONDS = 60.0
 # pathological entry can't blow up the task description.
 _CHANGELOG_BRIEF_CHARS = 4000
 
+# ``open_video_task``'s "no open task for this occasion yet" check and its
+# insert are NOT atomic on their own: unlike the other engines' `run_cycle`
+# (each driven by exactly one sequential orchestrator-loop task, so it can
+# never overlap itself), this method is reachable from several genuinely
+# concurrent callers — the release-publish hook, the feature-spotlight hook,
+# and the CEO's on-demand ``POST /video/request`` route (two overlapping
+# requests, e.g. a double-click, each get their own DB session/transaction).
+# A flat SET NX (mirrors XPostService's identical short-critical-section
+# lock) keyed by occasion closes that window instead of letting a duplicate
+# authoring task slip through.
+_OCCASION_LOCK_PREFIX = "roboco:video_engine:occasion:"
+_OCCASION_LOCK_TTL_SECONDS = 60  # the check+insert completes in ms; crash backstop
+
 # Shared by every open_video_task caller (release/spotlight/on-demand) so the
 # authoring dev always lands on the demo kit instead of a text card.
 _MOTION_DESIGN_POINTER = (
@@ -357,9 +371,65 @@ class VideoEngine(BaseService):
         ``fallback_acceptance_criterion`` (when supplied) is appended instead
         — ``reauthor_from_rejection`` uses this for its
         feedback-must-be-addressed criterion.
+
+        The dedup check + insert below run under a short-lived Redis mutex
+        keyed by ``occasion`` (see ``_OCCASION_LOCK_PREFIX``) — this method,
+        unlike the other engines' single-loop ``run_cycle``, is reachable
+        from several genuinely concurrent callers (the release/spotlight
+        hooks and the on-demand ``/video/request`` route), so the "no open
+        task yet" read and the create must be atomic against each other.
+        Fails closed: a Redis outage or a lock already held both no-op
+        (return None) rather than risk a duplicate authoring task.
         """
         if not settings.video_engine_enabled:
             return None
+        lock = HeartbeatMutex(
+            f"{_OCCASION_LOCK_PREFIX}{occasion}",
+            ttl_seconds=_OCCASION_LOCK_TTL_SECONDS,
+            heartbeat_seconds=_OCCASION_LOCK_TTL_SECONDS,
+        )
+        try:
+            token = await lock.acquire()
+        except HeartbeatLockUnavailable as exc:
+            self.log.warning(
+                "video-engine: occasion lock unavailable (redis down); "
+                "not opening authoring task",
+                occasion=occasion,
+                error=str(exc),
+            )
+            return None
+        if token is None:
+            self.log.info(
+                "video-engine: another call is already opening this occasion; skipping",
+                occasion=occasion,
+            )
+            return None
+        try:
+            return await self._open_video_task_locked(
+                occasion=occasion,
+                script=script,
+                platforms=platforms,
+                brief=brief,
+                suggested_input_props=suggested_input_props,
+                project_id=project_id,
+                fallback_acceptance_criterion=fallback_acceptance_criterion,
+            )
+        finally:
+            await lock.release(token)
+
+    async def _open_video_task_locked(
+        self,
+        *,
+        occasion: str,
+        script: str,
+        platforms: list[str],
+        brief: str,
+        suggested_input_props: dict[str, Any] | None,
+        project_id: UUID | None,
+        fallback_acceptance_criterion: str | None,
+    ) -> TaskTable | None:
+        """The dedup-check + insert body, run while ``open_video_task`` holds
+        the per-occasion lock."""
         task_svc = get_task_service(self.session)
         open_tasks = await task_svc.list_open_video_posts()
         for existing in open_tasks:
diff --git a/roboco/services/x_post_service.py b/roboco/services/x_post_service.py
index 5762fc07..20ab0c5d 100644
--- a/roboco/services/x_post_service.py
+++ b/roboco/services/x_post_service.py
@@ -253,7 +253,17 @@ class XPostService(BaseService):
             logger.warning("spotlight video draft failed (best-effort): %s", exc)
 
     async def reject(self, task_id: UUID, reason: str) -> TaskTable | None:
-        """Record the CEO's reason and cancel the draft (never posted)."""
+        """Record the CEO's reason and cancel the draft (never posted).
+
+        Acquires the same post-mutex ``approve()`` holds (same key, same
+        non-blocking acquire style) so a reject can't interleave with a
+        concurrent in-flight approve — an unguarded write here used to race a
+        locked approve() straight to CANCELLED even while the approve was
+        mid-post, and depending on commit ordering could clobber the
+        approve's own COMPLETED write right after it landed. Fails CLOSED
+        like approve, both when the lock is held (approve mid-post) and when
+        Redis is unreachable — the CEO retries the reject once it clears.
+        """
         task = await get_task_service(self.session).get(task_id)
         if task is None or task.source not in X_SOURCES:
             return None
@@ -261,10 +271,32 @@ class XPostService(BaseService):
             raise TaskAlreadyCompletedError(
                 f"X draft {task_id} already posted (COMPLETED); cannot be rejected"
             )
-        markers.set_x_reject_reason(task, reason)
-        task.status = TaskStatus.CANCELLED
-        await self.session.flush()
-        return task
+
+        lock_key = f"{_LOCK_PREFIX}{task_id}"
+        try:
+            token = await self._acquire_lock(lock_key)
+        except _LockUnavailable as exc:
+            logger.error("x-post reject lock unavailable (redis down): %s", exc)
+            return None
+        if token is None:
+            return None  # a concurrent approve is mid-post; refuse the reject
+        try:
+            # Re-read under the lock: a concurrent approve may have posted +
+            # committed COMPLETED between the pre-lock check and here.
+            self.session.expire(task)
+            locked = await get_task_service(self.session).get(task_id)
+            if locked is None:
+                return None
+            if locked.status == TaskStatus.COMPLETED:
+                raise TaskAlreadyCompletedError(
+                    f"X draft {task_id} already posted (COMPLETED); cannot be rejected"
+                )
+            markers.set_x_reject_reason(locked, reason)
+            locked.status = TaskStatus.CANCELLED
+            await self.session.flush()
+            return locked
+        finally:
+            await self._release_lock(lock_key, token)
 
     # ---- Redis single-flight lock (plain SET NX — no heartbeat needed) -----
 
diff --git a/tests/foundation/test_lifecycle_spec.py b/tests/foundation/test_lifecycle_spec.py
index eff2c3c7..238a7ce9 100644
--- a/tests/foundation/test_lifecycle_spec.py
+++ b/tests/foundation/test_lifecycle_spec.py
@@ -500,10 +500,18 @@ def test_claim_rules_match_pre_gateway_table() -> None:
         {spec.Status.PENDING, spec.Status.AWAITING_DOCUMENTATION}
     )
     assert spec.CLAIM_RULES[spec.Role.CELL_PM] == frozenset(
-        {spec.Status.PENDING, spec.Status.NEEDS_REVISION}
+        {
+            spec.Status.PENDING,
+            spec.Status.NEEDS_REVISION,
+            spec.Status.AWAITING_PM_REVIEW,
+        }
     )
     assert spec.CLAIM_RULES[spec.Role.MAIN_PM] == frozenset(
-        {spec.Status.PENDING, spec.Status.NEEDS_REVISION}
+        {
+            spec.Status.PENDING,
+            spec.Status.NEEDS_REVISION,
+            spec.Status.AWAITING_PM_REVIEW,
+        }
     )
 
 
diff --git a/tests/integration/test_release_routes.py b/tests/integration/test_release_routes.py
index b7866706..90c30308 100644
--- a/tests/integration/test_release_routes.py
+++ b/tests/integration/test_release_routes.py
@@ -368,10 +368,22 @@ async def test_reject_records_changes_and_cancels_frees_dedup(
     frees and the release manager can re-assess next cycle. The required-changes
     marker stays on the cancelled row for history."""
     task = await _seed_proposal(db_session)
-    resp = await ceo_client.post(
-        "/api/release/proposal/reject",
-        json={"required_changes": "Tighten the CHANGELOG wording for the API change."},
-    )
+    with (
+        patch.object(
+            ReleaseProposalService, "_acquire_release_lock", AsyncMock(return_value="t")
+        ),
+        patch.object(
+            ReleaseProposalService,
+            "_release_release_lock",
+            AsyncMock(return_value=None),
+        ),
+    ):
+        resp = await ceo_client.post(
+            "/api/release/proposal/reject",
+            json={
+                "required_changes": "Tighten the CHANGELOG wording for the API change."
+            },
+        )
     assert resp.status_code == HTTPStatus.OK
     assert "Tighten the CHANGELOG" in (resp.json()["required_changes"] or "")
     refreshed = await db_session.get(TaskTable, task.id)
@@ -382,6 +394,24 @@ async def test_reject_records_changes_and_cancels_frees_dedup(
     assert task.id not in {t.id for t in open_proposals}
 
 
+@pytest.mark.asyncio
+async def test_reject_refused_while_approve_lock_held(
+    db_session: AsyncSession, ceo_client: AsyncClient
+) -> None:
+    """A concurrent approve holds the release mutex (mid ~40min execute);
+    reject must fail closed with 409 instead of racing an unguarded write
+    under it — previously reject() never even attempted the lock."""
+    await _seed_proposal(db_session)
+    with patch.object(
+        ReleaseProposalService, "_acquire_release_lock", AsyncMock(return_value=None)
+    ):
+        resp = await ceo_client.post(
+            "/api/release/proposal/reject",
+            json={"required_changes": "Tighten the CHANGELOG wording."},
+        )
+    assert resp.status_code == HTTPStatus.CONFLICT
+
+
 @pytest.mark.asyncio
 async def test_non_ceo_is_forbidden(db_session: AsyncSession) -> None:
     await _seed_proposal(db_session)
diff --git a/tests/integration/test_task_service_transitions.py b/tests/integration/test_task_service_transitions.py
index 21c0c779..0f6ec477 100644
--- a/tests/integration/test_task_service_transitions.py
+++ b/tests/integration/test_task_service_transitions.py
@@ -2112,6 +2112,59 @@ async def test_docs_complete_advances_when_pr_already_created(
     assert out.status == TaskStatus.AWAITING_PM_REVIEW
 
 
+@pytest.mark.asyncio
+async def test_docs_complete_advance_clears_stale_documenter_claim(
+    task_setup: dict, db_session: AsyncSession
+) -> None:
+    """F-audit: docs_complete's PM hand-off must not leave the outgoing
+    documenter as claimed_by/active_claimant_id once a specific owning PM is
+    assigned — unlike its siblings (qa_pass/fail_qa/submit_for_qa/pr_pass/
+    pr_fail/request_changes), which always reassign or clear both fields
+    together, _maybe_advance_to_pm_review used to only update assigned_to.
+    A stale active_claimant_id then makes content_actions.py's
+    `_active_claim_violation` wrongly reject the newly-assigned PM's own
+    explicit-task_id note()/commit() calls before it formally claims.
+    """
+    svc = task_setup["svc"]
+    pm_agent = AgentTable(
+        id=uuid4(),
+        name="PM",
+        slug=f"be-pm-{uuid4().hex[:8]}",
+        role=AgentRole.CELL_PM,
+        team=Team.BACKEND,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt="pm",
+        capabilities=[],
+        permissions={},
+        metrics={},
+    )
+    db_session.add(pm_agent)
+    await db_session.flush()
+    parent = await svc.create(_req(task_setup, assigned_to=pm_agent.id))
+    task = await svc.create(_req(task_setup, parent_task_id=parent.id))
+    task.status = TaskStatus.AWAITING_DOCUMENTATION
+    documenter_id = task_setup["agent_id"]
+    task.assigned_to = documenter_id
+    task.claimed_by = documenter_id
+    task.active_claimant_id = documenter_id
+    task.pr_number = 1
+    task.pr_url = "u"
+    task.pr_created = True
+    await db_session.flush()
+
+    out = await svc.docs_complete(task.id, doc_notes="documented all flows")
+
+    assert out is not None
+    assert out.status == TaskStatus.AWAITING_PM_REVIEW
+    assert out.assigned_to == pm_agent.id
+    # The documenter's claim must not survive the hand-off.
+    assert out.claimed_by == pm_agent.id
+    assert out.active_claimant_id == pm_agent.id
+    assert out.claimed_by != documenter_id
+    assert out.active_claimant_id != documenter_id
+
+
 # ---------------------------------------------------------------------------
 # mark_pr_created edge cases
 # ---------------------------------------------------------------------------
diff --git a/tests/integration/test_video_routes.py b/tests/integration/test_video_routes.py
index a75a7751..e9865078 100644
--- a/tests/integration/test_video_routes.py
+++ b/tests/integration/test_video_routes.py
@@ -265,15 +265,16 @@ async def test_request_video_opens_authoring_task(
     project = (
         await db_session.execute(select(ProjectTable).where(ProjectTable.slug == SLUG))
     ).scalar_one()
-    resp = await ceo_client.post(
-        "/api/video/request",
-        json={
-            "occasion": "CEO on-demand: launch teaser",
-            "brief": "A short teaser for the new dashboard",
-            "platforms": ["x", "tiktok"],
-            "project_id": str(project.id),
-        },
-    )
+    with _LOCKED[0], _LOCKED[1]:
+        resp = await ceo_client.post(
+            "/api/video/request",
+            json={
+                "occasion": "CEO on-demand: launch teaser",
+                "brief": "A short teaser for the new dashboard",
+                "platforms": ["x", "tiktok"],
+                "project_id": str(project.id),
+            },
+        )
     assert resp.status_code == HTTPStatus.OK
     body = resp.json()
     assert body["status"] == "opened"
@@ -396,12 +397,14 @@ async def test_request_video_not_opened_on_duplicate_occasion(
         "platforms": ["x"],
         "project_id": str(project.id),
     }
-    first = await ceo_client.post("/api/video/request", json=payload)
+    with _LOCKED[0], _LOCKED[1]:
+        first = await ceo_client.post("/api/video/request", json=payload)
     assert first.status_code == HTTPStatus.OK
     assert first.json()["status"] == "opened"
     task_id = first.json()["task_id"]
     try:
-        second = await ceo_client.post("/api/video/request", json=payload)
+        with _LOCKED[0], _LOCKED[1]:
+            second = await ceo_client.post("/api/video/request", json=payload)
         assert second.status_code == HTTPStatus.OK
         assert second.json()["status"] == "not_opened"
         assert second.json()["task_id"] is None
diff --git a/tests/integration/test_x_routes.py b/tests/integration/test_x_routes.py
index 47386bbc..5c397422 100644
--- a/tests/integration/test_x_routes.py
+++ b/tests/integration/test_x_routes.py
@@ -189,9 +189,13 @@ 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/x/posts/{task.id}/reject", json={"reason": "Not our voice"}
-    )
+    with (
+        patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
+        patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
+    ):
+        resp = await ceo_client.post(
+            f"/api/x/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)
@@ -204,9 +208,13 @@ async def test_history_returns_posted_and_rejected_newest_first(
     db_session: AsyncSession, ceo_client: AsyncClient
 ) -> None:
     rejected = await _seed_draft(db_session)
-    await ceo_client.post(
-        f"/api/x/posts/{rejected.id}/reject", json={"reason": "off-brand tone"}
-    )
+    with (
+        patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
+        patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
+    ):
+        await ceo_client.post(
+            f"/api/x/posts/{rejected.id}/reject", json={"reason": "off-brand tone"}
+        )
     posted = await _seed_draft(db_session)
     posted_project = await db_session.get(ProjectTable, posted.project_id)
     with (
@@ -258,9 +266,13 @@ async def test_history_respects_limit(
 ) -> None:
     for _ in range(3):
         t = await _seed_draft(db_session)
-        await ceo_client.post(
-            f"/api/x/posts/{t.id}/reject", json={"reason": "not relevant"}
-        )
+        with (
+            patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
+            patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
+        ):
+            await ceo_client.post(
+                f"/api/x/posts/{t.id}/reject", json={"reason": "not relevant"}
+            )
     resp = await ceo_client.get("/api/x/posts/history", params={"limit": HISTORY_LIMIT})
     assert resp.status_code == HTTPStatus.OK
     assert len(resp.json()) == HISTORY_LIMIT
diff --git a/tests/unit/runtime/test_video_render_loop.py b/tests/unit/runtime/test_video_render_loop.py
index b19d5837..7f215fca 100644
--- a/tests/unit/runtime/test_video_render_loop.py
+++ b/tests/unit/runtime/test_video_render_loop.py
@@ -28,6 +28,7 @@ from roboco.runtime.orchestrator import (
     _MAX_VIDEO_RENDER_ATTEMPTS,
     AgentOrchestrator,
 )
+from roboco.services import video_engine as video_engine_module
 from roboco.services.task import VIDEO_POST_SOURCE, get_task_service
 from roboco.services.video_engine import VideoEngine
 from sqlalchemy import select
@@ -49,6 +50,26 @@ def _orch() -> Any:
     return AgentOrchestrator.__new__(AgentOrchestrator)
 
 
+class _AlwaysAcquiredMutex:
+    """Stand-in for ``HeartbeatMutex``: always acquires immediately, no live
+    Redis required (matches the project's ``_no_live_redis`` fixture and
+    mirrors ``test_video_engine.py``'s identical stub)."""
+
+    def __init__(self, *_args: object, **_kwargs: object) -> None:
+        pass
+
+    async def acquire(self) -> str | None:
+        return "tok"
+
+    async def release(self, _token: str) -> None:
+        return None
+
+
+@pytest.fixture(autouse=True)
+def _stub_occasion_lock(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(video_engine_module, "HeartbeatMutex", _AlwaysAcquiredMutex)
+
+
 class _FakeRenderer:
     """Records every render() call; returns a deterministic path or raises."""
 
diff --git a/tests/unit/services/test_release_executor.py b/tests/unit/services/test_release_executor.py
index 7d5e5d08..59e9e039 100644
--- a/tests/unit/services/test_release_executor.py
+++ b/tests/unit/services/test_release_executor.py
@@ -9,8 +9,9 @@ call sequence; the production git/gh ops is exercised live (CEO-gated).
 from __future__ import annotations
 
 import base64
+import subprocess
 from types import SimpleNamespace
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any, cast
 from unittest.mock import MagicMock
 
 import pytest
@@ -444,6 +445,108 @@ def test_release_ci_workflow_decoupled_from_self_heal_setting(
     assert _resolve_release_ci_workflow() == "ci.yml"
 
 
+# --------------------------------------------------------------------------- #
+# _GitReleaseOps.release_commit_sha — half-landed retry detection against a
+# REAL git repo (not the fake ops above): verifies the exact worry the task
+# named — that ``_current_version`` could return the bumped version while the
+# changelog hasn't actually been written yet. It can't: ``apply_version_bumps``
+# and ``write_changelog_entry`` both run as uncommitted working-tree edits
+# BEFORE ``commit_and_push``'s single ``git add -A`` + commit, so nothing ever
+# reaches origin (and therefore no fresh retry clone can ever observe it) with
+# one written but not the other.
+# --------------------------------------------------------------------------- #
+
+
+def _git_sync(repo: Path, *args: str) -> str:
+    result = subprocess.run(
+        ["git", "-C", str(repo), *args], check=True, capture_output=True, text=True
+    )
+    return result.stdout
+
+
+def _init_release_repo(repo: Path, *, initial_version: str = "0.12.0") -> None:
+    repo.mkdir(parents=True, exist_ok=True)
+    _git_sync(repo, "init", "-b", "master")
+    _git_sync(repo, "config", "user.email", "t@example.com")
+    _git_sync(repo, "config", "user.name", "T")
+    _git_sync(repo, "config", "commit.gpgsign", "false")
+    (repo / "pyproject.toml").write_text(f'[project]\nversion = "{initial_version}"\n')
+    (repo / "CHANGELOG.md").write_text("# Changelog\n")
+    _git_sync(repo, "add", "-A")
+    _git_sync(repo, "commit", "-m", "init")
+
+
+def _ops(session: object, root: Path) -> _GitReleaseOps:
+    ctx = _ReleaseContext(
+        slug="roboco-api",
+        prod_branch="master",
+        root=root,
+        git_url="x",
+        git_prefix=[],
+        ci_workflow="ci.yml",
+        env_chain=[],
+    )
+    return _GitReleaseOps(session=cast("Any", session), ctx=ctx)
+
+
+@pytest.mark.asyncio
+async def test_release_commit_sha_detects_a_real_half_landed_commit(
+    tmp_path: Path,
+) -> None:
+    """A genuine publish_failed retry: a prior execute already ran
+    ``apply_version_bumps`` + ``write_changelog_entry`` + committed (both
+    landed in the SAME commit, exactly as ``commit_and_push`` does with one
+    ``git add -A``). The fresh clone must detect that commit and its
+    changelog entry must already be present — never re-bump, never skip the
+    changelog."""
+    _init_release_repo(tmp_path)
+    ops = _ops(MagicMock(), tmp_path)
+    await ops.apply_version_bumps(["pyproject.toml"], "0.13.0")
+    await ops.write_changelog_entry(
+        "## [0.13.0] - 2026-07-21\n\n### Added\n- a thing\n"
+    )
+    _git_sync(tmp_path, "add", "-A")
+    _git_sync(tmp_path, "commit", "-m", "chore(release): 0.13.0")
+
+    sha = await ops.release_commit_sha("0.13.0")
+
+    assert sha is not None
+    assert sha == _git_sync(tmp_path, "rev-parse", "HEAD").strip()
+    # The changelog entry landed in the SAME commit as the bump — never split.
+    assert "0.13.0" in (tmp_path / "pyproject.toml").read_text()
+    assert "a thing" in (tmp_path / "CHANGELOG.md").read_text()
+
+
+@pytest.mark.asyncio
+async def test_release_commit_sha_none_for_uncommitted_bump_no_false_half_landed(
+    tmp_path: Path,
+) -> None:
+    """An uncommitted version bump (e.g. a crash between apply_version_bumps
+    and commit_and_push) must NOT be mistaken for a half-landed release —
+    only a matching COMMIT counts. A fresh retry clone starts from origin's
+    unchanged HEAD anyway (this isolates release_commit_sha's own check)."""
+    _init_release_repo(tmp_path)
+    ops = _ops(MagicMock(), tmp_path)
+    # Bump the working tree WITHOUT committing (write_changelog_entry never ran).
+    await ops.apply_version_bumps(["pyproject.toml"], "0.13.0")
+
+    sha = await ops.release_commit_sha("0.13.0")
+
+    assert sha is None  # falls through to the normal bump/changelog/gate/commit path
+
+
+@pytest.mark.asyncio
+async def test_release_commit_sha_none_when_version_not_yet_bumped(
+    tmp_path: Path,
+) -> None:
+    """No prior attempt at all: the clone is still at the old version, so
+    there is nothing half-landed to detect."""
+    _init_release_repo(tmp_path)
+    ops = _ops(MagicMock(), tmp_path)
+
+    assert await ops.release_commit_sha("0.13.0") is None
+
+
 # --------------------------------------------------------------------------- #
 # H11: the PAT must never appear in a git subprocess argv. The release clone
 # and the release push carry the token via ``-c http.extraheader=Authorization:
diff --git a/tests/unit/services/test_release_executor_subprocess_timeout.py b/tests/unit/services/test_release_executor_subprocess_timeout.py
index 8e144da5..66dfbf4e 100644
--- a/tests/unit/services/test_release_executor_subprocess_timeout.py
+++ b/tests/unit/services/test_release_executor_subprocess_timeout.py
@@ -10,9 +10,10 @@ under a second — never relying on real wall-clock timing of the defaults.
 from __future__ import annotations
 
 import asyncio
+import os
 from pathlib import Path
-from typing import TYPE_CHECKING, cast
-from unittest.mock import AsyncMock, MagicMock
+from typing import TYPE_CHECKING, Any, cast
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import httpx
 import pytest
@@ -337,6 +338,42 @@ async def test_clone_run_times_out_and_kills_proc(
     assert proc.killed
 
 
+@pytest.mark.asyncio
+async def test_await_proc_kills_child_on_outer_cancellation() -> None:
+    """Redis mutex pre-lock write audit sibling finding: an outer cancellation
+    (e.g. the release loop's own task being cancelled mid-op) throws
+    ``CancelledError`` into ``_await_proc``'s ``wait_for``, bypassing the
+    ``TimeoutError`` handler. Without a dedicated handler (which
+    ``quality_gate.py``'s ``_run_one`` already has) the child is orphaned and
+    keeps running past the cancelled release op. ``_await_proc`` must kill +
+    reap it and re-raise, mirroring the already-shipped ``quality_gate.py``
+    pattern exactly."""
+    real_create_subprocess_exec = asyncio.create_subprocess_exec
+    spawned: dict[str, asyncio.subprocess.Process] = {}
+
+    async def _capturing_create(*args: Any, **kwargs: Any) -> Any:
+        proc = await real_create_subprocess_exec(*args, **kwargs)
+        spawned["proc"] = proc
+        return proc
+
+    with patch(
+        "roboco.services.release_executor.asyncio.create_subprocess_exec",
+        _capturing_create,
+    ):
+        task = asyncio.ensure_future(_run(["sleep", "30"]))
+        while "proc" not in spawned:
+            await asyncio.sleep(0.01)
+        await asyncio.sleep(0.1)  # let the child actually exec
+        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)
+
+
 # ---------------------------------------------------------------------------
 # Regression: the happy path still returns the real rc + decoded output.
 # ---------------------------------------------------------------------------
diff --git a/tests/unit/services/test_release_proposal_status_guards.py b/tests/unit/services/test_release_proposal_status_guards.py
index 23b4ecfb..ceeb0c76 100644
--- a/tests/unit/services/test_release_proposal_status_guards.py
+++ b/tests/unit/services/test_release_proposal_status_guards.py
@@ -12,6 +12,7 @@ do for their own seams.
 
 from __future__ import annotations
 
+import contextlib
 from typing import TYPE_CHECKING, cast
 from unittest.mock import AsyncMock, patch
 from uuid import uuid4
@@ -19,6 +20,7 @@ from uuid import uuid4
 import pytest
 from roboco.db.tables import AgentTable, ProjectTable, TaskTable
 from roboco.foundation import identity as _foundation
+from roboco.foundation.policy.content import markers
 from roboco.models.base import AgentRole, AgentStatus, TaskNature, TaskStatus, TaskType
 from roboco.models.base import Team as T
 from roboco.services.release_proposal import (
@@ -26,13 +28,17 @@ from roboco.services.release_proposal import (
     TaskAlreadyCompletedError,
 )
 from roboco.services.release_readiness import ReleaseReadinessReport, report_to_dict
-from roboco.services.task import RELEASE_MANAGER_SOURCE
+from roboco.services.task import RELEASE_MANAGER_SOURCE, TaskService
+from sqlalchemy.ext.asyncio import (
+    AsyncEngine,
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
 
 if TYPE_CHECKING:
     from uuid import UUID
 
-    from sqlalchemy.ext.asyncio import AsyncSession
-
 _VERSION = "0.18.0"
 
 
@@ -170,3 +176,89 @@ async def test_reject_raises_when_already_published(db_session: AsyncSession) ->
         )
     await db_session.refresh(task)
     assert task.status == TaskStatus.COMPLETED  # untouched, never cancelled
+
+
+async def _fresh_session(url: str) -> tuple[AsyncSession, AsyncEngine]:
+    """A session on a brand-new engine/connection (caller disposes)."""
+    engine = create_async_engine(url, future=True)
+    factory = async_sessionmaker(
+        bind=engine, class_=AsyncSession, expire_on_commit=False
+    )
+    return factory(), engine
+
+
+async def _dispose(session: AsyncSession, engine: AsyncEngine) -> None:
+    with contextlib.suppress(Exception):
+        await session.rollback()
+    await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_reject_concurrent_approve_completes_during_lock_wait(
+    db_session: AsyncSession, _test_database_url: str
+) -> None:
+    """Redis mutex pre-lock write audit regression for ``reject()``: a
+    genuinely concurrent approve (a real second session/connection) publishes
+    + commits COMPLETED in the window between reject's pre-lock read and its
+    lock acquisition. The in-lock re-read must see that committed state and
+    refuse — the CANCELLED status write and required-changes marker must
+    never land on the just-published row, proving the fix holds across
+    sessions, not merely within one. Mirrors
+    ``test_x_post_service.test_reject_concurrent_approve_completes_during_lock_wait``.
+    """
+    task = await _seed_proposal(db_session)
+    task_id = cast("UUID", task.id)
+    await db_session.commit()
+
+    real_get = TaskService.get
+    injected = False
+
+    async def _get_then_inject_concurrent_publish(
+        self: TaskService, tid: UUID
+    ) -> TaskTable | None:
+        """Fires once, right after reject's pre-lock read — the exact window
+        between that read and reject's own (would-be) pre-lock write."""
+        nonlocal injected
+        result = await real_get(self, tid)
+        if not injected:
+            injected = True
+            other, other_engine = await _fresh_session(_test_database_url)
+            try:
+                other_task = await other.get(TaskTable, tid)
+                assert other_task is not None
+                other_task.status = TaskStatus.COMPLETED
+                await other.commit()
+            finally:
+                await _dispose(other, other_engine)
+        return result
+
+    with (
+        patch.object(TaskService, "get", _get_then_inject_concurrent_publish),
+        patch.object(
+            ReleaseProposalService,
+            "_acquire_release_lock",
+            AsyncMock(return_value="tok"),
+        ),
+        patch.object(
+            ReleaseProposalService,
+            "_release_release_lock",
+            AsyncMock(return_value=None),
+        ),
+        patch.object(
+            ReleaseProposalService, "_close_redis", AsyncMock(return_value=None)
+        ),
+        pytest.raises(TaskAlreadyCompletedError),
+    ):
+        await ReleaseProposalService(db_session).reject(
+            task_id, "needs another migration check"
+        )
+
+    fresh, fresh_engine = await _fresh_session(_test_database_url)
+    try:
+        final = await fresh.get(TaskTable, task_id)
+        assert final is not None
+        assert final.status == TaskStatus.COMPLETED
+        # The reject must never have landed on the just-published row.
+        assert markers.get_release_required_changes(final) is None
+    finally:
+        await _dispose(fresh, fresh_engine)
diff --git a/tests/unit/services/test_sequencing.py b/tests/unit/services/test_sequencing.py
index 0632e5a2..20a64fb2 100644
--- a/tests/unit/services/test_sequencing.py
+++ b/tests/unit/services/test_sequencing.py
@@ -83,6 +83,27 @@ def test_touches_shared_runs_last() -> None:
     assert plan.waves[-1] == [2]  # the shared task is the final wave
 
 
+def test_all_shared_batch_with_disjoint_surfaces_generates_no_edges() -> None:
+    # Verifying the claimed rule-3 property: when every draft in the batch
+    # touches_shared, ``_shared_last_edges`` skips every candidate pair (its
+    # inner loop continues on `other.touches_shared`), so it contributes no
+    # edges on its own. With disjoint file surfaces rule 1 (same-shared-status
+    # overlap) also contributes nothing, so the whole batch runs in one
+    # parallel wave — confirmed correct, no fix needed.
+    s = [
+        DraftSurface(0, 1, ["fe/app/a.tsx"], False, True),
+        DraftSurface(1, 1, ["fe/app/b.tsx"], False, True),
+        DraftSurface(2, 1, ["fe/app/c.tsx"], False, True),
+    ]
+    plan = SequencingService().analyze(s, _frontend, {"frontend": 3})
+    assert plan.edges == []
+    assert plan.waves == [[0, 1, 2]]
+    # And rule 3 in isolation truly contributes zero edges for an all-shared
+    # set, regardless of overlap — it is rule 1 (same-shared-status overlap),
+    # not rule 3, that would serialize two OVERLAPPING shared surfaces.
+    assert SequencingService()._shared_last_edges(s) == []
+
+
 def test_cycle_is_rejected() -> None:
     with pytest.raises(SequencingError):
         SequencingService()._toposort([(0, 1), (1, 0)], 2)
@@ -218,6 +239,27 @@ def _edge_set(pairs: list[tuple[object, object]]) -> set[tuple[object, object]]:
     return set(pairs)
 
 
+def _has_cycle(pairs: list[tuple[object, object]]) -> bool:
+    """True if the (depends_on, task) edge list contains a directed cycle."""
+    graph: dict[object, set[object]] = {}
+    for dep_on, task in pairs:
+        graph.setdefault(dep_on, set()).add(task)
+    visiting: set[object] = set()
+    done: set[object] = set()
+
+    def _visit(node: object) -> bool:
+        visiting.add(node)
+        for nxt in graph.get(node, ()):
+            if nxt in visiting or (nxt not in done and _visit(nxt)):
+                return True
+        visiting.discard(node)
+        done.add(node)
+        return False
+
+    nodes = {n for pair in pairs for n in pair}
+    return any(n not in done and _visit(n) for n in nodes)
+
+
 def test_dev_collision_disjoint_surfaces_are_parallel() -> None:
     # Same project, disjoint files → no edge (the two dev tasks run together).
     a, b = (
@@ -354,6 +396,74 @@ def test_dev_collision_fallback_idempotent_on_rerun() -> None:
     assert dev_task_collision_edges([a, b]) == dev_task_collision_edges([a, b])
 
 
+def test_dev_collision_fallback_still_applies_when_another_pair_collides() -> None:
+    # Regression: a `if edges: return edges` short-circuit used to drop the
+    # assignee-lane fallback ENTIRELY whenever ANY surfaced pair produced a
+    # collision edge, even for a totally unrelated same-assignee pair with no
+    # declared surface at all. (a, b) collide on a.py (different assignees, so
+    # no lane relationship between them); (c, d) share an assignee/project but
+    # declare no surface — they must still get lane-ordered.
+    a = _Sib(
+        uuid4(),
+        sequence=0,
+        intends_to_touch=["a.py"],
+        assigned_to="be-dev-1",
+    )
+    b = _Sib(
+        uuid4(),
+        sequence=1,
+        intends_to_touch=["a.py"],
+        assigned_to="be-dev-2",
+    )
+    c = _Sib(uuid4(), sequence=2, assigned_to="be-dev-3")
+    d = _Sib(uuid4(), sequence=3, assigned_to="be-dev-3")
+    edges = _edge_set(dev_task_collision_edges([a, b, c, d]))
+    assert edges == {(a.id, b.id), (c.id, d.id)}
+
+
+def test_dev_collision_fallback_covers_unsurfaced_sibling_in_surfaced_lane() -> None:
+    # Same assignee/project lane mixes a surfaced sibling (touches a.py) with
+    # an unsurfaced one (no declared surface) and a third surfaced sibling
+    # that doesn't overlap the first — the analyzer alone wires nothing for
+    # this lane (no pair overlaps), so the fallback must still chain all three
+    # by (priority, sequence).
+    first = _Sib(uuid4(), sequence=0, assigned_to="be-dev-1", intends_to_touch=["a.py"])
+    bare = _Sib(uuid4(), sequence=1, assigned_to="be-dev-1")
+    other = _Sib(uuid4(), sequence=2, assigned_to="be-dev-1", intends_to_touch=["b.py"])
+    edges = dev_task_collision_edges([first, bare, other])
+    assert edges == [(first.id, bare.id), (bare.id, other.id)]
+
+
+def test_dev_collision_fallback_never_closes_cycle_against_analyzer() -> None:
+    # Regression: the analyzer's shared-last migration order inverts priority
+    # order (s3 before s1), while the same-assignee lane fallback chains by
+    # priority through the unsurfaced middle sibling (s1 -> s2 -> s3). Naively
+    # unioning the two closed a 3-cycle s1 -> s3 -> s2 -> s1 that made
+    # add_dependency raise ConflictError and wedged every later delegate. The
+    # analyzer edge wins; the fallback edge that would cycle is dropped.
+    s1 = _Sib(
+        uuid4(),
+        priority=1,
+        sequence=0,
+        assigned_to="be-dev-1",
+        adds_migration=True,
+        touches_shared=True,
+    )
+    s2 = _Sib(uuid4(), priority=2, sequence=1, assigned_to="be-dev-1")  # unsurfaced
+    s3 = _Sib(
+        uuid4(),
+        priority=3,
+        sequence=2,
+        assigned_to="be-dev-1",
+        adds_migration=True,
+        touches_shared=False,
+    )
+    edges = dev_task_collision_edges([s1, s2, s3])
+    assert not _has_cycle(edges)
+    assert (s3.id, s1.id) in edges  # authoritative analyzer edge preserved
+    assert (s2.id, s3.id) not in edges  # the cycling fallback edge is dropped
+
+
 # ---------------------------------------------------------------------------
 # cell_task_wave_chain_depends_on — the cell-task wave chain (edge kind 2).
 # Pure glue: a new cell-task under root-subtask UT_n depends on every cell-task
diff --git a/tests/unit/services/test_video_engine.py b/tests/unit/services/test_video_engine.py
index 5dedc221..78e678c8 100644
--- a/tests/unit/services/test_video_engine.py
+++ b/tests/unit/services/test_video_engine.py
@@ -8,6 +8,7 @@ Secretary-owned and held for the CEO. Asserted against a real Postgres DB.
 
 from __future__ import annotations
 
+import asyncio
 from typing import TYPE_CHECKING, cast
 from unittest.mock import AsyncMock
 from uuid import UUID, uuid4
@@ -21,6 +22,7 @@ from roboco.models.base import AgentRole, AgentStatus, Complexity, Team
 from roboco.models.base import TaskStatus as TS
 from roboco.services import video_engine as video_engine_module
 from roboco.services.company_goals import get_company_goals_service
+from roboco.services.heartbeat_mutex import HeartbeatLockUnavailable
 from roboco.services.task import VIDEO_POST_SOURCE, VIDEO_SOURCE, get_task_service
 from sqlalchemy import delete, select
 
@@ -99,6 +101,27 @@ def _mock_local_model(monkeypatch: pytest.MonkeyPatch, reply: str | None) -> Asy
     return mock
 
 
+class _AlwaysAcquiredMutex:
+    """Stand-in for ``HeartbeatMutex``: always acquires immediately, no live
+    Redis required (matches the project's ``_no_live_redis`` fixture). Used
+    as the default so every scenario below that isn't specifically testing
+    the occasion-lock behavior is unaffected by its introduction."""
+
+    def __init__(self, *_args: object, **_kwargs: object) -> None:
+        pass
+
+    async def acquire(self) -> str | None:
+        return "tok"
+
+    async def release(self, _token: str) -> None:
+        return None
+
+
+@pytest.fixture(autouse=True)
+def _stub_occasion_lock(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(video_engine_module, "HeartbeatMutex", _AlwaysAcquiredMutex)
+
+
 # --------------------------------------------------------------------------- #
 # open_video_task
 # --------------------------------------------------------------------------- #
@@ -203,6 +226,109 @@ async def test_open_video_task_dedupes_same_occasion(
     assert len(open_tasks) == ONE
 
 
+@pytest.mark.asyncio
+async def test_open_video_task_returns_none_when_occasion_lock_held(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A held per-occasion lock (another call already owns it) is a no-op,
+    not an error — it opens nothing and leaves the shared session usable."""
+    await _seed(db_session)
+    _enable(monkeypatch)
+
+    class _HeldMutex:
+        def __init__(self, *_a: object, **_kw: object) -> None:
+            pass
+
+        async def acquire(self) -> str | None:
+            return None  # another call already holds this occasion's lock
+
+        async def release(self, _token: str) -> None:
+            raise AssertionError("release must not be called when acquire failed")
+
+    monkeypatch.setattr(video_engine_module, "HeartbeatMutex", _HeldMutex)
+    engine = video_engine_module.VideoEngine(db_session)
+    task = await engine.open_video_task(
+        occasion="release v1.0.0", script="s", platforms=["x"], brief="b"
+    )
+    assert task is None
+    assert await get_task_service(db_session).list_open_video_posts() == []
+
+
+@pytest.mark.asyncio
+async def test_open_video_task_returns_none_when_lock_unavailable(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A Redis outage on the occasion lock fails closed (no-op), not a crash."""
+    await _seed(db_session)
+    _enable(monkeypatch)
+
+    class _BrokenMutex:
+        def __init__(self, *_a: object, **_kw: object) -> None:
+            pass
+
+        async def acquire(self) -> str | None:
+            raise HeartbeatLockUnavailable("redis down")
+
+        async def release(self, _token: str) -> None:
+            raise AssertionError("release must not be called when acquire failed")
+
+    monkeypatch.setattr(video_engine_module, "HeartbeatMutex", _BrokenMutex)
+    engine = video_engine_module.VideoEngine(db_session)
+    task = await engine.open_video_task(
+        occasion="release v1.0.0", script="s", platforms=["x"], brief="b"
+    )
+    assert task is None
+    assert await get_task_service(db_session).list_open_video_posts() == []
+
+
+@pytest.mark.asyncio
+async def test_open_video_task_concurrent_same_occasion_creates_only_one(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Regression (engine dedup race): ``open_video_task`` is reachable from
+    several genuinely concurrent callers for the same occasion (a double-click
+    on ``/video/request``, or an on-demand call racing the release hook) —
+    unlike the other engines' single-loop ``run_cycle``, two overlapping calls
+    are NOT serialized by the orchestrator's own scheduling. A real
+    ``asyncio.Lock`` stands in for the Redis SET NX mutex's mutual exclusion
+    (no live Redis in tests): the first caller to reach the DB's genuinely
+    suspending await wins the lock and creates the task; the second finds it
+    held and returns None immediately — never both passing the dedup check."""
+    await _seed(db_session)
+    _enable(monkeypatch)
+    engine = video_engine_module.VideoEngine(db_session)
+
+    real_lock = asyncio.Lock()
+
+    class _RaceMutex:
+        def __init__(self, *_a: object, **_kw: object) -> None:
+            pass
+
+        async def acquire(self) -> str | None:
+            if real_lock.locked():
+                return None
+            await real_lock.acquire()
+            return "tok"
+
+        async def release(self, _token: str) -> None:
+            real_lock.release()
+
+    monkeypatch.setattr(video_engine_module, "HeartbeatMutex", _RaceMutex)
+
+    results = await asyncio.gather(
+        engine.open_video_task(
+            occasion="race me", script="s1", platforms=["x"], brief="b1"
+        ),
+        engine.open_video_task(
+            occasion="race me", script="s2", platforms=["x"], brief="b2"
+        ),
+    )
+    created = [r for r in results if r is not None]
+    assert len(created) == ONE
+    open_tasks = await get_task_service(db_session).list_open_video_posts()
+    assert len(open_tasks) == ONE
+
+
 @pytest.mark.asyncio
 async def test_open_video_task_respects_open_cap(
     db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
diff --git a/tests/unit/services/test_video_post_service.py b/tests/unit/services/test_video_post_service.py
index f7ee8b75..2b6baf24 100644
--- a/tests/unit/services/test_video_post_service.py
+++ b/tests/unit/services/test_video_post_service.py
@@ -274,7 +274,7 @@ async def test_approve_completes_when_unconfigured_platform_is_skipped(
     assert result.status == "posted"
     assert result.posted == {"x": "x-vid-1"}
     assert "skipped (unconfigured): tiktok" in result.detail
-    assert tiktok_poster.calls == []  # never attempted without credentials
+    assert tiktok_poster.calls == []
     await db_session.refresh(task)
     assert task.status == TS.COMPLETED
     draft = markers.get_video_draft(task)
@@ -967,7 +967,7 @@ async def test_approve_concurrent_caption_edit_does_not_erase_a_committed_posted
     commit — the retry re-posted the platform (a double-post)."""
     task = await _seed_video_post(db_session, platforms=["x", "tiktok"])
     task_id = _id(task)
-    await db_session.commit()  # externally visible to the "concurrent" session below
+    await db_session.commit()
 
     real_get = TaskService.get
     injected = False
diff --git a/tests/unit/services/test_x_post_service.py b/tests/unit/services/test_x_post_service.py
index cfc59012..e7092692 100644
--- a/tests/unit/services/test_x_post_service.py
+++ b/tests/unit/services/test_x_post_service.py
@@ -7,6 +7,8 @@ fixture) so approve exercises the real post + status-transition path.
 
 from __future__ import annotations
 
+import contextlib
+from contextlib import contextmanager
 from typing import TYPE_CHECKING, cast
 from unittest.mock import AsyncMock, patch
 from uuid import uuid4
@@ -25,7 +27,12 @@ from roboco.models.base import (
 from roboco.models.base import TaskNature as TN
 from roboco.models.base import TaskStatus as TS
 from roboco.models.base import TaskType as TT
-from roboco.services.task import X_FEATURE_SOURCE, X_POST_SOURCE, X_REPLY_SOURCE
+from roboco.services.task import (
+    X_FEATURE_SOURCE,
+    X_POST_SOURCE,
+    X_REPLY_SOURCE,
+    TaskService,
+)
 from roboco.services.x_client import XClient, XMention, XPostResult
 from roboco.services.x_post_service import (
     TaskAlreadyCompletedError,
@@ -34,18 +41,34 @@ from roboco.services.x_post_service import (
     XPostService,
     get_x_post_service,
 )
+from sqlalchemy.ext.asyncio import (
+    AsyncEngine,
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
 
 if TYPE_CHECKING:
+    from collections.abc import Iterator
     from uuid import UUID
 
-    from sqlalchemy.ext.asyncio import AsyncSession
-
 SYSTEM_UUID = _foundation.AGENTS["system"].uuid
 SECRETARY_UUID = _foundation.AGENTS["secretary-1"].uuid
 ONE = 1
 TWO = 2
 
 
+@contextmanager
+def _lock_free() -> Iterator[None]:
+    """Patch XPostService's lock helpers so approve/reject exercise the real
+    post/cancel path without touching the (test-blocked) Redis."""
+    with (
+        patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
+        patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
+    ):
+        yield
+
+
 class _StubClient(XClient):
     def __init__(self, *, posted: bool = True, tweet_id: str = "999") -> None:
         self._posted = posted
@@ -268,7 +291,7 @@ async def test_approve_no_credentials_result(db_session: AsyncSession) -> None:
     assert result is not None
     assert result.status == "no_credentials"
     await db_session.refresh(task)
-    assert task.status == TS.PENDING  # never advanced without credentials
+    assert task.status == TS.PENDING
 
 
 @pytest.mark.asyncio
@@ -326,7 +349,8 @@ async def test_approve_refuses_already_rejected_draft(
     refuses and never calls the X client — the reproduced bug (a stale
     Approve after reject re-posting)."""
     task = await _seed_draft(db_session)
-    await _svc(db_session).reject(_id(task), "not on-brand")
+    with _lock_free():
+        await _svc(db_session).reject(_id(task), "not on-brand")
     client = _StubClient()
     with (
         patch("roboco.services.x_post_service.build_x_client", return_value=client),
@@ -387,17 +411,39 @@ 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_draft(db_session, source=X_REPLY_SOURCE)
-    updated = await _svc(db_session).reject(_id(task), "Tone doesn't match our voice")
+    with _lock_free():
+        updated = await _svc(db_session).reject(
+            _id(task), "Tone doesn't match our voice"
+        )
     assert updated is not None
     assert updated.status == TS.CANCELLED
     assert markers.get_x_reject_reason(updated) == "Tone doesn't match our voice"
 
 
+@pytest.mark.asyncio
+async def test_reject_refused_while_lock_held_by_concurrent_approve(
+    db_session: AsyncSession,
+) -> None:
+    """A concurrent approve holds the post lock (mid-tweet-POST); reject must
+    fail closed instead of racing a CANCEL under it — previously reject()
+    never even attempted the lock, so it could commit CANCELLED to a draft a
+    concurrent approve was about to mark COMPLETED, or clobber the approve's
+    outcome depending on commit ordering."""
+    task = await _seed_draft(db_session)
+    with patch.object(XPostService, "_acquire_lock", AsyncMock(return_value=None)):
+        result = await _svc(db_session).reject(_id(task), "not relevant")
+    assert result is None
+    await db_session.refresh(task)
+    assert task.status == TS.PENDING
+    assert markers.get_x_reject_reason(task) is None
+
+
 @pytest.mark.asyncio
 async def test_list_open_posts_excludes_terminal(db_session: AsyncSession) -> None:
     open_task = await _seed_draft(db_session)
     rejected_task = await _seed_draft(db_session, source=X_REPLY_SOURCE)
-    await _svc(db_session).reject(_id(rejected_task), "not relevant")
+    with _lock_free():
+        await _svc(db_session).reject(_id(rejected_task), "not relevant")
     open_posts = await _svc(db_session).list_open_posts()
     ids = {t.id for t in open_posts}
     assert open_task.id in ids
@@ -450,13 +496,73 @@ async def test_reject_completed_raises(db_session: AsyncSession) -> None:
         await _svc(db_session).reject(_id(task), "nope")
 
 
+@pytest.mark.asyncio
+async def test_reject_concurrent_approve_completes_during_lock_wait(
+    db_session: AsyncSession, _test_database_url: str
+) -> None:
+    """Redis mutex pre-lock write audit regression for ``reject()``: a
+    genuinely concurrent approve (a real second session/connection) posts +
+    commits COMPLETED in the window between reject's pre-lock read and its
+    lock acquisition. The in-lock re-read must see that committed state and
+    refuse — the CANCELLED status write and reject reason must never land on
+    the just-posted row, proving the fix holds across sessions, not merely
+    within one. Mirrors
+    ``test_approve_concurrent_edit_does_not_clobber_a_committed_post``."""
+    task = await _seed_draft(db_session)
+    task_id = _id(task)
+    await db_session.commit()
+
+    real_get = TaskService.get
+    injected = False
+
+    async def _get_then_inject_concurrent_post(
+        self: TaskService, tid: UUID
+    ) -> TaskTable | None:
+        """Fires once, right after reject's pre-lock read — the exact window
+        between that read and reject's own (would-be) pre-lock write."""
+        nonlocal injected
+        result = await real_get(self, tid)
+        if not injected:
+            injected = True
+            other, other_engine = await _fresh_session(_test_database_url)
+            try:
+                other_task = await other.get(TaskTable, tid)
+                assert other_task is not None
+                markers.set_x_posted_tweet_id(other_task, "concurrent-999")
+                other_task.status = TS.COMPLETED
+                await other.commit()
+            finally:
+                await _dispose(other, other_engine)
+        return result
+
+    with (
+        patch.object(TaskService, "get", _get_then_inject_concurrent_post),
+        patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
+        patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
+        pytest.raises(TaskAlreadyCompletedError),
+    ):
+        await _svc(db_session).reject(task_id, "Tone doesn't match")
+
+    fresh, fresh_engine = await _fresh_session(_test_database_url)
+    try:
+        final = await fresh.get(TaskTable, task_id)
+        assert final is not None
+        assert final.status == TS.COMPLETED
+        assert markers.get_x_posted_tweet_id(final) == "concurrent-999"
+        # The reject must never have landed on the just-posted row.
+        assert markers.get_x_reject_reason(final) is None
+    finally:
+        await _dispose(fresh, fresh_engine)
+
+
 @pytest.mark.asyncio
 async def test_list_post_history_excludes_open_drafts(
     db_session: AsyncSession,
 ) -> None:
     open_task = await _seed_draft(db_session)
     rejected_task = await _seed_draft(db_session, source=X_REPLY_SOURCE)
-    await _svc(db_session).reject(_id(rejected_task), "not relevant")
+    with _lock_free():
+        await _svc(db_session).reject(_id(rejected_task), "not relevant")
     history = await _svc(db_session).list_post_history()
     ids = {t.id for t in history}
     assert rejected_task.id in ids
@@ -468,7 +574,8 @@ async def test_list_post_history_newest_acted_first(
     db_session: AsyncSession,
 ) -> None:
     rejected_task = await _seed_draft(db_session, source=X_REPLY_SOURCE)
-    await _svc(db_session).reject(_id(rejected_task), "not relevant")
+    with _lock_free():
+        await _svc(db_session).reject(_id(rejected_task), "not relevant")
     posted_task = await _seed_draft(db_session)
     client = _StubClient()
     with (
@@ -495,7 +602,8 @@ async def test_list_post_history_includes_marker_fields(
     ):
         await _svc(db_session).approve(_id(posted_task))
     rejected_task = await _seed_draft(db_session, source=X_REPLY_SOURCE)
-    await _svc(db_session).reject(_id(rejected_task), "off-brand tone")
+    with _lock_free():
+        await _svc(db_session).reject(_id(rejected_task), "off-brand tone")
 
     history = await _svc(db_session).list_post_history()
     by_id = {t.id: t for t in history}
@@ -508,7 +616,8 @@ async def test_list_post_history_respects_limit(db_session: AsyncSession) -> Non
     tasks = []
     for _ in range(3):
         t = await _seed_draft(db_session, source=X_REPLY_SOURCE)
-        await _svc(db_session).reject(_id(t), "not relevant")
+        with _lock_free():
+            await _svc(db_session).reject(_id(t), "not relevant")
         tasks.append(t)
     history = await _svc(db_session).list_post_history(limit=2)
     assert len(history) == TWO
@@ -550,6 +659,86 @@ async def test_approve_does_not_flush_edited_body_before_lock(
     assert markers.get_x_draft_body(task) == original_body
 
 
+async def _fresh_session(url: str) -> tuple[AsyncSession, AsyncEngine]:
+    """A session on a brand-new engine/connection (caller disposes)."""
+    engine = create_async_engine(url, future=True)
+    factory = async_sessionmaker(
+        bind=engine, class_=AsyncSession, expire_on_commit=False
+    )
+    return factory(), engine
+
+
+async def _dispose(session: AsyncSession, engine: AsyncEngine) -> None:
+    with contextlib.suppress(Exception):
+        await session.rollback()
+    await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_approve_concurrent_edit_does_not_clobber_a_committed_post(
+    db_session: AsyncSession, _test_database_url: str
+) -> None:
+    """Redis mutex pre-lock write audit regression: a genuinely concurrent
+    approve (a real second session/connection, not an in-process mock) posts
+    + commits COMPLETED in the window between our pre-lock read and our lock
+    acquisition. The in-lock re-read must see that committed state and the
+    CEO's edited body must never land on the just-posted row — proving the
+    fix holds across sessions, not merely within one, mirroring
+    VideoPostService's identical cross-session regression test."""
+    task = await _seed_draft(db_session, body="Original")
+    task_id = _id(task)
+    await db_session.commit()
+
+    real_get = TaskService.get
+    injected = False
+
+    async def _get_then_inject_concurrent_post(
+        self: TaskService, tid: UUID
+    ) -> TaskTable | None:
+        """Fires once, right after the outer pre-lock read — the exact
+        window between our read and our own (would-be) pre-lock write."""
+        nonlocal injected
+        result = await real_get(self, tid)
+        if not injected:
+            injected = True
+            other, other_engine = await _fresh_session(_test_database_url)
+            try:
+                other_task = await other.get(TaskTable, tid)
+                assert other_task is not None
+                markers.set_x_posted_tweet_id(other_task, "concurrent-999")
+                other_task.status = TS.COMPLETED
+                await other.commit()
+            finally:
+                await _dispose(other, other_engine)
+        return result
+
+    client = _StubClient()
+    with (
+        patch("roboco.services.x_post_service.build_x_client", return_value=client),
+        patch.object(TaskService, "get", _get_then_inject_concurrent_post),
+        patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
+        patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
+    ):
+        result = await _svc(db_session).approve(task_id, "Edited body")
+
+    assert result is not None
+    assert result.status == "already_posted"
+    assert result.tweet_id == "concurrent-999"
+    # No double-post: the concurrently-committed tweet wins, ours never fires.
+    assert client.calls == []
+
+    fresh, fresh_engine = await _fresh_session(_test_database_url)
+    try:
+        final = await fresh.get(TaskTable, task_id)
+        assert final is not None
+        assert final.status == TS.COMPLETED
+        assert markers.get_x_posted_tweet_id(final) == "concurrent-999"
+        # The edit must never have landed on the just-posted row.
+        assert markers.get_x_draft_body(final) == "Original"
+    finally:
+        await _dispose(fresh, fresh_engine)
+
+
 # --------------------------------------------------------------------------- #
 # Spotlight video hook (Task 4, 2026-07-09 pipeline fixes): moved from
 # authoring time (propose_feature_spotlight) to this posted-success branch so
@@ -684,9 +873,12 @@ async def test_reject_feature_spotlight_with_wants_video_opens_none(
     task = await _seed_feature_draft(db_session)
     video_engine = AsyncMock()
     video_engine.open_video_task = AsyncMock(return_value=None)
-    with patch(
-        "roboco.services.video_engine.get_video_engine",
-        return_value=video_engine,
+    with (
+        patch(
+            "roboco.services.video_engine.get_video_engine",
+            return_value=video_engine,
+        ),
+        _lock_free(),
     ):
         updated = await _svc(db_session).reject(_id(task), "not on-brand")
     assert updated is not None