improve lease renewal eval based on real user issues

- add concurrency limit fixture (like integration tests)
- flow logs show acquiring concurrency slot before crash
- connects the flow to the concurrency limit via logs
- agent investigates logs + crash state to diagnose

based on patterns from:
- PrefectHQ/prefect#19068
- PrefectHQ/prefect#18839

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: zzstoatzz

evals/test_lease_renewal_crash.py

@@ -14,15 +14,13 @@
 from uuid import uuid4
 
 import pytest
-from prefect import flow
+from prefect import flow, get_run_logger
 from prefect.client.orchestration import PrefectClient
 from prefect.client.schemas.actions import GlobalConcurrencyLimitCreate
 from prefect.client.schemas.objects import FlowRun, GlobalConcurrencyLimit
 from prefect.states import Crashed
 from pydantic_ai import Agent
 
-from evals._tools.spy import ToolCallSpy
-
 LEASE_RENEWAL_ERROR = (
     "Concurrency lease renewal failed - slots are no longer reserved. "
     "Terminating execution to prevent over-allocation."
@@ -31,37 +29,41 @@
 
 @pytest.fixture
 async def concurrency_limit(prefect_client: PrefectClient) -> GlobalConcurrencyLimit:
-    """Create a global concurrency limit that the flow was using."""
-    limit_name = f"api-calls-{uuid4().hex[:8]}"
+    """Create a global concurrency limit."""
+    limit_name = f"database-connections-{uuid4().hex[:8]}"
     await prefect_client.create_global_concurrency_limit(
         concurrency_limit=GlobalConcurrencyLimitCreate(
             name=limit_name,
-            limit=5,
+            limit=1,
         )
     )
     return await prefect_client.read_global_concurrency_limit_by_name(limit_name)
 
 
 @pytest.fixture
-async def crashed_lease_renewal_flow_run(
+async def crashed_flow_run(
     prefect_client: PrefectClient,
     concurrency_limit: GlobalConcurrencyLimit,
 ) -> FlowRun:
     """Create a flow run that crashed due to lease renewal failure.
 
-    The flow was using a concurrency limit when its lease renewal failed,
-    causing Prefect to crash the run to prevent over-allocation.
+    The flow logs show it acquired a concurrency slot before crashing,
+    matching the real user experience from GitHub issues.
     """
 
-    @flow(name=f"api-consumer-{uuid4().hex[:8]}")
-    def api_consumer():
+    @flow(name=f"db-sync-job-{uuid4().hex[:8]}")
+    def db_sync_job():
+        logger = get_run_logger()
+        logger.info(f"Acquired concurrency slot for '{concurrency_limit.name}'")
+        logger.info("Starting database sync operation")
+        logger.info("Processing batch 1 of 5")
+        logger.info("Processing batch 2 of 5")
         return "completed"
 
-    # Run the flow to create a flow run
-    state = api_consumer(return_state=True)
+    state = db_sync_job(return_state=True)
     flow_run = await prefect_client.read_flow_run(state.state_details.flow_run_id)
 
-    # Force to Crashed state with lease renewal error message
+    # Simulate the crash that occurs when lease renewal fails
     crashed_state = Crashed(message=LEASE_RENEWAL_ERROR)
     await prefect_client.set_flow_run_state(
         flow_run_id=flow_run.id,
@@ -74,15 +76,12 @@ def api_consumer():
 
 async def test_diagnoses_lease_renewal_failure(
     simple_agent: Agent,
-    crashed_lease_renewal_flow_run: FlowRun,
-    concurrency_limit: GlobalConcurrencyLimit,
+    crashed_flow_run: FlowRun,
     evaluate_response: Callable[[str, str], Awaitable[None]],
-    tool_call_spy: ToolCallSpy,
 ) -> None:
     """Test agent identifies concurrency lease renewal failure as crash cause."""
     prompt = (
-        f"My flow run '{crashed_lease_renewal_flow_run.name}' crashed unexpectedly. "
-        "What happened?"
+        f"My flow run '{crashed_flow_run.name}' crashed unexpectedly. What happened?"
     )
 
     async with simple_agent:
@@ -95,5 +94,3 @@ async def test_diagnoses_lease_renewal_failure(
         "terminated because the lease could not be renewed.",
         result.output,
     )
-
-    tool_call_spy.assert_tool_was_called("get_flow_runs")