From 3bd77e84027c1dc70b8cb690a133b514b84a49e6 Mon Sep 17 00:00:00 2001 From: infstellar Date: Thu, 24 Sep 2026 11:56:54 +0800 Subject: [PATCH] Support task outcome settlement and reservation deadlines --- api.py | 2 +- backend/operation_policy.py | 17 ++++++++++ backend/persistence/repositories/memes.py | 2 ++ backend/persistence/repositories/tasks.py | 2 ++ backend/services/tasks.py | 35 ++++++++++++++++---- backend/task_deadline.py | 18 +++++++++++ backend/tasks.py | 2 ++ tests/test_task_deadline.py | 39 +++++++++++++++++++++++ 8 files changed, 109 insertions(+), 8 deletions(-) create mode 100644 backend/task_deadline.py create mode 100644 tests/test_task_deadline.py diff --git a/api.py b/api.py index ff902f6..fe2bdc3 100644 --- a/api.py +++ b/api.py @@ -741,7 +741,7 @@ async def lifespan(app: FastAPI): raise RuntimeError("operation_grant_invalid") grant = association.grant try: - if association.state == "acquired": + if association.state == "acquired" and not gateway.uses_task_settlement: commit_result = gateway.commit(grant) if not commit_result.ok: raise OperationPolicyError(commit_result.reason or "operation_policy_unavailable", retry_at=commit_result.retry_at) diff --git a/backend/operation_policy.py b/backend/operation_policy.py index 719426e..0749a9f 100644 --- a/backend/operation_policy.py +++ b/backend/operation_policy.py @@ -403,6 +403,23 @@ class OperationPolicyGateway: self.policy = policy if policy is not None else UnavailableOperationPolicy() self.allow_all = allow_all + @property + def uses_task_settlement(self) -> bool: + """宿主提供任务结算扩展时，由任务终态决定 Agent 计量。""" + return callable(getattr(self.policy, "settle_task", None)) + + def prepare_task(self, grant: GrantRef) -> datetime: + """取得宿主预留的固定截止时间，供 Worker 保存到可信任务字段。""" + deadline = self.policy.prepare_task(grant) + if not isinstance(deadline, datetime) or deadline.tzinfo is None: + raise OperationPolicyError("agent_reservation_deadline_invalid") + return deadline + + def settle_task(self, session: Any, scope: ScopeContext, task_id: str) -> None: + """复用调用方事务，将任务最终状态交给宿主完成计量。""" + if self.uses_task_settlement: + self.policy.settle_task(session, scope, task_id) + @staticmethod def request(scope: ScopeContext | str, operation: str, idempotency_key: str, **kwargs: Any) -> OperationRequest: """从服务端事实构造 policy request，忽略客户端身份字段。""" diff --git a/backend/persistence/repositories/memes.py b/backend/persistence/repositories/memes.py index 5627158..68d585d 100644 --- a/backend/persistence/repositories/memes.py +++ b/backend/persistence/repositories/memes.py @@ -19,6 +19,7 @@ from backend.image_naming import normalize_display_name, normalize_extension, sa from backend.paths import validate_business_storage_key from backend.persistence.engine import DatabaseError from backend.persistence.models import Meme, ScopeContext, StorageOperation, Task, utcnow +from backend.task_deadline import ensure_task_deadline ImageSort = Literal["name_asc", "updated_desc"] @@ -193,6 +194,7 @@ class MemeRepository: task = self.session.scalar(select(Task).where(Task.scope_id == self.scope.scope_id, Task.id == task_id).with_for_update()) if task is None or task.status != "running" or task.claim_generation != claim_generation or task.lease_owner != owner or task.lease_expires_at is None or task.lease_expires_at <= now: raise DatabaseError("claim_expired") + ensure_task_deadline(task.payload or {}, now) if expected_revision is not None and record.revision != expected_revision: raise DatabaseError("target_changed") if expected_sha256 is not None and record.sha256 != expected_sha256: diff --git a/backend/persistence/repositories/tasks.py b/backend/persistence/repositories/tasks.py index d3ff42f..fd4e239 100644 --- a/backend/persistence/repositories/tasks.py +++ b/backend/persistence/repositories/tasks.py @@ -18,6 +18,7 @@ from sqlalchemy.orm import Session from backend.agent_resume import append_error_history, append_task_error_history, normalize_identifier, sanitize_error from executor.agent_limits import validate_agent_concurrency, validate_agent_concurrency_at_most from backend.persistence.engine import DatabaseError +from backend.task_deadline import ensure_task_deadline from backend.persistence.models import ( Meme, ImageProcessingJob, @@ -1561,6 +1562,7 @@ class TaskRepository: ) if task is None: return False + ensure_task_deadline(task.payload or {}, now) task.status = "succeeded" task.progress = 1.0 task.message = "任务完成" diff --git a/backend/services/tasks.py b/backend/services/tasks.py index 3bac352..81db363 100644 --- a/backend/services/tasks.py +++ b/backend/services/tasks.py @@ -1043,7 +1043,7 @@ class PostgresTaskService: return False def _commit_agent_grant(self, claim: Task, payload: dict[str, Any]) -> None: - """在 Agent 外部执行前幂等提交服务端 grant。""" + """校验 Agent grant，并保存宿主预留期限或提交执行授权。""" operation_policy = getattr(self, "_operation_policy", None) grant_store = getattr(self, "_grant_store", None) if claim.task_type != "meme_context_generation" or operation_policy is None or grant_store is None: @@ -1115,6 +1115,12 @@ class PostgresTaskService: # 事实恢复；兼容测试夹具没有该扩展点时仍使用当前内存 payload。 if mode == "standalone" and callable(getattr(self, "_persist_claim_payload_updates", None)): self._persist_claim_payload_updates(claim, {"agent_grant_key": logical_key}) + if operation_policy.uses_task_settlement: + deadline = operation_policy.prepare_task(association.grant) + updates = {"agent_reservation_expires_at": deadline.isoformat()} + self._persist_claim_payload_updates(claim, updates) + payload.update(updates) + return try: result = self._operation_policy.commit(association.grant) except OperationPolicyError: @@ -1666,17 +1672,16 @@ class PostgresTaskService: # 先把已校验的 snapshot 摘要绑定到当前 attempt；即使后续 # grant 或 OpenCode 失败，恢复器也只能复用这一份事实。 self._image_attempt_state(claim, task_payload, "prepared") - # 图片阶段均可能触发外部模型或持久副作用；Agent 先完成 - # grant commit，再进入外部执行窗口，恢复者才能区分计量边界。 + # 宿主结算策略可保留预留，外部执行阶段单独记录恢复事实。 self._commit_agent_grant(claim, task_payload) task_payload["_agent_grant_committed"] = True - if claim.task_type == "meme_context_generation": + if claim.task_type == "meme_context_generation" and not (self._operation_policy is not None and self._operation_policy.uses_task_settlement): self._image_attempt_state(claim, task_payload, "grant_committed") # 恢复者无法证明结果时必须收束 unknown_execution。 self._image_attempt_state(claim, task_payload, "external_started") result = handler(task_payload, progress) except Exception as exc: # noqa: BLE001 - if not task_payload.get("_agent_grant_committed"): + if not task_payload.get("_agent_grant_committed") and not (self._operation_policy is not None and self._operation_policy.uses_task_settlement): self._release_uncommitted_agent_grant(claim, task_payload) if isinstance(exc, OperationPolicyError): code = exc.code @@ -1711,6 +1716,8 @@ class PostgresTaskService: max_seconds=self.resume_max_backoff_seconds, ) retry = code not in { + "agent_reservation_expired", + "agent_reservation_deadline_invalid", "target_changed", "agent_output_schema_invalid", "agent_output_invalid_json", @@ -1796,7 +1803,16 @@ class PostgresTaskService: # 只有当前 claim 仍有效时才写入任务终态和 Meme provenance。 self._image_attempt_state(claim, task_payload, "completed") audit_result = self._with_reverse_image_audit(task_id, result, write_provenance=False) - self._fenced_success(task_id, generation, audit_result) + try: + self._fenced_success(task_id, generation, audit_result) + except DatabaseError as exc: + if str(exc) != "agent_reservation_expired": + raise + self._fenced_failure( + task_id, generation, message="额度预留已到期", + error={"error": "agent_reservation_expired", "message": "额度预留已达到 24 小时"}, + retry=False, + ) finally: heartbeat_stop.set() self._maybe_finalize(task_id) @@ -1836,7 +1852,10 @@ class PostgresTaskService: with self.resources.environment(self.scope.scope_id) as environment: complete = getattr(environment.tasks, "complete_fenced_with_provenance", None) if callable(complete): - return bool(complete(task_id, generation, self.owner, result=result)) + changed = bool(complete(task_id, generation, self.owner, result=result)) + if changed and self._operation_policy is not None: + self._operation_policy.settle_task(environment.uow.session, self.scope, task_id) + return changed # 兼容尚未提供原子扩展的宿主 repository；标准 PostgreSQL # repository 始终走上面的单事务路径。 changed = environment.tasks.update_fenced( @@ -1869,6 +1888,8 @@ class PostgresTaskService: session_id=session_id, executor_attempt_id=executor_attempt_id, ) + if changed and not _should_retry and self._operation_policy is not None: + self._operation_policy.settle_task(environment.uow.session, self.scope, task_id) # 当前执行线程的 ``finally`` 会在释放本地调度标记后统一扫描 queued # 任务。这里不能提前按 task_id 单独提交新的 future：preclaimed 的 # 兼容调用可能尚未登记调度标记，会让同一任务在旧 claim 收束事务刚 diff --git a/backend/task_deadline.py b/backend/task_deadline.py new file mode 100644 index 0000000..c97e0a1 --- /dev/null +++ b/backend/task_deadline.py @@ -0,0 +1,18 @@ +# 任务持久化边界共用的执行期限校验，期限由宿主策略写入可信 Task payload。 + +from datetime import datetime +from typing import Any, Mapping + +from backend.persistence.engine import DatabaseError + + +def ensure_task_deadline(payload: Mapping[str, Any], now: datetime) -> None: + """在语境保存和任务成功提交前检查宿主给出的绝对期限。""" + value = payload.get("agent_reservation_expires_at") + if value is None: + return + deadline = datetime.fromisoformat(value) + if deadline.tzinfo is None: + raise DatabaseError("agent_reservation_deadline_invalid") + if now >= deadline: + raise DatabaseError("agent_reservation_expired") diff --git a/backend/tasks.py b/backend/tasks.py index 0d4c876..9dbf705 100644 --- a/backend/tasks.py +++ b/backend/tasks.py @@ -156,6 +156,8 @@ STABLE_TASK_ERRORS = { "operation_limit_exceeded", "operation_policy_unavailable", "operation_grant_invalid", + "agent_reservation_expired", + "agent_reservation_deadline_invalid", "blocked", "agent_provider_rate_limited", "agent_provider_server_error", diff --git a/tests/test_task_deadline.py b/tests/test_task_deadline.py new file mode 100644 index 0000000..49ebae7 --- /dev/null +++ b/tests/test_task_deadline.py @@ -0,0 +1,39 @@ +# 检查宿主预留期限在任务持久化边界的判定，覆盖到期瞬间及时间区域。 + +from datetime import datetime, timedelta, timezone + +import pytest + +from backend.persistence.engine import DatabaseError +from backend.task_deadline import ensure_task_deadline + + +def test_reservation_deadline_boundary() -> None: + """到期前允许提交，到期瞬间及之后拒绝提交。""" + deadline = datetime(2026, 9, 25, tzinfo=timezone.utc) + payload = {"agent_reservation_expires_at": deadline.isoformat()} + ensure_task_deadline(payload, deadline - timedelta(microseconds=1)) + for now in (deadline, deadline + timedelta(seconds=1)): + with pytest.raises(DatabaseError, match="agent_reservation_expired"): + ensure_task_deadline(payload, now) + + +def test_reservation_deadline_timezone() -> None: + """带不同时间区域的时刻按同一绝对期限判断。""" + payload = {"agent_reservation_expires_at": "2026-09-25T08:00:00+08:00"} + with pytest.raises(DatabaseError, match="agent_reservation_expired"): + ensure_task_deadline(payload, datetime(2026, 9, 25, tzinfo=timezone.utc)) + + +def test_reservation_deadline_requires_timezone() -> None: + """宿主提供缺少时间区域的期限时明确报错。""" + with pytest.raises(DatabaseError, match="agent_reservation_deadline_invalid"): + ensure_task_deadline( + {"agent_reservation_expires_at": "2026-09-25T00:00:00"}, + datetime(2026, 9, 25, tzinfo=timezone.utc), + ) + + +def test_task_without_reservation_deadline() -> None: + """没有宿主预留期限的任务可以使用原有执行条件。""" + ensure_task_deadline({}, datetime.now(timezone.utc))