From 17de29545ab122b4378c12e508b6cf9205eb0186 Mon Sep 17 00:00:00 2001 From: Renzo F <45401804+rennf93@users.noreply.github.com> Date: Wed, 22 Jul 2026 08:11:28 +0200 Subject: [PATCH] [6788ce7f] Silent bug sweep: concurrency, state integrity, engine edge-cases, panel data freshness (#638) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * [943d8c4d] Frontend data freshness and approval-queue reliability audit (#631) * [233a8b0f] WebSocket reconnect message-loss audit and fix (#625) * [233a8b0f] fix(panel): add REST catch-up to useNotificationStream on WS reconnect connection.ts has no message buffering/replay, so a notification published while the CEO bell's socket was down (disconnected/reconnecting) was lost forever instead of merely delayed. Add a reconnect-triggered GET /notifications?unread_only=true catch-up folded into the existing notification_id dedup so a notification delivered both via catch-up and live WS is never double-counted, and make clearMessages drop the held catch-up batch too. use-a2a-live.ts and use-rate-limit-websocket.ts were audited and already have working reconnect-triggered REST fallbacks (verified via a2a/page.tsx, rate-limit-banner.tsx, usage-overview-panel.tsx and their existing F083 tests) so no fix was needed there. * [233a8b0f] docs(panel): add comprehensive WebSocket hooks reference and reconnect architecture guide Add panel/docs/frontend/hooks.md with full API reference for useWebSocket, useNotificationStream (with new REST catch-up behavior), useAgentStream, useA2ALiveStream, and useConnectionStatus. Include examples, best practices, and testing guidance. Add panel/docs/architecture/websocket-reconnect.md documenting the message-loss mitigation pattern: Strategy 1 (REST catch-up for events, used by useNotificationStream) and Strategy 2 (REST invalidation for state, used by A2A/rate-limit consumers), plus the dedup logic ensuring no notification is double-counted on reconnect. --------- Co-authored-by: Frontend Developer 1 Co-authored-by: Frontend Documenter * [d5315683] fix(frontend): add distinct toast feedback for silently-swallowed x-post and release-proposal statuses, plus regression tests for all 4 approval queues (#626) Co-authored-by: Frontend Developer 2 * [cd953838] Data-hook null-guard audit and API client 429 retry-by-method fix (#630) * [cd953838] fix(panel): gate 429 retry by HTTP method, add hook null-guard regression tests * [cd953838] chore(conventions): waive test-fixture wrapper in hooks null-guard test * [cd953838] docs(frontend): document API rate-limit retry behavior and null-guard audit results Added `docs/frontend/api-rate-limiting.md` to document the 429 retry strategy: GET/PUT auto-retry, POST/PATCH/DELETE require X-Idempotency-Key header. Updated `docs/frontend/hooks.md` to confirm the data-hook null-guard audit found all hooks already have correct `enabled` guards and include a regression test suite for the board-review poll on/off behavior and enabled-guard assertions. --------- Co-authored-by: Frontend Developer 1 Co-authored-by: Frontend Documenter --------- Co-authored-by: Frontend Developer 1 Co-authored-by: Frontend Documenter Co-authored-by: Frontend Developer 2 * [4534c71a] Backend concurrency, state-machine, and engine audit (#634) * [41de844a] fix(lifecycle): sync CLAIM_RULES with runtime + clear stale claimant on PM hand-off (#627) Two confirmed state-machine gaps found while auditing lifecycle.py, task_lifecycle.py, the _ESCALATABLE_TO_BLOCKED bypass, and every _REVIEW_QUEUE_STATES entry point: - lifecycle.py's CLAIM_RULES/claim-ActionSpec/StatusTransition table did not grant CELL_PM/MAIN_PM re-claim of AWAITING_PM_REVIEW even though task.py's runtime _ROLE_CLAIM_STATUSES already granted it and claimed the spec agreed -- the two tables had silently drifted, breaking i_will_plan re-claim on an awaiting_pm_review task. - docs_complete's _maybe_advance_to_pm_review pre-assigns a specific owning PM via assigned_to but left claimed_by/active_claimant_id pointing at the outgoing documenter, unlike every sibling transition into a review-queue state. A stale active_claimant_id makes content_actions.py's _active_claim_violation wrongly reject the newly-assigned PM's own content writes before it formally claims. Reassign claimed_by + active_claimant_id to the owning PM alongside assigned_to. Adds a regression test asserting the documenter's stale claim does not survive the docs_complete -> awaiting_pm_review hand-off. Co-authored-by: Backend Developer 2 * [0c46f666] Engine dedup race + sequencing.py edge-case audit (#628) * [0c46f666] fix(sequencing): dedup race audit + collision-edge fallback bug Audited the list-open-then-originate dedup pattern across six engines: RoadmapEngine, XEngine.run_cycle, DepUpdateEngine, and CIWatchEngine each run inside exactly one sequential orchestrator-loop asyncio task (no other call site invokes run_cycle), so they cannot race with themselves; their in-cycle dedup sets/keys are correctly built before any commit. SelfHealEngine is the same shape. VideoEngine.open_video_task is genuinely different: it is reachable from the release-publish hook, the feature-spotlight hook, and the on-demand POST /video/request route, so two overlapping calls for the same occasion can both pass the "no open task yet" check before either commits. Fixed by wrapping the check+insert in a short-lived Redis mutex (reusing HeartbeatMutex) keyed by occasion, mirroring XPostService's existing lock pattern, with a regression test proving only one of two concurrent calls creates a task. Verified ReleaseExecutor's half-landed retry path (release_commit_sha): 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 a bumped-version-without-changelog state can never reach origin (and therefore can never be observed by a fresh retry clone) - confirmed correct with a real-git-repo regression test, no fix needed. Fixed sequencing.py's dev_task_collision_edges: the `if edges: return edges` short-circuit dropped the same-assignee-lane fallback entirely whenever ANY surfaced sibling pair produced a collision edge, even for a completely unrelated same-assignee pair with no declared surface. Now the fallback always runs, skipping only pairs the analyzer already ordered (so the two mechanisms can never disagree on direction for the same pair). Verified sequencing.py rule 3 (all-shared batch generates no edges): correct by inspection (_shared_last_edges skips every pair when both are shared) and confirmed with a regression test - no fix needed. * [0c46f666] docs(reference): concurrency audit summary - engine races, fixes, verified patterns --------- Co-authored-by: Backend Developer 2 Co-authored-by: Backend Documenter * [8f7f167a] Redis mutex pre-lock write audit (#629) * [8f7f167a] Redis mutex pre-lock write audit: add cross-session regression test for XPostService.approve Audited x_post_service.py, video_post_service.py, release_proposal.py, and heartbeat_mutex.py for the pre-lock DB-write anti-pattern (a session write that happens before the SET NX / HeartbeatMutex acquire returns a token, letting a losing racer's stale write clobber a winner's committed state). XPostService.approve, VideoPostService.approve, and ReleaseProposalService.approve/reject already implement the correct validate-pure-pre-lock, apply-under-lock pattern (the XPostService fix already shipped per CHANGELOG.md: "X edited_body write deferred into the single-flight lock (M5)"). HeartbeatMutex holds no AsyncSession at all, so the anti-pattern is structurally inapplicable there. Adds a genuine cross-session concurrency regression test to test_x_post_service.py (a real second DB connection, not an in-process mock) mirroring VideoPostService's existing cross-session test, proving a concurrently-committed post survives and the CEO's edited body never lands on the just-posted row. * [8f7f167a] Remove redundant inline comments flagged by QA in cross-session regression test Both comments restated what the surrounding docstrings already say explicitly, per QA findings F-dbadd8f0 (line 294) and F-27ac051e (line 631) — no behavior change, tests re-verified green against a sandbox Postgres. * [8f7f167a] Remove inline trailing comments flagged by QA (correct file this time) QA findings F-e6f3e6a6 and F-24189858 cited tests/unit/services/ test_x_post_service.py:294 and :631 across 5 revision rounds, but that file never contained the flagged comment text — a repo-wide grep for the exact quoted strings shows both comments actually live in the mirrored tests/unit/services/test_video_post_service.py file, in its own cross-session concurrency regression tests (the caption-edit and tiktok-skip tests). Removed both there: - "# externally visible to the "concurrent" session below" on the db_session.commit() call - "# never attempted without credentials" on the tiktok_poster.calls assertion Both restated what the surrounding docstrings/test names already say; no behavior change. Verified with the full make quality gate against a sandbox Postgres/Redis: 13,717 passed, 94.41% coverage, clean except one pre-existing unrelated failure in tests/unit/api/test_cloud_auth.py ::test_login_route_parses_oauth2_form_not_query_params, which connects to the app's default localhost:5432 Postgres (not the db_session sandbox fixture) and is unreachable in this sandboxed environment — structurally unrelated to the auth subsystem this task never touches. * [8f7f167a] Redis mutex pre-lock write audit (round 7): add cross-session regression tests for reject() lock protection Round-7 QA findings F-7eb9fbcb, F-06f39a2e, and F-4d56e49b claim XPostService.reject(), ReleaseProposalService.reject(), and release_executor._await_proc() lack lock protection / a CancelledError handler — but their cited line ranges (255-267, 429-454, 241-257) describe a pre-fix, shorter version of these functions that predates commit fb293a787d, already on this branch. At current HEAD: - x_post_service.py reject() (lines 275-299) acquires _LOCK_PREFIX, re-reads under the lock, applies markers.set_x_reject_reason() + CANCELLED only inside the critical section, releases in finally. - release_proposal.py reject() (lines 460-486) does the identical dance with _RELEASE_LOCK_PREFIX. - release_executor.py _await_proc() (lines 257-265) already has an `except asyncio.CancelledError` block that kills + reaps the child and re-raises, mirroring the TimeoutError handler, with an existing dedicated regression test (test_await_proc_kills_child_on_outer_cancellation). The one genuine gap: neither reject() path had a cross-session (real second DB connection, not an in-process mock) regression test proving the in-lock re-read catches a concurrent approve/publish that completes mid-lock-wait — only approve() had one. Added test_reject_concurrent_approve_completes_during_lock_wait to both test_x_post_service.py and test_release_proposal_status_guards.py, mirroring the existing approve() cross-session test: a second engine commits COMPLETED between reject's pre-lock read and lock acquisition, and the test asserts the CANCELLED write / reject-reason marker never lands on the just-completed row. No production code changed — verified via 103 targeted tests green against a sandbox Postgres/Redis, plus `make -o sync gate` clean. * [8f7f167a] Regenerate stale lifecycle artifacts (restore auditor waive_finding) foundation-check was the only failing gate: the committed lifecycle artifacts were missing the auditor's waive_finding verb that the lifecycle source defines, so make quality regenerated them and failed on the diff — nothing to do with the mutex fix (which passes ruff/mypy/tests/coverage/bandit clean). make lifecycle restores the drift; this is what the 8 revision rounds kept missing. --------- Co-authored-by: Backend Developer 1 Co-authored-by: Renn F * [d615f2e3] fix(tests): sync stale CLAIM_RULES pinning assertions with lifecycle.py (#635) test_claim_rules_match_pre_gateway_table still asserted the pre-audit two-member frozenset for CELL_PM/MAIN_PM claim rules. CLAIM_RULES in lifecycle.py already grants both roles claim rights on Status.AWAITING_PM_REVIEW (added by the state-machine exhaustiveness audit) so a PM can re-claim its own review-queue task after a respawn. Updated both assertions to include AWAITING_PM_REVIEW, matching the actual dict. Grepped the repo for sibling stale copies of the old literal; found none beyond this test. Co-authored-by: Backend Developer 1 * [3c4e7a35] fix(quality-gate): reflow CONCURRENCY_AUDIT.md and stub occasion lock in video tests (#636) Root cause: PR #634's CI failed at the markdown-reflow check (make quality Makefile:285) on CONCURRENCY_AUDIT.md — a hard-wrapped audit doc left over from the merged "Engine dedup race + sequencing.py edge-case audit" unit (PR #628). Fixed with `make reflow-docs` (the exact remedy the CI output itself named). Running the full local `make quality` (with a sandbox Postgres/Redis to get past DB-gated skips) surfaced a second real regression from that same PR #628 unit: it added a Redis-backed HeartbeatMutex occasion lock to VideoEngine.open_video_task, but two pre-existing test files (tests/unit/runtime/test_video_render_loop.py and tests/integration/test_video_routes.py) call open_video_task without stubbing that lock, so they failed closed against the suite's deliberately-unreachable test Redis (_no_live_redis). Fixed by applying the same lock-stub pattern tests/unit/services/test_video_engine.py already uses for its own occasion-lock tests: an autouse HeartbeatMutex stand-in fixture in test_video_render_loop.py, and wrapping the two route-level video-request tests in test_video_routes.py with the file's existing _LOCKED patch pair (already used by every other lock-dependent test in that file). The one remaining local failure, test_cloud_auth.py::test_login_route_parses_oauth2_form_not_query_params, is a pre-existing environment gap unrelated to this branch: it needs a real Postgres reachable at localhost:5432 (which .github/workflows/ci.yml provides as a service container) but this dev sandbox has no such binding — confirmed unrelated to any of the four merged audit units. make quality now passes clean: 13729 passed, 0 regressions, 94.49% coverage. Co-authored-by: Backend Developer 1 * [ddc8121f] regenerate lifecycle artifacts for awaiting_pm_review claim rules and waive_finding intent (#637) Co-authored-by: Backend Developer 1 --------- Co-authored-by: Backend Developer 2 Co-authored-by: Backend Documenter Co-authored-by: Backend Developer 1 Co-authored-by: Renn F * fix(sequencing): drop lane-fallback edges that would cycle against analyzer edges The dev-task collision fallback unioned the analyzer's authoritative edges with same-assignee lane edges, deduping only the direct pair. A lane chain through an unsurfaced middle sibling could still contradict an analyzer edge transitively (the shared-last migration order inverts plain priority order), closing a 3-cycle that made add_dependency raise ConflictError and wedged every later delegate to that parent. Fallback edges are now accepted only when they can't close a cycle against the edges already kept; a regression test reproduces the exact scenario. Also strip pre-merge cruft: remove the root CONCURRENCY_AUDIT.md working report, delete the near-duplicate websocket-reconnect.md doc, fix the stale a2a/page.tsx doc citation, correct the api-rate-limiting doc to state idempotency-key retry is unimplemented, and fix two lifecycle.py comments that referenced a guard function which never existed. --------- Co-authored-by: Frontend Developer 1 Co-authored-by: Frontend Documenter Co-authored-by: Frontend Developer 2 Co-authored-by: Backend Developer 2 Co-authored-by: Backend Documenter Co-authored-by: Backend Developer 1 Co-authored-by: Renn F --- .roboco/conventions.yml | 9 +- docs/frontend/api-rate-limiting.md | 44 ++ docs/frontend/hooks.md | 33 ++ docs/rag/lifecycle/status-transitions.md | 1 + panel/docs/frontend/hooks.md | 432 ++++++++++++++++++ panel/lib/lifecycle.json | 11 + ...ase-proposal-card-status-feedback.test.tsx | 160 +++++++ .../__tests__/roadmap-review-queue.test.tsx | 52 +++ .../__tests__/video-post-queue.test.tsx | 72 +++ .../dashboard/__tests__/x-post-queue.test.tsx | 59 +++ .../dashboard/release-proposal-card.tsx | 16 +- .../src/components/dashboard/x-post-queue.tsx | 4 + .../use-notification-stream.test.tsx | 222 +++++++++ .../__tests__/use-tasks-null-guards.test.tsx | 110 +++++ .../hooks/__tests__/use-websocket.test.tsx | 7 + panel/src/hooks/use-websocket.ts | 59 ++- panel/src/lib/__tests__/client.test.ts | 37 +- panel/src/lib/api/client.ts | 54 ++- roboco/api/routes/release.py | 10 +- roboco/foundation/policy/lifecycle.py | 34 +- roboco/services/release_executor.py | 9 + roboco/services/release_proposal.py | 40 +- roboco/services/sequencing.py | 66 ++- roboco/services/task.py | 11 + roboco/services/video_engine.py | 70 +++ roboco/services/x_post_service.py | 42 +- tests/foundation/test_lifecycle_spec.py | 12 +- tests/integration/test_release_routes.py | 38 +- .../test_task_service_transitions.py | 53 +++ tests/integration/test_video_routes.py | 25 +- tests/integration/test_x_routes.py | 30 +- tests/unit/runtime/test_video_render_loop.py | 21 + tests/unit/services/test_release_executor.py | 105 ++++- ...est_release_executor_subprocess_timeout.py | 41 +- .../test_release_proposal_status_guards.py | 98 +++- tests/unit/services/test_sequencing.py | 110 +++++ tests/unit/services/test_video_engine.py | 126 +++++ .../unit/services/test_video_post_service.py | 4 +- tests/unit/services/test_x_post_service.py | 220 ++++++++- 39 files changed, 2458 insertions(+), 89 deletions(-) create mode 100644 docs/frontend/api-rate-limiting.md create mode 100644 panel/docs/frontend/hooks.md create mode 100644 panel/src/components/dashboard/__tests__/release-proposal-card-status-feedback.test.tsx create mode 100644 panel/src/hooks/__tests__/use-notification-stream.test.tsx create mode 100644 panel/src/hooks/__tests__/use-tasks-null-guards.test.tsx 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 . 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 . 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

No task selected

; + if (isLoading) return

Loading...

; + if (error) return

Error: {error.message}

; + + return
{data?.title}
; +} +``` + +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(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` | `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 ( + <> +

Connection: {state}

+ {isConnected &&

Last update: {lastMessage?.timestamp}

} +
    + {messages.map((msg, i) => ( +
  • {JSON.stringify(msg)}
    • + ))} +
+ + ); +} +``` + +--- + +## `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 ( + <> + + {!isConnected && } + + ); +} +``` + +--- + +## `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 ( +
+ {isConnecting && Connecting…} + {streamOutput} +

+ {messages.length} frames • {streamOutput.length} chars +

+
+ ); +} +``` + +--- + +## `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 ( +
+

+ {isConnected ? "Connected" : "Offline"} + {a2aMessages.length > 0 && " (live updates)"} +

+ {/* Render conversations with full bodies from REST */} +
+ ); +} +``` + +--- + +## `useConnectionStatus()` + +Tracks the connection state of **all active subscriptions** in a single hook. Useful for global connection indicators. + +### Return Value + +```typescript +{ + connections: Record; // { [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 ( +
+ {allConnected && Connected} + {hasActiveConnections && !allConnected && ( + Reconnecting… + )} + {!hasActiveConnections && Offline} +
+ ); +} +``` + +--- + +## 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 && } + {isConnected && } + ``` + +--- + +## 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 => ({ + 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 ( + + {ui} + + ); +} + +async function clickApprove() { + render(withProviders()); + 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()); + 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()); + 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()); + 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()); + 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()); + 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 | 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(); + 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(); + 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(); + 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(); + 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( + "@/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 {children}; +} + +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([]); + 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(); 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((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((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