Support task outcome settlement and reservation deadlines

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)

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，忽略客户端身份字段。"""

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:

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 = "任务完成"

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 收束事务刚

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")

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",

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))