diff --git a/tests/ai/test_evaluation_authorization.py b/tests/ai/test_evaluation_authorization.py index 9de5c932d..ea44d275d 100644 --- a/tests/ai/test_evaluation_authorization.py +++ b/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) diff --git a/tests/ai/test_evaluation_consumers.py b/tests/ai/test_evaluation_consumers.py index 4d570d73f..946524294 100644 --- a/tests/ai/test_evaluation_consumers.py +++ b/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,7 +475,7 @@ 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, ): @@ -483,7 +483,8 @@ async def target(context: TaskNodeContext[None]) -> JsonValue: 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: diff --git a/tests/ai/test_keyed_execution_recovery.py b/tests/ai/test_keyed_execution_recovery.py index 6f005e425..c19f99e46 100644 --- a/tests/ai/test_keyed_execution_recovery.py +++ b/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") diff --git a/tests/ai/test_session_repository_instructions.py b/tests/ai/test_session_repository_instructions.py index 313942560..c7bdb1d01 100644 --- a/tests/ai/test_session_repository_instructions.py +++ b/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)