test(ai): reduce repeated session and recovery setup

tests/ai/test_evaluation_authorization.py

@@ -233,7 +233,7 @@ async def test_shared_dataset_reopens_with_scoped_policy_and_operator_owned_expe
 
 
 @pytest.mark.asyncio
-async def test_shared_graph_capture_requires_its_own_permission(tmp_path: Path) -> None:
+async def test_shared_graph_capture_requires_its_own_permission() -> None:
     async def answer(context: TaskNodeContext[None]) -> JsonValue:
         return context.input["answer"]
 
@@ -242,7 +242,7 @@ async def answer(context: TaskNodeContext[None]) -> JsonValue:
     target = Task("shared.graph-target", answer, effect_policy="none")
     scorer = Task("shared.graph-score", exact, effect_policy="none")
     async with (
-        Persistence.open('shared-graph', storage=PersistenceStorage.from_root(tmp_path, exclusive_writer=True), tenant_id=TENANT) as persistence_232,
+        Persistence.open('shared-graph', storage=PersistenceStorage.in_memory(), tenant_id=TENANT) as persistence_232,
         TaskEngine.open(target, scorer, persistence=persistence_232, context=CONTEXT, authorization=authorization) as engine,
         Evaluations.open(engine) as evaluations,
     ):
@@ -258,7 +258,7 @@ async def answer(context: TaskNodeContext[None]) -> JsonValue:
         )), principal=PUBLISHER, idempotency_key="dataset")
         request = StartEvaluationRequest(EvaluationSpec(
             dataset, (CandidateSpec("graph", graph_template=GraphTargetSpec(capture=capture, selector="terminal_sinks")),),
-            (rule_scorer(scorer),),
+            (rule_scorer(scorer),), policy=EvaluationPolicy(allow_volatile=True),
         ), OPERATOR, "evaluate")
         with pytest.raises(AIError) as denied:
             await evaluations.start(request)

tests/ai/test_evaluation_consumers.py

@@ -465,7 +465,7 @@ def storage() -> PersistenceStorage:
 
 
 @pytest.mark.asyncio
-async def test_human_score_waits_for_one_decision_and_rescore_keeps_prior_result(tmp_path: Path) -> None:
+async def test_human_score_waits_for_one_decision_and_rescore_keeps_prior_result() -> None:
     calls: list[str] = []
 
     async def target(context: TaskNodeContext[None]) -> JsonValue:
@@ -475,15 +475,16 @@ async def target(context: TaskNodeContext[None]) -> JsonValue:
     task = Task("consumer.human-target", target, effect_policy="none")
     scorer = ScorerSpec("human", TaskRef.deferred_input(), (DIMENSION,))
     async with (
-        Persistence.open('human-consumers', storage=PersistenceStorage.from_root(tmp_path, exclusive_writer=True), tenant_id=TENANT) as persistence_430,
+        Persistence.open('human-consumers', storage=PersistenceStorage.in_memory(), tenant_id=TENANT) as persistence_430,
         TaskEngine.open(task, persistence=persistence_430, context=CONTEXT) as engine,
         Evaluations.open(engine) as evaluations,
     ):
         dataset = await evaluations.publish_dataset(DatasetSpec(DatasetRef("human", 1), cases=(
             CaseSpec.task(CaseRef("human", "one", 1), input={"answer": "yes"}, expected="yes"),
         )), principal=PRINCIPAL, idempotency_key="publish-human")
         run = await evaluations.start(StartEvaluationRequest(EvaluationSpec(dataset,
-            (CandidateSpec("current", task=task.ref),), (scorer,)), PRINCIPAL, "start-human"))
+            (CandidateSpec("current", task=task.ref),), (scorer,),
+            policy=EvaluationPolicy(allow_volatile=True)), PRINCIPAL, "start-human"))
 
         async def wait_for_human(scoring_run: EvaluationRun) -> ScoreAttemptView:
             while True:

tests/ai/test_keyed_execution_recovery.py

@@ -304,12 +304,29 @@ async def test_keyed_recovery_observes_cancel_intent_after_many_unfinished_recei
             OperationKind.EXECUTION_RECOVER, OperationStatus.RUNNING,
             "a" * 64, "previous-claim", None, None, False, now, now,
         )
-        for index in range(257):
-            await state.execution.operations.append(replace(candidate, operation_id=f"old-recover-{index}"))
-        await state.execution.operations.append(replace(
-            candidate, operation_id="pending-cancel", operation_kind=OperationKind.EXECUTION_CANCEL,
-            status=OperationStatus.PENDING, result_ref=None,
-        ))
+        operations = state.execution.operations
+
+        async def seed_receipts(transaction):
+            for index in range(257):
+                await operations.append_in_transaction(
+                    transaction, replace(candidate, operation_id=f"old-recover-{index}"),
+                )
+            await operations.append_in_transaction(transaction, replace(
+                candidate, operation_id="pending-cancel", operation_kind=OperationKind.EXECUTION_CANCEL,
+                status=OperationStatus.PENDING, result_ref=None,
+            ))
+
+        # Recovery observes a committed stream, not the intermediate seed commits.
+        await operations.state_store.mutate(seed_receipts)
+        seeded = await operations.list_pending(
+            ResourceKind.EXECUTION, execution_id, tenant_id="default", limit=259,
+        )
+        assert len(seeded) == 258
+        assert [receipt.operation_id for receipt in seeded[:-1]] == [f"old-recover-{index}" for index in range(257)]
+        assert all(receipt.status is OperationStatus.RUNNING for receipt in seeded[:-1])
+        assert all(left.sequence < right.sequence for left, right in zip(seeded, seeded[1:]))
+        assert seeded[-1].operation_id == "pending-cancel"
+        assert seeded[-1].status is OperationStatus.PENDING
         original = await _producer(state, execution_id)
         await _recover(runtime, execution_id, "recovery")
         receipt = await _receipt(state, "recovery")

tests/ai/test_session_repository_instructions.py

@@ -66,10 +66,13 @@ async def _storage(root: Path, backend: str) -> AsyncIterator[PersistenceStorage
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("backend", ("memory", "sqlite", "sql", "split"))
-@pytest.mark.parametrize("restart", (False, True))
+@pytest.mark.parametrize("capture_frontier", (False, True))
 async def test_session_reloads_known_scopes_before_first_model_request(
-    tmp_path: Path, backend: str, restart: bool,
+    tmp_path: Path, backend: str, capture_frontier: bool,
 ) -> None:
+    # Backend/checkpoint coverage exercises durable scope reload and union.
+    # Filesystem mutations and new-session isolation share the same implementation.
+    full_mutation_path = backend == "sql" and capture_frontier
     workspace_root = tmp_path / "workspace"
     workspace_root.mkdir()
     for scope in ("pkg", "other"):
@@ -125,7 +128,7 @@ async def run(session, name: str, *, initial: str | None = None,
 
     async with _storage(tmp_path, backend) as storage:
         async with Persistence.open("scope-memory", storage=storage) as persistence:
-            if restart:
+            if capture_frontier:
                 if backend == "sql":
                     steps = SqlStepStore(
                         persistence.execution.executions.state_store.context,
@@ -143,22 +146,33 @@ async def run(session, name: str, *, initial: str | None = None,
                 assert stored is not None and stored.repository_instruction_scopes == ("pkg",)
                 assert "PACKAGE-V1" not in str(encode_domain(stored))
 
-            # A new Runtime must consume the durable Session scope set, not a local cache.
+            if full_mutation_path:
+                async with Runtime.attach(persistence=persistence, features=features) as runtime:
+                    session = runtime.agents.get("reader").session("session")
+                    await run(session, "same", initial="PACKAGE-V1")
+
+            # Reload the scope, not the previous execution's content pin, before
+            # the first model request in the new Runtime.
+            rules.write_text("PACKAGE-V2", encoding="utf-8")
             async with Runtime.attach(persistence=persistence, features=features) as runtime:
                 session = runtime.agents.get("reader").session("session")
-                await run(session, "same", initial="PACKAGE-V1")
-                rules.write_text("PACKAGE-V2", encoding="utf-8")
                 await run(session, "changed", initial="PACKAGE-V2", absent=("PACKAGE-V1",))
-                rules.unlink()
-                await run(session, "deleted", absent=("PACKAGE-V1", "PACKAGE-V2"))
-                rules.write_text("PACKAGE-V3", encoding="utf-8")
-                await run(session, "recreated", initial="PACKAGE-V3", absent=("PACKAGE-V1", "PACKAGE-V2"))
+                current_rule = "PACKAGE-V2"
+                if full_mutation_path:
+                    rules.write_text("PACKAGE-V3", encoding="utf-8")
+                    await run(session, "overwritten", initial="PACKAGE-V3", absent=("PACKAGE-V1", "PACKAGE-V2"))
+                    rules.unlink()
+                    await run(session, "deleted", absent=("PACKAGE-V1", "PACKAGE-V2", "PACKAGE-V3"))
+                    rules.write_text("PACKAGE-V4", encoding="utf-8")
+                    await run(session, "recreated", initial="PACKAGE-V4", absent=("PACKAGE-V1", "PACKAGE-V2", "PACKAGE-V3"))
+                    current_rule = "PACKAGE-V4"
                 target = "other/file.txt"
-                await run(session, "unknown", initial="PACKAGE-V3", discover="OTHER-RULE")
-                fork = await session.fork("fork")
-                await run(fork, "fork", discover="OTHER-RULE", absent=("PACKAGE-V3",))
-                isolated = await runtime.agents.get("reader").create_session("isolated")
-                await run(isolated, "isolated", discover="OTHER-RULE", absent=("PACKAGE-V3",))
+                await run(session, "unknown", initial=current_rule, discover="OTHER-RULE")
+                if full_mutation_path:
+                    fork = await session.fork("fork")
+                    await run(fork, "fork", discover="OTHER-RULE", absent=(current_rule,))
+                    isolated = await runtime.agents.get("reader").create_session("isolated")
+                    await run(isolated, "isolated", discover="OTHER-RULE", absent=(current_rule,))
         if backend != "memory":
             async with Persistence.open("scope-memory", storage=storage) as reopened:
                 stored = await reopened.conversation.sessions.get("session", tenant_id=reopened.tenant_id)