add concurrency limit context to lease renewal eval

- adds a global concurrency limit fixture as environmental context
- removes leading hint from prompt - agent discovers cause from state message
- follows pattern of other evals: context exists, generic prompt, agent discovers

🤖 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

@@ -16,7 +16,8 @@
 import pytest
 from prefect import flow
 from prefect.client.orchestration import PrefectClient
-from prefect.client.schemas.objects import FlowRun
+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
 
@@ -29,15 +30,35 @@
 
 
 @pytest.fixture
-async def crashed_lease_renewal_flow_run(prefect_client: PrefectClient) -> FlowRun:
-    """Create a flow run that crashed due to lease renewal failure."""
+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]}"
+    await prefect_client.create_global_concurrency_limit(
+        concurrency_limit=GlobalConcurrencyLimitCreate(
+            name=limit_name,
+            limit=5,
+        )
+    )
+    return await prefect_client.read_global_concurrency_limit_by_name(limit_name)
+
 
-    @flow(name=f"data-pipeline-{uuid4().hex[:8]}")
-    def data_pipeline():
+@pytest.fixture
+async def crashed_lease_renewal_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.
+    """
+
+    @flow(name=f"api-consumer-{uuid4().hex[:8]}")
+    def api_consumer():
         return "completed"
 
     # Run the flow to create a flow run
-    state = data_pipeline(return_state=True)
+    state = api_consumer(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
@@ -54,13 +75,15 @@ def data_pipeline():
 async def test_diagnoses_lease_renewal_failure(
     simple_agent: Agent,
     crashed_lease_renewal_flow_run: FlowRun,
+    concurrency_limit: GlobalConcurrencyLimit,
     evaluate_response: Callable[[str, str], Awaitable[None]],
     tool_call_spy: ToolCallSpy,
 ) -> None:
     """Test agent identifies concurrency lease renewal failure as crash cause."""
+    # Don't mention concurrency - let the agent discover the cause from the crash state
     prompt = (
-        f"Why did my flow run '{crashed_lease_renewal_flow_run.name}' crash "
-        "unexpectedly during execution? It was running fine and then suddenly crashed."
+        f"My flow run '{crashed_lease_renewal_flow_run.name}' crashed unexpectedly. "
+        "What happened?"
     )
 
     async with simple_agent:
@@ -74,5 +97,4 @@ async def test_diagnoses_lease_renewal_failure(
         result.output,
     )
 
-    # Agent must use get_flow_runs to retrieve the crash details
     tool_call_spy.assert_tool_was_called("get_flow_runs")