From d3b8c9d6ec119c238d5156aadc8c41eb73550197 Mon Sep 17 00:00:00 2001 From: ice-black-tea <30665081+ice-black-tea@users.noreply.github.com> Date: Sat, 14 Nov 2026 21:43:32 +0800 Subject: [PATCH 1/2] test(ai): reduce repeated session and recovery setup --- tests/ai/test_evaluation_authorization.py | 6 +-- tests/ai/test_keyed_execution_recovery.py | 29 +++++++++--- .../test_session_repository_instructions.py | 44 ++++++++++++------- 3 files changed, 55 insertions(+), 24 deletions(-) diff --git a/tests/ai/test_evaluation_authorization.py b/tests/ai/test_evaluation_authorization.py index 9de5c932d9..ea44d275d3 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_keyed_execution_recovery.py b/tests/ai/test_keyed_execution_recovery.py index 6f005e4250..c19f99e46d 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 3139425604..c7bdb1d01d 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) From bde0924dfe5511b3887519c8fabd016f963dbeb3 Mon Sep 17 00:00:00 2001 From: ice-black-tea <30665081+ice-black-tea@users.noreply.github.com> Date: Sun, 15 Nov 2026 03:34:44 +0800 Subject: [PATCH 2/2] test(ai): avoid disk setup for in-process human scoring --- tests/ai/test_evaluation_consumers.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/ai/test_evaluation_consumers.py b/tests/ai/test_evaluation_consumers.py index 4d570d73fc..9465242946 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: