""" 验证整理执行证据、CAS fencing 与终态结算持久化。 """ from dataclasses import replace from datetime import datetime, timezone import pytest from sqlalchemy import create_engine, create_mock_engine, select, text from sqlalchemy.dialects import postgresql from sqlalchemy.orm import Session, sessionmaker from app.application.transfer.execution import ( TransferExecutionCheckpoint, TransferExecutionCommand, TransferExecutionConflictError, TransferExecutionState, TransferManualReviewDecision, TransferManualReviewQuery, TransferOperationObservation, TransferOperationObservationState, TransferStepIntent, TransferStepResult, TransferStepState, build_transfer_checkpoint_fingerprint, build_transfer_operation_id, ) from app.application.transfer.recovery import TransferRecoveryCommand from app.application.transfer.workflow import ( TransferPlanCheckpoint, TransferPlanItem, TransferPlanningInput, TransferProviderInvocationSnapshot, TransferProviderReference, ) from app.db.adapters.transfer.admission import TransactionalTransferAdmissionRepository from app.db.adapters.transfer.execution import ( TransactionalTransferExecutionRepository, ) from app.db.base import Base from app.db.models.transferexecutionstep import TransferExecutionStep from app.db.models.transferhistory import TransferHistory from app.db.models.transferpending import TransferPending from app.db.oper.transferexecutionstep import TransferExecutionStepOper from app.db.oper.transferpending import TransferPendingOper @pytest.fixture def execution_store(): """ 构造只含整理执行相关表的独立内存数据库。 """ engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all( engine, tables=[ TransferPending.__table__, TransferHistory.__table__, TransferExecutionStep.__table__, ], ) factory = sessionmaker(bind=engine, expire_on_commit=False) try: yield factory finally: engine.dispose() def _planning_input(task_id: str) -> TransferPlanningInput: """ 构造与测试源文件一致的持久规划输入。 """ return TransferPlanningInput( source_fileitem={ "storage": "local", "path": f"/{task_id}.mkv", "type": "file", }, target_storage="local", target_path="/media", requested_transfer_type="copy", ) def _plan_checkpoint(task_id: str) -> TransferPlanCheckpoint: """ 构造包含一个叶子文件计划的完整宿主检查点。 """ planning_input = _planning_input(task_id) return TransferPlanCheckpoint( planning_input=planning_input, target_storage="local", root_target_path="/media", final_target_path=f"/media/{task_id}.mkv", resolved_transfer_type="copy", items=(TransferPlanItem( sequence=0, source_fileitem=planning_input.source_fileitem, target_storage="local", target_path=f"/media/{task_id}.mkv", ),), ) def _plan_fingerprint(task_id: str) -> str: """ 返回测试冻结计划使用的 canonical 指纹。 """ return build_transfer_checkpoint_fingerprint( _plan_checkpoint(task_id).to_payload() ) def _seed_pending( factory, *, task_id: str = "task-1", lease_token: str = "lease-1", state: str = "planned", with_checkpoint: bool = True, ): """ 写入一条带有效租约与合法 planning checkpoint 的待执行任务。 """ planning_input = _planning_input(task_id) checkpoint = _plan_checkpoint(task_id) if with_checkpoint else None with factory() as session: session.add(TransferPending( task_id=task_id, storage="local", src_path=f"/{task_id}.mkv", created_at="2026-08-27 09:00:00", state=state, updated_at="2026-08-27 09:00:00", input_version=1, planning_input=planning_input.to_payload(), input_fingerprint=planning_input.fingerprint, checkpoint_version=checkpoint.schema_version if checkpoint else None, checkpoint_payload=checkpoint.to_payload() if checkpoint else None, planned_at="2026-08-27 09:00:00" if checkpoint else None, lease_owner="worker-1", lease_token=lease_token, lease_expires_at="2099-01-01 00:00:00.000000", heartbeat_at="2026-08-27 01:00:00.000000", attempt_count=1, execution_state="not_started", retry_generation=0, retry_count=0, settlement_revision=0, )) session.commit() def _seed_failed_receipt( factory, *, task_id: str = "task-1", execution_state: str = "failed", settlement_revision: int = 1, ) -> int: """ 写入带失败历史和步骤证据的无租约 durable 终态任务。 """ _seed_pending(factory, task_id=task_id) with factory() as session: pending = session.scalar( select(TransferPending).where(TransferPending.task_id == task_id) ) assert pending is not None history = TransferHistory( transfer_task_id=task_id, transfer_settlement_revision=settlement_revision, src=f"/{task_id}.mkv", src_storage="local", status=False, errmsg="terminal failure", ) session.add(history) session.flush() pending.execution_state = execution_state pending.settlement_revision = settlement_revision pending.terminal_history_id = history.id pending.lease_owner = None pending.lease_token = None pending.lease_expires_at = None session.add(TransferExecutionStep( task_id=task_id, operation_id=f"operation-{task_id}", checkpoint_fingerprint=_plan_fingerprint(task_id), ordinal=0, phase="transfer", kind="materialize_target", state="failed", attempt_token=None, attempt_count=1, intent_version=1, intent_payload={"path": f"/{task_id}.mkv"}, last_error="terminal failure", prepared_at="2026-08-27 09:00:00", updated_at="2026-08-27 09:30:00", )) session.commit() return history.id def _repository(factory, token_values: list[str] | None = None): """ 构造固定时钟与可预测 attempt token 的执行命令。 """ repository = TransactionalTransferExecutionRepository( factory, local_clock=lambda: datetime(2026, 8, 27, 9, 30, 0), lease_clock=lambda: datetime(2026, 8, 27, 1, 30, 0, tzinfo=timezone.utc), ) values = iter(token_values or ["attempt-1", "attempt-2", "attempt-3"]) return repository, TransferExecutionCommand( repository, attempt_token_factory=lambda: next(values), ) def _intent(*, task_id: str = "task-1", ordinal: int = 0) -> TransferStepIntent: """ 构造稳定且可重复计算身份的测试步骤意图。 """ planning_input = _planning_input(task_id) return TransferStepIntent.create( task_id=task_id, checkpoint_fingerprint=_plan_fingerprint(task_id), ordinal=ordinal, phase="transfer", kind="materialize_target", payload={ "source": planning_input.source_fileitem, "target_storage": "local", "target_path": f"/media/{task_id}.mkv", "transfer_type": "copy", }, ) @pytest.mark.parametrize("state", ["accepted", "planned"]) @pytest.mark.parametrize("retry_count", [0, 318]) @pytest.mark.parametrize("execution_state", ["not_started", "retry_wait"]) def test_unhandled_failure_is_counted_once_and_stops_recovery( execution_store, state, retry_count, execution_state, ): """ 补丁在步骤外抛异常时立即隔离，即使旧任务已反复领取也不再自动回放。 """ _seed_pending(execution_store, state=state, with_checkpoint=state == "planned") with execution_store() as session: pending = session.scalar(select(TransferPending)) pending.retry_count = retry_count pending.attempt_count = 319 pending.execution_state = execution_state session.commit() repository, _ = _repository(execution_store) error = "TransferChain has no attribute _TransferChain__finish_scrape_batch_task" assert repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error=error) assert not repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error=error) snapshot = repository.get_snapshot(task_id="task-1") assert snapshot.state is TransferExecutionState.MANUAL_REVIEW assert snapshot.retry_count == retry_count + 1 assert snapshot.last_error == error assert snapshot.steps == () view = TransferManualReviewQuery(repository).get(task_id="task-1") assert view.step.kind == "unhandled_execution_review" assert view.step.error == error assert TransferManualReviewQuery(repository).list().total == 1 admissions = TransactionalTransferAdmissionRepository(execution_store) assert admissions.claim_recoverable(owner_id="recovery", limit=10, lease_seconds=120) == [] with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending.attempt_count == 319 assert pending.lease_owner is None assert pending.lease_token is None assert pending.lease_expires_at is None assert pending.heartbeat_at is None assert pending.retry_due_at is None @pytest.mark.parametrize("state,token,expires", [ ("retry_wait", None, "2099-01-01 00:00:00.000000"), ("manual_review", "lease-1", "2099-01-01 00:00:00.000000"), ("failed", "lease-1", "2099-01-01 00:00:00.000000"), ("settling", "lease-1", "2099-01-01 00:00:00.000000"), ("running", "new-owner-token", "2099-01-01 00:00:00.000000"), ("running", "lease-1", "2000-01-01 00:00:00.000000"), ]) def test_unhandled_failure_cannot_overwrite_existing_disposition_or_lease( execution_store, state, token, expires, ): """ 步骤已决定重试或终态、结算中及租约失效时，不重复计数或覆盖新 owner。 """ _seed_pending(execution_store) with execution_store() as session: pending = session.scalar(select(TransferPending)) pending.execution_state = state pending.lease_token = token pending.lease_expires_at = expires pending.last_error = "previous evidence" session.commit() repository, _ = _repository(execution_store) assert not repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error="outer") with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending.execution_state == state assert pending.retry_count == 0 assert pending.last_error == "previous evidence" assert pending.lease_token == token assert session.scalar(select(TransferExecutionStep)) is None def test_unhandled_failure_isolation_rolls_back_with_review_evidence(execution_store, monkeypatch): """ 复核凭据写入失败时必须同时回滚状态、失败次数和租约释放。 """ _seed_pending(execution_store) repository, _ = _repository(execution_store) def fail_review(*_args, **_kwargs): """ 模拟复核凭据无法写入。 """ raise RuntimeError("review write failed") monkeypatch.setattr(repository, "_stage_unhandled_review", fail_review) with pytest.raises(RuntimeError, match="review write failed"): repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error="outer") with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending.execution_state == "not_started" assert pending.retry_count == 0 assert pending.lease_token == "lease-1" def test_unhandled_review_preserves_started_step_and_resumes_after_manual_confirmation(execution_store): """ 复核未记账异常后，真实 STARTED 仍需严格探测，诊断步骤不参与执行或结算。 """ _seed_pending(execution_store) repository, command = _repository(execution_store) prepared = command.prepare(task_id="task-1", lease_token="lease-1", intent=_intent()) started = command.begin(task_id="task-1", lease_token="lease-1", operation_id=prepared.operation_id) assert repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error="outer") snapshot = repository.get_snapshot(task_id="task-1") assert snapshot.steps == (started,) review = repository.get_manual_review(task_id="task-1") with pytest.raises(TransferExecutionConflictError, match="人工回滚"): command.resolve_manual_review( task_id="task-1", operation_id=review.step.operation_id, decision=TransferManualReviewDecision.APPLIED, actor="admin", reason="checked", result=TransferStepResult(payload={"success": True}), ) resolved = command.resolve_manual_review( task_id="task-1", operation_id=review.step.operation_id, decision=TransferManualReviewDecision.NOT_APPLIED, actor="admin", reason="plugin rolled back", ) assert resolved.state is TransferExecutionState.RETRY_WAIT assert repository.get_manual_review(task_id="task-1").review_revision == 1 admission = TransactionalTransferAdmissionRepository(execution_store).claim_task( task_id="task-1", owner_id="recovery", lease_seconds=120, ) command.prepare(task_id="task-1", lease_token=admission.lease_token, intent=_intent()) resumed = command.restart_after_not_applied( task_id="task-1", lease_token=admission.lease_token, step=started, evidence=TransferStepResult(payload={"observation": "not_applied"}), ) completed = command.complete( task_id="task-1", lease_token=admission.lease_token, step=resumed, result=TransferStepResult(payload={"success": True}), ) assert completed.state is TransferStepState.SUCCEEDED assert repository.isolate_unhandled_failure(task_id="task-1", lease_token=admission.lease_token, error="again") with execution_store() as session: assert len(session.scalars(select(TransferExecutionStep)).all()) == 2 snapshot = repository.get_snapshot(task_id="task-1") assert len(snapshot.steps) == 1 assert snapshot.steps[0].state is TransferStepState.SUCCEEDED assert snapshot.retry_count == 2 def test_stable_operation_and_checkpoint_identities_are_canonical(): """ 字段顺序不能改变 operation ID 或执行 checkpoint 指纹。 """ first = build_transfer_operation_id( task_id="task", checkpoint_fingerprint="plan", ordinal=2, phase="scrape", kind="write", intent_payload={"b": 2, "a": 1}, ) second = build_transfer_operation_id( task_id="task", checkpoint_fingerprint="plan", ordinal=2, phase="scrape", kind="write", intent_payload={"a": 1, "b": 2}, ) assert first == second assert build_transfer_checkpoint_fingerprint({"b": 2, "a": 1}) == ( build_transfer_checkpoint_fingerprint({"a": 1, "b": 2}) ) mutable_payload = {"path": "/original"} intent = TransferStepIntent.create( task_id="task", checkpoint_fingerprint="plan", ordinal=0, phase="transfer", kind="copy", payload=mutable_payload, ) mutable_payload["path"] = "/mutated" assert intent.payload == {"path": "/original"} @pytest.mark.parametrize( ("state", "with_checkpoint"), (("accepted", False), ("planned", False)), ) def test_prepare_rejects_task_without_executable_plan( execution_store, state, with_checkpoint, ): """ 接纳态或缺失完整 checkpoint 的任务不得进入外部步骤准备。 """ _seed_pending( execution_store, state=state, with_checkpoint=with_checkpoint, ) _, command = _repository(execution_store) with pytest.raises(TransferExecutionConflictError): command.prepare( task_id="task-1", lease_token="lease-1", intent=_intent(), ) with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending is not None assert pending.execution_state == "not_started" assert session.scalar(select(TransferExecutionStep)) is None @pytest.mark.parametrize("field", ["checkpoint_version", "checkpoint_payload", "planned_at"]) def test_prepare_rejects_incomplete_plan_fields(execution_store, field): """ 缺少计划冻结字段仍须拒绝执行，并准确指出计划检查点缺失。 """ _seed_pending(execution_store) with execution_store() as session: pending = session.scalar(select(TransferPending)) setattr(pending, field, None) session.commit() _, command = _repository(execution_store) with pytest.raises(TransferExecutionConflictError, match="计划检查点字段不完整"): command.prepare(task_id="task-1", lease_token="lease-1", intent=_intent()) with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending.execution_state == "not_started" assert session.scalar(select(TransferExecutionStep)) is None def test_snapshot_reports_inconsistent_execution_checkpoint_fields(execution_store): """ 执行检查点列不一致的提示应与计划冻结字段缺失区分。 """ _seed_pending(execution_store) with execution_store() as session: pending = session.scalar(select(TransferPending)) pending.execution_payload = {"outcome": "succeeded"} session.commit() repository, _ = _repository(execution_store) with pytest.raises(TransferExecutionConflictError, match="执行检查点字段不一致"): repository.get_snapshot(task_id="task-1") def test_prepare_rejects_noncanonical_plan_fingerprint(execution_store): """ 步骤 intent 必须精确绑定当前持久计划的 canonical 指纹。 """ _seed_pending(execution_store) _, command = _repository(execution_store) intent = TransferStepIntent.create( task_id="task-1", checkpoint_fingerprint="0" * 64, ordinal=0, phase="transfer", kind="materialize_target", payload=_intent().payload, ) with pytest.raises(TransferExecutionConflictError, match="整理任务记录已失效"): command.prepare( task_id="task-1", lease_token="lease-1", intent=intent, ) def test_prepare_rejects_forged_operation_id(execution_store): """ 调用方手工构造的伪 operation ID 不得绕过稳定身份计算。 """ _seed_pending(execution_store) _, command = _repository(execution_store) valid = _intent() forged = TransferStepIntent( operation_id="f" * 64, checkpoint_fingerprint=valid.checkpoint_fingerprint, ordinal=valid.ordinal, phase=valid.phase, kind=valid.kind, payload=valid.payload, ) with pytest.raises(TransferExecutionConflictError, match="整理任务记录已失效"): command.prepare( task_id="task-1", lease_token="lease-1", intent=forged, ) def test_prepare_rejects_arbitrary_intent_with_known_plan_fingerprint( execution_store, ): """ 已知计划指纹也不能构造计划未授权的操作类型或参数。 """ _seed_pending(execution_store) _, command = _repository(execution_store) arbitrary = TransferStepIntent.create( task_id="task-1", checkpoint_fingerprint=_plan_fingerprint("task-1"), ordinal=0, phase="transfer", kind="delete_move_source", payload={ "source": _planning_input("task-1").source_fileitem, "target_storage": "local", "target_path": "/media/task-1.mkv", }, ) with pytest.raises(TransferExecutionConflictError, match="整理任务记录已失效"): command.prepare( task_id="task-1", lease_token="lease-1", intent=arbitrary, ) def test_prepare_rejects_noncontiguous_ordinal(execution_store): """ 新步骤只能在完整既有序列尾部连续追加。 """ _seed_pending(execution_store) _, command = _repository(execution_store) with pytest.raises(TransferExecutionConflictError, match="整理任务记录已失效"): command.prepare( task_id="task-1", lease_token="lease-1", intent=_intent(ordinal=1), ) def test_stage_execution_running_cas_binds_exact_plan_identity(execution_store): """ running CAS 必须拒绝读取后已变化的准入状态或 checkpoint payload。 """ _seed_pending(execution_store) checkpoint = _plan_checkpoint("task-1") with execution_store() as session: stale = TransferPending.stage_execution_running( session, task_id="task-1", lease_token="lease-1", admission_state="planned", checkpoint_version=checkpoint.schema_version, checkpoint_payload={"schema_version": checkpoint.schema_version}, now_utc="2026-08-27 01:30:00.000000", updated_at="2026-08-27 09:30:00", ) assert stale == 0 current = TransferPending.stage_execution_running( session, task_id="task-1", lease_token="lease-1", admission_state="planned", checkpoint_version=checkpoint.schema_version, checkpoint_payload=checkpoint.to_payload(), now_utc="2026-08-27 01:30:00.000000", updated_at="2026-08-27 09:30:00", ) assert current == 1 def test_stage_execution_running_casts_postgresql_json_before_comparison( monkeypatch, ): """ PostgreSQL 的 JSON 计划比较必须转为 JSONB，避免缺少 json 等值操作符。 """ engine = create_mock_engine( "postgresql+psycopg://", lambda *_args, **_kwargs: None, ) statements = [] def capture_statement(_db, statement, execution_options=None): """ 捕获模型生成的 DML，检查生产 PostgreSQL 方言的 SQL。 """ del execution_options statements.append(statement) return 1 monkeypatch.setattr( "app.db.models.transferpending.execute_dml", capture_statement, ) payload = {"schema_version": 1, "items": []} with Session(engine) as session: updated = TransferPending.stage_execution_running( session, task_id="task-1", lease_token="lease-1", admission_state="planned", checkpoint_version=1, checkpoint_payload=payload, now_utc="2026-08-27 01:30:00.000000", updated_at="2026-08-27 09:30:00", ) assert updated == 1 assert len(statements) == 1 sql = str(statements[0].compile(dialect=postgresql.dialect())) assert "CAST(transferpending.checkpoint_payload AS JSONB)" in sql assert "CAST(%(param_1)s::JSONB AS JSONB)" in sql def test_success_path_persists_steps_and_execution_checkpoint(execution_store): """ 成功路径应保留每步证据，并提交可供唯一 durable writer 结算的检查点。 """ _seed_pending(execution_store) repository, command = _repository(execution_store) prepared = command.prepare( task_id="task-1", lease_token="lease-1", intent=_intent(), ) assert prepared.state is TransferStepState.PREPARED started = command.begin( task_id="task-1", lease_token="lease-1", operation_id=prepared.operation_id, ) assert started.attempt_token == "attempt-1" succeeded = command.complete( task_id="task-1", lease_token="lease-1", step=started, result=TransferStepResult(payload={"dest_exists": True}), ) checkpoint = TransferExecutionCheckpoint.create( payload={"outcome": "succeeded", "dest": "/media/task-1.mkv"}, operation_ids=(succeeded.operation_id,), ) snapshot = command.checkpoint( task_id="task-1", lease_token="lease-1", checkpoint=checkpoint, ) assert snapshot.state is TransferExecutionState.SETTLING with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending is not None assert pending.execution_fingerprint == checkpoint.fingerprint step = session.scalar(select(TransferExecutionStep)) assert step is not None and step.state == "succeeded" def test_provider_predecessor_remains_owned_after_host_plan_promotion( execution_store, monkeypatch, ): """ 真实准入冻结 provider 后可执行，回退升级宿主计划仍保留序号零证据。 """ admission_repository = TransactionalTransferAdmissionRepository(execution_store) planning_input = _planning_input("task-1") admitted = admission_repository.admit( storage="local", src_path="/task-1.mkv", planning_input=planning_input, ) task_id = admitted.task_id claimed = admission_repository.claim_task( task_id=task_id, owner_id="provider-worker", lease_seconds=3600, ) assert claimed is not None lease_token = claimed.lease_token provider = TransferProviderReference( plugin_id="provider-a", plugin_name="Provider A", ) invocation = TransferProviderInvocationSnapshot( fileitem=planning_input.source_fileitem, meta={"title": "Movie"}, meta_kind="MetaVideo", mediainfo={"title": "Movie"}, mediainfo_kind="MediaInfo", ) provider_checkpoint = TransferPlanCheckpoint( planning_input=planning_input, target_storage="", root_target_path="", final_target_path="", resolved_transfer_type="", items=(), resolved_meta=invocation.meta, resolved_meta_kind=invocation.meta_kind, resolved_mediainfo=invocation.mediainfo, resolved_mediainfo_kind=invocation.mediainfo_kind, legacy_transfer_providers=(provider,), provider_invocation=invocation, ) monkeypatch.setattr(admission_repository, "_now", lambda: "2026-08-27 09:00:00") admission_repository.checkpoint_plan( task_id=task_id, lease_token=lease_token, input_fingerprint=planning_input.fingerprint, checkpoint=provider_checkpoint, ) with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending is not None assert pending.state == "provider_pending" assert pending.planned_at == "2026-08-27 09:00:00" _, command = _repository(execution_store) provider_intent = TransferStepIntent.create( task_id=task_id, checkpoint_fingerprint=build_transfer_checkpoint_fingerprint( provider_checkpoint.to_payload() ), ordinal=0, phase="provider", kind="legacy_transfer_provider_sequence", payload={ "providers": [provider.to_payload()], "invocation": invocation.to_payload(), }, ) prepared = command.prepare( task_id=task_id, lease_token=lease_token, intent=provider_intent, ) with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending is not None assert pending.execution_state == "running" assert pending.state == "provider_pending" assert session.scalar(select(TransferExecutionStep)).state == "prepared" started = command.begin( task_id=task_id, lease_token=lease_token, operation_id=prepared.operation_id, ) provider_succeeded = command.complete( task_id=task_id, lease_token=lease_token, step=started, result=TransferStepResult(payload={"handled": False}), ) promoted_checkpoint = replace( _plan_checkpoint("task-1"), pre_execution_cleanup_completed=True, ) monkeypatch.setattr(admission_repository, "_now", lambda: "2026-08-27 09:01:00") admission_repository.checkpoint_plan( task_id=task_id, lease_token=lease_token, input_fingerprint=planning_input.fingerprint, checkpoint=promoted_checkpoint, ) with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending is not None assert pending.state == "planned" assert pending.planned_at == "2026-08-27 09:01:00" promoted_fingerprint = build_transfer_checkpoint_fingerprint( promoted_checkpoint.to_payload() ) host_intent = TransferStepIntent.create( task_id=task_id, checkpoint_fingerprint=promoted_fingerprint, ordinal=1, phase="transfer", kind="materialize_target", payload=_intent().payload, ) host_prepared = command.prepare( task_id=task_id, lease_token=lease_token, intent=host_intent, ) host_started = command.begin( task_id=task_id, lease_token=lease_token, operation_id=host_prepared.operation_id, ) host_succeeded = command.complete( task_id=task_id, lease_token=lease_token, step=host_started, result=TransferStepResult(payload={"dest_exists": True}), ) execution_checkpoint = TransferExecutionCheckpoint.create( payload={"outcome": "succeeded", "dest": "/media/task-1.mkv"}, operation_ids=( provider_succeeded.operation_id, host_succeeded.operation_id, ), ) snapshot = command.checkpoint( task_id=task_id, lease_token=lease_token, checkpoint=execution_checkpoint, ) assert snapshot.state is TransferExecutionState.SETTLING assert tuple(step.ordinal for step in snapshot.steps) == (0, 1) def test_checkpoint_rejects_operation_ids_out_of_ordinal_order(execution_store): """ 执行检查点必须按严格 ordinal 保存操作身份，集合相同也不能乱序。 """ _seed_pending(execution_store) _, command = _repository(execution_store) completed = [] for ordinal in range(2): prepared = command.prepare( task_id="task-1", lease_token="lease-1", intent=_intent(ordinal=ordinal), ) started = command.begin( task_id="task-1", lease_token="lease-1", operation_id=prepared.operation_id, ) completed.append(command.complete( task_id="task-1", lease_token="lease-1", step=started, result=TransferStepResult(payload={"ordinal": ordinal}), )) checkpoint = TransferExecutionCheckpoint.create( payload={"outcome": "succeeded", "dest": "/media/task-1.mkv"}, operation_ids=tuple(step.operation_id for step in reversed(completed)), ) with pytest.raises(TransferExecutionConflictError, match="整理任务状态已发生变化"): command.checkpoint( task_id="task-1", lease_token="lease-1", checkpoint=checkpoint, ) def test_checkpoint_rejects_corrupted_persisted_operation_id(execution_store): """ checkpoint 前必须重新计算每个持久步骤的 operation ID。 """ _seed_pending(execution_store) _, command = _repository(execution_store) prepared = command.prepare( task_id="task-1", lease_token="lease-1", intent=_intent(), ) started = command.begin( task_id="task-1", lease_token="lease-1", operation_id=prepared.operation_id, ) command.complete( task_id="task-1", lease_token="lease-1", step=started, result=TransferStepResult(payload={"dest_exists": True}), ) with execution_store() as session: step = session.scalar(select(TransferExecutionStep)) assert step is not None step.operation_id = "e" * 64 session.commit() checkpoint = TransferExecutionCheckpoint.create( payload={"outcome": "succeeded", "dest": "/media/task-1.mkv"}, operation_ids=("e" * 64,), ) with pytest.raises(TransferExecutionConflictError, match="整理任务记录已失效"): command.checkpoint( task_id="task-1", lease_token="lease-1", checkpoint=checkpoint, ) def test_checkpoint_rejects_noncontiguous_persisted_ordinals(execution_store): """ checkpoint 前必须拒绝缺口或从非零开始的持久步骤序列。 """ _seed_pending(execution_store) _, command = _repository(execution_store) prepared = command.prepare( task_id="task-1", lease_token="lease-1", intent=_intent(), ) started = command.begin( task_id="task-1", lease_token="lease-1", operation_id=prepared.operation_id, ) command.complete( task_id="task-1", lease_token="lease-1", step=started, result=TransferStepResult(payload={"dest_exists": True}), ) with execution_store() as session: step = session.scalar(select(TransferExecutionStep)) assert step is not None step.ordinal = 2 step.operation_id = build_transfer_operation_id( task_id=step.task_id, checkpoint_fingerprint=step.checkpoint_fingerprint, ordinal=step.ordinal, phase=step.phase, kind=step.kind, intent_payload=step.intent_payload, ) corrupted_operation_id = step.operation_id session.commit() checkpoint = TransferExecutionCheckpoint.create( payload={"outcome": "succeeded", "dest": "/media/task-1.mkv"}, operation_ids=(corrupted_operation_id,), ) with pytest.raises(TransferExecutionConflictError, match="整理任务正在被其他操作处理"): command.checkpoint( task_id="task-1", lease_token="lease-1", checkpoint=checkpoint, ) def test_retry_wait_resumes_same_failed_operation_with_new_attempt(execution_store): """ 到期重试必须复用 operation ID、保留失败证据并轮换 attempt token。 """ _seed_pending(execution_store) repository, command = _repository(execution_store) prepared = command.prepare( task_id="task-1", lease_token="lease-1", intent=_intent() ) started = command.begin( task_id="task-1", lease_token="lease-1", operation_id=prepared.operation_id, ) deferred = command.defer( task_id="task-1", lease_token="lease-1", step=started, error="destination unavailable", retry_due_at="2026-08-27 01:30:01.000000", evidence=TransferStepResult(payload={"applied": False}), ) assert deferred.state is TransferExecutionState.RETRY_WAIT assert deferred.retry_generation == 1 with execution_store() as session: claimed = TransferPending.claim_task( session, task_id="task-1", states=("planned",), owner_id="worker-2", lease_token="lease-2", now_time="2026-08-27 01:30:02.000000", lease_expires_at="2099-01-01 00:00:00.000000", updated_at="2026-08-27 09:30:02", ) session.commit() assert claimed == 1 resumed = command.resume_failed( task_id="task-1", lease_token="lease-2", step=deferred.steps[0], ) assert resumed.operation_id == prepared.operation_id assert resumed.attempt_token == "attempt-2" assert resumed.attempt_count == 2 assert resumed.result == TransferStepResult(payload={"applied": False}) assert resumed.last_error == "destination unavailable" def test_orphan_started_requires_observation_before_attempt_rotation(execution_store): """ 遗留 STARTED 只能凭 NOT_APPLIED 证据轮换 attempt，旧 attempt 随即失效。 """ _seed_pending(execution_store) repository, command = _repository(execution_store) prepared = command.prepare( task_id="task-1", lease_token="lease-1", intent=_intent() ) started = command.begin( task_id="task-1", lease_token="lease-1", operation_id=prepared.operation_id, ) observation = TransferOperationObservation( state=TransferOperationObservationState.NOT_APPLIED, evidence=TransferStepResult(payload={"dest_exists": False}), ) restarted = command.restart_after_not_applied( task_id="task-1", lease_token="lease-1", step=started, evidence=observation.evidence, ) assert restarted.attempt_token == "attempt-2" assert restarted.attempt_count == 2 with pytest.raises(TransferExecutionConflictError): repository.complete_step( task_id="task-1", lease_token="lease-1", operation_id=started.operation_id, attempt_token="attempt-1", result=TransferStepResult(payload={"stale": True}), ) def test_zero_side_effect_checkpoint_is_vacuously_complete(execution_store): """ 纯策略拒绝可在没有步骤行时提交带 skip_reason 的确定执行结果。 """ _seed_pending(execution_store) _, command = _repository(execution_store) checkpoint = TransferExecutionCheckpoint.create( payload={"outcome": "failed", "preview": True, "accepted": False}, operation_ids=(), skip_reason="preview", ) snapshot = command.checkpoint( task_id="task-1", lease_token="lease-1", checkpoint=checkpoint, ) assert snapshot.state is TransferExecutionState.SETTLING assert snapshot.checkpoint == checkpoint assert snapshot.steps == () def test_exhausted_step_builds_failure_checkpoint_and_keeps_lease(execution_store): """ 预算耗尽应原子建立失败结算检查点，并为 durable writer 保留 lease。 """ _seed_pending(execution_store) _, command = _repository(execution_store) prepared = command.prepare( task_id="task-1", lease_token="lease-1", intent=_intent() ) started = command.begin( task_id="task-1", lease_token="lease-1", operation_id=prepared.operation_id, ) snapshot = command.exhaust( task_id="task-1", lease_token="lease-1", step=started, error="retry budget exhausted", evidence=TransferStepResult(payload={"applied": False}), ) assert snapshot.state is TransferExecutionState.SETTLING assert snapshot.checkpoint is not None assert snapshot.checkpoint.payload["outcome"] == "failed" assert snapshot.checkpoint.payload["error"] == "retry budget exhausted" assert snapshot.checkpoint.operation_ids == (started.operation_id,) assert snapshot.steps[0].state is TransferStepState.FAILED with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending is not None assert pending.lease_token == "lease-1" def test_user_retry_is_single_generation_and_rejects_manual_review(execution_store): """ FAILED 用户重试只递增一次世代，重复请求幂等且人工复核必须拒绝。 """ _seed_pending(execution_store) with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending is not None pending.execution_state = "failed" pending.lease_owner = None pending.lease_token = None pending.lease_expires_at = None pending.terminal_history_id = 42 pending.retry_count = 3 session.commit() _, command = _repository(execution_store) first = command.request_retry( task_id="task-1", reason="用户确认目标未落地", requested_by="admin", ) repeated = command.request_retry( task_id="task-1", reason="重复点击", requested_by="admin", ) assert first.accepted and repeated.accepted assert first.retry_generation == repeated.retry_generation == 1 with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending is not None assert pending.execution_state == "retry_wait" assert pending.retry_count == 3 assert pending.terminal_history_id == 42 assert pending.retry_reason == "用户确认目标未落地" assert pending.retry_requested_by == "admin" pending.execution_state = "manual_review" session.commit() rejected = command.request_retry( task_id="task-1", reason="强制重试", requested_by="admin", ) assert not rejected.accepted assert rejected.state is TransferExecutionState.MANUAL_REVIEW assert "人工" in rejected.message @pytest.mark.parametrize("expires_at", ( None, "2026-08-27 01:29:59.000000", "2026-08-27 01:30:00.000000", )) def test_user_retry_clears_expired_claim_once_and_keeps_execution_evidence( execution_store, expires_at, ) -> None: """ 失败任务残留的失效 token 应允许直接重试，重复点击不重复增加世代。 """ history_id = _seed_failed_receipt(execution_store) with execution_store() as session: pending = session.scalar(select(TransferPending)) pending.lease_owner = "old-owner" pending.lease_token = "old-token" pending.lease_expires_at = expires_at session.commit() _, command = _repository(execution_store) first = command.request_retry(task_id="task-1", reason="重试", requested_by="admin") repeated = command.request_retry(task_id="task-1", reason="再次点击", requested_by="admin") assert first.accepted and repeated.accepted assert first.retry_generation == repeated.retry_generation == 1 with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending.execution_state == "retry_wait" assert pending.lease_token is None assert pending.lease_owner is None assert pending.lease_expires_at is None assert pending.heartbeat_at is None assert pending.retry_reason == "重试" assert session.scalar(select(TransferExecutionStep)) is not None assert session.get(TransferHistory, history_id).transfer_task_id == "task-1" def test_user_retry_preserves_active_failed_claim(execution_store) -> None: """ 失败状态仍持有有效租约时，重试不得撤销活动 owner 的执行资格。 """ history_id = _seed_failed_receipt(execution_store) with execution_store() as session: pending = session.scalar(select(TransferPending)) pending.lease_owner = "active-owner" pending.lease_token = "active-token" pending.lease_expires_at = "2026-08-27 01:30:01.000000" session.commit() _, command = _repository(execution_store) result = command.request_retry(task_id="task-1", reason="重试", requested_by="admin") assert result.accepted is False assert "正在处理中" in result.message with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending.execution_state == "failed" assert pending.lease_token == "active-token" assert pending.retry_generation == 0 assert session.scalar(select(TransferExecutionStep)) is not None assert session.get(TransferHistory, history_id).transfer_task_id == "task-1" @pytest.mark.parametrize("winner", ("heartbeat", "retry_claim")) def test_user_retry_cas_preserves_newly_committed_lease( execution_store, monkeypatch, winner, ) -> None: """ 并发续租或重试接管先提交时，迟到的重试不能清掉新 owner 或重复调度。 """ history_id = _seed_failed_receipt(execution_store) with execution_store() as session: pending = session.scalar(select(TransferPending)) pending.lease_owner = "old-owner" pending.lease_token = "old-token" pending.lease_expires_at = "2026-08-27 01:29:59.000000" session.commit() repository, command = _repository(execution_store) _, winning_command = _repository(execution_store) admissions = TransactionalTransferAdmissionRepository(execution_store) clear_claim = repository._clear_inactive_claim claimed = [] def commit_winning_lease(session, *, pending, now_utc): """ 在读取旧状态后由另一仓储提交续租或重试领取，再执行实际 CAS。 """ if winner == "heartbeat": monkeypatch.setattr(admissions, "_lease_now", lambda: datetime( 2026, 8, 27, 1, 29, 30, tzinfo=timezone.utc, )) claimed.append(admissions.heartbeat( task_id="task-1", lease_token="old-token", lease_seconds=120, )) else: assert winning_command.request_retry( task_id="task-1", reason="先提交的重试", requested_by="admin", ).accepted monkeypatch.setattr(admissions, "_lease_now", lambda: datetime( 2026, 8, 27, 1, 30, 0, tzinfo=timezone.utc, )) claimed.append(admissions.claim_task( task_id="task-1", owner_id="new-owner", lease_seconds=120, )) return clear_claim(session, pending=pending, now_utc=now_utc) monkeypatch.setattr(repository, "_clear_inactive_claim", commit_winning_lease) result = command.request_retry(task_id="task-1", reason="迟到的重试", requested_by="admin") assert result.accepted is False assert "正在处理中或状态已变化" in result.message assert len(claimed) == 1 and claimed[0] is not None with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending.lease_owner == claimed[0].lease_owner assert pending.lease_token == claimed[0].lease_token assert pending.lease_expires_at == claimed[0].lease_expires_at assert pending.retry_generation == (1 if winner == "retry_claim" else 0) assert session.scalar(select(TransferExecutionStep)) is not None assert session.get(TransferHistory, history_id).transfer_task_id == "task-1" def test_user_retry_failure_rolls_back_expired_claim(execution_store, monkeypatch) -> None: """ 调度写入失败必须回滚失效租约清理，保持原任务、世代和历史完整。 """ history_id = _seed_failed_receipt(execution_store) with execution_store() as session: pending = session.scalar(select(TransferPending)) pending.lease_owner = "old-owner" pending.lease_token = "old-token" pending.lease_expires_at = "2026-08-27 01:29:59.000000" session.commit() _, command = _repository(execution_store) def reject_retry(_self, **_kwargs): """ 模拟取得恢复权后持久重试调度写入失败。 """ raise RuntimeError("重试写入失败") monkeypatch.setattr(TransferPendingOper, "stage_request_execution_retry", reject_retry) with pytest.raises(RuntimeError, match="重试写入失败"): command.request_retry(task_id="task-1", reason="重试", requested_by="admin") with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending.execution_state == "failed" assert pending.lease_token == "old-token" assert pending.lease_expires_at == "2026-08-27 01:29:59.000000" assert pending.retry_generation == 0 assert session.scalar(select(TransferExecutionStep)) is not None assert session.get(TransferHistory, history_id).transfer_task_id == "task-1" def test_discard_failed_removes_execution_evidence_and_detaches_history( execution_store, ) -> None: """ 放弃匹配的 FAILED 任务应删除执行证据，并把历史恢复为普通记录。 """ history_id = _seed_failed_receipt(execution_store) repository, _ = _repository(execution_store) command = TransferRecoveryCommand(repository) result = command.discard_failed( task_id="task-1", history_id=history_id, settlement_revision=1, ) assert result.discarded is True assert result.state is TransferExecutionState.FAILED with execution_store() as session: assert session.scalar(select(TransferPending)) is None assert session.scalar(select(TransferExecutionStep)) is None history = session.scalar(select(TransferHistory)) assert history is not None assert history.id == history_id assert history.transfer_task_id is None assert history.transfer_settlement_revision is None @pytest.mark.parametrize("execution_state", ("retry_wait", "manual_review")) def test_discard_failed_rejects_nonfailed_execution_state( execution_store, execution_state, ) -> None: """ 等待重试和人工复核任务不得通过历史维护入口删除执行证据。 """ history_id = _seed_failed_receipt( execution_store, execution_state=execution_state, ) repository, _ = _repository(execution_store) command = TransferRecoveryCommand(repository) result = command.discard_failed( task_id="task-1", history_id=history_id, settlement_revision=1, ) assert result.discarded is False assert result.state is TransferExecutionState(execution_state) with execution_store() as session: assert session.scalar(select(TransferPending)) is not None assert session.scalar(select(TransferExecutionStep)) is not None history = session.scalar(select(TransferHistory)) assert history is not None assert history.transfer_task_id == "task-1" def test_discard_failed_rejects_stale_settlement_revision(execution_store) -> None: """ 陈旧页面携带的结算版本不能放弃已经变化的失败任务。 """ history_id = _seed_failed_receipt(execution_store, settlement_revision=3) repository, _ = _repository(execution_store) command = TransferRecoveryCommand(repository) result = command.discard_failed( task_id="task-1", history_id=history_id, settlement_revision=2, ) assert result.discarded is False assert "已变化" in result.message with execution_store() as session: pending = session.scalar(select(TransferPending)) history = session.scalar(select(TransferHistory)) assert pending is not None assert pending.settlement_revision == 3 assert history is not None assert history.transfer_settlement_revision == 3 def _discard_history_task(repository, history_id: int, method_name: str): """ 统一调用普通失败与损坏历史的显式放弃入口。 """ arguments = {"task_id": "task-1", "history_id": history_id} if method_name == "discard_failed": arguments["settlement_revision"] = 1 return getattr(TransferRecoveryCommand(repository), method_name)(**arguments) @pytest.mark.parametrize("active", [False, True]) def test_explicit_history_discard_does_not_decode_corrupt_planning_json(execution_store, active): """ 历史重整可放弃无法反序列化的旧记录，同时有效租约仍保护全部证据。 """ history_id = _seed_failed_receipt(execution_store) with execution_store() as session: session.execute(text( "UPDATE transferpending SET planning_input=:bad, checkpoint_payload=:bad, " "execution_payload=:bad, execution_state='corrupt-state', lease_token='old-token', " "lease_owner='old-owner', lease_expires_at=:expires" ), {"bad": "{", "expires": "2026-08-27 01:31:00.000000" if active else "2026-08-27 01:29:00.000000"}) session.commit() repository, _ = _repository(execution_store) result = TransferRecoveryCommand(repository).discard_corrupt_by_history( task_id="task-1", history_id=history_id, ) assert result.discarded is not active assert result.state is None with execution_store() as session: assert session.scalar(select(TransferPending.task_id)) == ("task-1" if active else None) assert session.scalar(select(TransferExecutionStep.task_id)) == ("task-1" if active else None) assert session.get(TransferHistory, history_id).transfer_task_id == ("task-1" if active else None) @pytest.mark.parametrize("method_name, state", ( ("discard_failed", "failed"), ("discard_corrupt_by_history", "failed"), ("discard_corrupt_by_history", "corrupt-state"), )) @pytest.mark.parametrize("expires_at", ( None, "2026-08-27 01:29:59.000000", "2026-08-27 01:30:00.000000", )) def test_discard_history_releases_expired_claim_and_preserves_history( execution_store, method_name, state, expires_at, ) -> None: """ 过期或缺失期限的残留 token 不得永久阻止用户放弃旧任务重新整理。 """ history_id = _seed_failed_receipt(execution_store) with execution_store() as session: pending = session.scalar(select(TransferPending)) pending.lease_owner = "old-owner" pending.lease_token = "old-token" pending.lease_expires_at = expires_at pending.execution_state = state session.commit() repository, _ = _repository(execution_store) result = _discard_history_task(repository, history_id, method_name) assert result.discarded is True with execution_store() as session: assert session.scalar(select(TransferPending)) is None assert session.scalar(select(TransferExecutionStep)) is None history = session.get(TransferHistory, history_id) assert history is not None assert history.transfer_task_id is None assert history.transfer_settlement_revision is None @pytest.mark.parametrize("method_name", ("discard_failed", "discard_corrupt_by_history")) def test_discard_history_preserves_active_claim_and_execution_evidence( execution_store, method_name, ) -> None: """ 有效租约必须保留，不能因用户放弃旧历史而让活动执行失去证据。 """ history_id = _seed_failed_receipt(execution_store) with execution_store() as session: pending = session.scalar(select(TransferPending)) pending.lease_owner = "active-owner" pending.lease_token = "active-token" pending.lease_expires_at = "2026-08-27 01:30:01.000000" session.commit() repository, _ = _repository(execution_store) result = _discard_history_task(repository, history_id, method_name) assert result.discarded is False with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending.lease_token == "active-token" assert pending.lease_expires_at == "2026-08-27 01:30:01.000000" assert session.scalar(select(TransferExecutionStep)) is not None assert session.get(TransferHistory, history_id).transfer_task_id == "task-1" @pytest.mark.parametrize("method_name", ("discard_failed", "discard_corrupt_by_history")) @pytest.mark.parametrize("winner", ("heartbeat", "retry_claim")) def test_discard_history_cas_preserves_newly_committed_lease( execution_store, monkeypatch, method_name, winner, ) -> None: """ 读到无活动租约后若续租或重试接管先提交，放弃必须保留新 owner。 """ history_id = _seed_failed_receipt(execution_store) if winner == "heartbeat": with execution_store() as session: pending = session.scalar(select(TransferPending)) pending.lease_owner = "old-owner" pending.lease_token = "old-token" pending.lease_expires_at = "2026-08-27 01:29:59.000000" session.commit() repository, _ = _repository(execution_store) _, retry_command = _repository(execution_store) admissions = TransactionalTransferAdmissionRepository(execution_store) clear_claim = repository._clear_inactive_claim claimed = [] def commit_winning_lease(session, *, pending, now_utc): """ 在旧投影读取与实际 CAS 之间提交另一条真实仓储续租或领取操作。 """ if winner == "heartbeat": # 心跳在到期前开始，但数据库提交晚于放弃入口读到旧期限。 monkeypatch.setattr(admissions, "_lease_now", lambda: datetime( 2026, 8, 27, 1, 29, 30, tzinfo=timezone.utc, )) claimed.append(admissions.heartbeat( task_id="task-1", lease_token="old-token", lease_seconds=120, )) else: assert retry_command.request_retry( task_id="task-1", reason="重新整理", requested_by="admin", ).accepted monkeypatch.setattr(admissions, "_lease_now", lambda: datetime( 2026, 8, 27, 1, 30, 0, tzinfo=timezone.utc, )) claimed.append(admissions.claim_task( task_id="task-1", owner_id="new-owner", lease_seconds=120, )) return clear_claim(session, pending=pending, now_utc=now_utc) monkeypatch.setattr(repository, "_clear_inactive_claim", commit_winning_lease) result = _discard_history_task(repository, history_id, method_name) assert result.discarded is False assert len(claimed) == 1 and claimed[0] is not None with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending.lease_owner == claimed[0].lease_owner assert pending.lease_token == claimed[0].lease_token assert pending.lease_expires_at == claimed[0].lease_expires_at assert session.scalar(select(TransferExecutionStep)) is not None assert session.get(TransferHistory, history_id).transfer_task_id == "task-1" @pytest.mark.parametrize("method_name", ("discard_failed", "discard_corrupt_by_history")) def test_discard_history_cleanup_failure_rolls_back_expired_claim( execution_store, monkeypatch, method_name, ) -> None: """ 清理步骤失败必须回滚租约、任务和历史，避免只解锁却留下半份证据。 """ history_id = _seed_failed_receipt(execution_store) with execution_store() as session: pending = session.scalar(select(TransferPending)) pending.lease_owner = "old-owner" pending.lease_token = "old-token" pending.lease_expires_at = "2026-08-27 01:29:59.000000" session.commit() repository, _ = _repository(execution_store) def reject_cleanup(_self, *, task_id): """ 模拟取得放弃权后，步骤证据删除失败。 """ assert task_id == "task-1" raise RuntimeError("步骤清理失败") monkeypatch.setattr(TransferExecutionStepOper, "stage_delete_task", reject_cleanup) with pytest.raises(RuntimeError, match="步骤清理失败"): _discard_history_task(repository, history_id, method_name) with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending.lease_token == "old-token" assert pending.lease_expires_at == "2026-08-27 01:29:59.000000" assert session.scalar(select(TransferExecutionStep)) is not None history = session.get(TransferHistory, history_id) assert history.transfer_task_id == "task-1" assert history.transfer_settlement_revision == 1 def test_manual_not_applied_decision_is_audited_and_schedules_same_step( execution_store, ) -> None: """ 人工判定未发生应无 lease 地恢复 FAILED，并只交给唯一调度器。 """ _seed_pending(execution_store) _, command = _repository(execution_store) prepared = command.prepare( task_id="task-1", lease_token="lease-1", intent=_intent() ) started = command.begin( task_id="task-1", lease_token="lease-1", operation_id=prepared.operation_id, ) manual = command.manual_review( task_id="task-1", lease_token="lease-1", step=started, error="external result unknown", ) assert manual.state is TransferExecutionState.MANUAL_REVIEW resolved = command.resolve_manual_review( task_id="task-1", operation_id=started.operation_id, decision=TransferManualReviewDecision.NOT_APPLIED, actor="admin", reason="目标与临时文件均不存在", result=TransferStepResult(payload={"dest_exists": False}), ) assert resolved.state is TransferExecutionState.RETRY_WAIT assert resolved.review_revision == 1 assert resolved.step.state is TransferStepState.FAILED with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending is not None assert pending.lease_token is None assert pending.reviewed_by == "admin" assert pending.review_decision == "not_applied" assert pending.review_reason == "目标与临时文件均不存在" with pytest.raises(TransferExecutionConflictError): command.resolve_manual_review( task_id="task-1", operation_id=started.operation_id, decision=TransferManualReviewDecision.NOT_APPLIED, actor="admin", reason="重复判定", ) def test_manual_review_discard_releases_only_matching_unleased_task(execution_store) -> None: """放弃人工复核只删除 durable 任务与步骤，绝不触碰外部文件。""" _seed_pending(execution_store) repository, command = _repository(execution_store) prepared = command.prepare(task_id="task-1", lease_token="lease-1", intent=_intent()) started = command.begin(task_id="task-1", lease_token="lease-1", operation_id=prepared.operation_id) command.manual_review( task_id="task-1", lease_token="lease-1", step=started, error="external result unknown", ) review = repository.get_manual_review(task_id="task-1") stale = command.discard_manual_review( task_id="task-1", operation_id="stale-operation", actor="admin", reason="stale", ) assert not stale.discarded assert repository.get_snapshot(task_id="task-1") is not None discarded = command.discard_manual_review( task_id="task-1", operation_id=review.step.operation_id, actor="admin", reason="cannot verify provider result", ) assert discarded.discarded assert repository.get_snapshot(task_id="task-1") is None with execution_store() as session: assert session.scalar(select(TransferExecutionStep)) is None def test_manual_applied_requires_result_and_failed_decision_is_rejected( execution_store, ) -> None: """ 人工判定已发生必须带结果证据，FAILED 不得绕过 lease durable 结算。 """ _seed_pending(execution_store) _, command = _repository(execution_store) prepared = command.prepare( task_id="task-1", lease_token="lease-1", intent=_intent() ) started = command.begin( task_id="task-1", lease_token="lease-1", operation_id=prepared.operation_id, ) command.manual_review( task_id="task-1", lease_token="lease-1", step=started, error="external result unknown", ) with pytest.raises(ValueError, match="结果证据"): command.resolve_manual_review( task_id="task-1", operation_id=started.operation_id, decision=TransferManualReviewDecision.APPLIED, actor="admin", reason="已确认目标存在", ) with pytest.raises(TransferExecutionConflictError, match="durable"): command.resolve_manual_review( task_id="task-1", operation_id=started.operation_id, decision=TransferManualReviewDecision.FAILED, actor="admin", reason="确认失败", ) resolved = command.resolve_manual_review( task_id="task-1", operation_id=started.operation_id, decision=TransferManualReviewDecision.APPLIED, actor="admin", reason="目标摘要匹配", result=TransferStepResult(payload={"dest_exists": True, "hash_match": True}), ) assert resolved.step.state is TransferStepState.SUCCEEDED assert resolved.step.result == TransferStepResult( payload={"dest_exists": True, "hash_match": True} )