ci(release): block publication on agent-server stress regressions

.github/workflows/README-RELEASE.md

@@ -38,17 +38,15 @@ The created PR will include a checklist. Complete the following:
 - [ ] Confirm any merged `release-note-required` PRs are accurately called out in the final release notes
 - [ ] Review and approve the PR
 
-### Step 3: Create the GitHub Release
-
-1. Go to [Releases](https://github.com/OpenHands/software-agent-sdk/releases/new)
-2. Click **"Draft a new release"**
-3. Configure the release:
-   - **Tag**: `vX.Y.Z` (must match the version)
-   - **Branch**: `rel-X.Y.Z` (the branch created by the workflow)
-   - **Previous tag**: Select the previous release version
-4. Click **"Generate release notes"** to auto-generate the changelog
-5. Review and edit the release notes as needed
-6. Click **"Publish release"**
+### Step 3: Merge the Release PR
+
+When the release PR is merged, `create-release.yml` first checks out that exact
+merge commit and runs the complete agent-server stress suite. The release job
+depends on this gate, so a timeout, deadlock, or resource-budget regression
+prevents the GitHub tag/release and all PyPI, image, and binary dispatches.
+
+Only after the gate succeeds does the workflow create the `vX.Y.Z` tag and
+GitHub Release, generate release notes, and dispatch the package/image builds.
 
 ### Step 4: PyPI Publication (Automated)
 

.github/workflows/create-release.yml

@@ -11,11 +11,43 @@ on:
         branches: [main]
 
 jobs:
+    pre-release-stress-tests:
+        # Validate the exact merge commit before creating any release artifact.
+        if: >
+            github.event.pull_request.merged == true &&
+            startsWith(github.event.pull_request.head.ref, 'rel-')
+        runs-on: blacksmith-2vcpu-ubuntu-2404
+        timeout-minutes: 10
+        steps:
+            - name: Checkout release merge commit
+              uses: actions/checkout@v7
+              with:
+                  ref: ${{ github.event.pull_request.merge_commit_sha }}
+
+            - name: Install uv
+              uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7
+              with:
+                  enable-cache: true
+                  python-version: '3.13'
+
+            - name: Install deps
+              run: uv sync --frozen --group dev
+
+            - name: Run Agent Server stress tests
+              run: |
+                  # Do not use xdist: these tests assert process-wide resource
+                  # and timing budgets that parallel collection would invalidate.
+                  CI=true uv run python -m pytest -vvs \
+                    -m stress \
+                    --durations=10 \
+                    tests/agent_server/stress
+
     create-release:
         # Only run when a release PR is merged (not just closed)
         if: >
             github.event.pull_request.merged == true &&
             startsWith(github.event.pull_request.head.ref, 'rel-')
+        needs: pre-release-stress-tests
         runs-on: ubuntu-24.04
         permissions:
             actions: write

tests/agent_server/stress/budgets.py

@@ -135,6 +135,16 @@ class LeaseContentionBudget:
     settle_timeout_s: float = 5.0
 
 
+@dataclass(frozen=True, slots=True)
+class LifecycleIsolationBudget:
+    # Enough simultaneous operations to expose a process-wide lifecycle lock
+    # without making teardown expensive on shared CI runners.
+    n_unrelated_conversations: int = 4
+    # A blocked close is intentionally unbounded. Unrelated lifecycle work must
+    # still finish within this deliberately loose CI-tolerant deadline.
+    unrelated_operations_timeout_s: float = 5.0
+
+
 PARALLEL_SUBAGENTS = ParallelSubagentBudget()
 CONVERSATION_LISTING = ConversationListingBudget()
 CONCURRENT_CONVERSATIONS = ConcurrentConversationsBudget()
@@ -145,3 +155,4 @@ class LeaseContentionBudget:
 WEBSOCKET_RECONNECT_STORM = WebsocketReconnectStormBudget()
 HIGH_VOLUME_BASH_OUTPUT = HighVolumeBashOutputBudget()
 LEASE_CONTENTION = LeaseContentionBudget()
+LIFECYCLE_ISOLATION = LifecycleIsolationBudget()

tests/agent_server/stress/test_lifecycle_isolation.py

@@ -0,0 +1,152 @@
+"""Stress test: one stuck close must not wedge unrelated conversations.
+
+Bug class this catches:
+    - A process-wide lifecycle lock held across ``EventService.close()``.
+      If one close hangs, create, load, and delete operations for every other
+      conversation queue behind it indefinitely (#4514, fixed by #4570).
+
+The blocking subscriber delays a real EventService close at its normal pub/sub
+teardown boundary. All operations under test still use the production
+ConversationService and persistence paths with credential-free TestLLMs.
+"""
+
+import asyncio
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+import pytest
+
+from openhands.agent_server.conversation_service import ConversationService
+from openhands.agent_server.pub_sub import Subscriber
+from openhands.sdk.event import Event
+from tests.agent_server.stress.budgets import LIFECYCLE_ISOLATION
+from tests.agent_server.stress.scripts import (
+    SlowTestLLM,
+    start_conversation_with_test_llm,
+    text_message,
+)
+
+
+pytestmark = [pytest.mark.stress, pytest.mark.timeout(30)]
+
+
+@dataclass(slots=True)
+class _BlockingCloseSubscriber(Subscriber[Event]):
+    close_entered: asyncio.Event = field(default_factory=asyncio.Event)
+    release_close: asyncio.Event = field(default_factory=asyncio.Event)
+
+    async def __call__(self, event: Event) -> None:
+        pass
+
+    async def close(self) -> None:
+        self.close_entered.set()
+        await self.release_close.wait()
+
+
+def _idle_test_llm() -> SlowTestLLM:
+    llm = SlowTestLLM.from_messages([text_message("done")], latency_s=0.0)
+    assert isinstance(llm, SlowTestLLM)
+    return llm
+
+
+async def _create_idle_conversation(
+    conversation_service: ConversationService,
+    *,
+    workspace_dir: str,
+    usage_id: str,
+):
+    return await start_conversation_with_test_llm(
+        conversation_service,
+        parent_llm=_idle_test_llm(),
+        workspace_dir=workspace_dir,
+        usage_id=usage_id,
+        initial_text=None,
+    )
+
+
+async def test_stuck_close_does_not_block_unrelated_lifecycle_operations(
+    conversation_service: ConversationService,
+    tmp_path,
+):
+    workspace = tmp_path / "ws"
+    workspace.mkdir()
+
+    blocked = await _create_idle_conversation(
+        conversation_service,
+        workspace_dir=str(workspace),
+        usage_id="lifecycle-blocked",
+    )
+    unrelated = await asyncio.gather(
+        *[
+            _create_idle_conversation(
+                conversation_service,
+                workspace_dir=str(workspace),
+                usage_id=f"lifecycle-unrelated-{i}",
+            )
+            for i in range(LIFECYCLE_ISOLATION.n_unrelated_conversations)
+        ]
+    )
+
+    blocked_service = await conversation_service.get_event_service(blocked.id)
+    assert blocked_service is not None
+    blocker = _BlockingCloseSubscriber()
+    blocked_service._pub_sub.subscribe(blocker)
+
+    blocked_delete = asyncio.create_task(
+        conversation_service.delete_conversation(blocked.id)
+    )
+    try:
+        await asyncio.wait_for(
+            blocker.close_entered.wait(),
+            timeout=LIFECYCLE_ISOLATION.unrelated_operations_timeout_s,
+        )
+
+        started_at = time.monotonic()
+        load_task = asyncio.create_task(
+            conversation_service.get_event_service(unrelated[0].id)
+        )
+        delete_tasks = [
+            asyncio.create_task(conversation_service.delete_conversation(info.id))
+            for info in unrelated[1:]
+        ]
+        create_task = asyncio.create_task(
+            _create_idle_conversation(
+                conversation_service,
+                workspace_dir=str(workspace),
+                usage_id="lifecycle-created-during-close",
+            )
+        )
+        operations: list[asyncio.Task[Any]] = [
+            load_task,
+            *delete_tasks,
+            create_task,
+        ]
+        _done, pending = await asyncio.wait(
+            operations,
+            timeout=LIFECYCLE_ISOLATION.unrelated_operations_timeout_s,
+        )
+        if pending:
+            for task in pending:
+                task.cancel()
+            await asyncio.gather(*pending, return_exceptions=True)
+            raise AssertionError(
+                "unrelated conversation create/load/delete operations blocked "
+                "behind a stuck close; lifecycle work may be globally serialized"
+            )
+        elapsed = time.monotonic() - started_at
+
+        assert load_task.result() is not None
+        assert all(task.result() for task in delete_tasks)
+        created = create_task.result()
+        assert await conversation_service.get_event_service(created.id) is not None
+        assert elapsed < LIFECYCLE_ISOLATION.unrelated_operations_timeout_s
+        assert not blocked_delete.done(), (
+            "blocked close unexpectedly completed before its subscriber was released"
+        )
+    finally:
+        blocker.release_close.set()
+        assert await asyncio.wait_for(
+            blocked_delete,
+            timeout=LIFECYCLE_ISOLATION.unrelated_operations_timeout_s,
+        )