Resume verified Agent sessions after graceful service shutdown

api.py

@@ -906,6 +906,8 @@ async def lifespan(app: FastAPI):
                 target_unchanged=True,
                 grant_state="committed" if grant is not None else None,
             )
+            if exc.code == "service_shutdown" and exc.process_reaped is not True:
+                decision = ResumeDecision(False, "unknown_execution", False)
             if not resume_enabled:
                 # rollout 关闭时保留旧任务级 retry，但不得把 session 标成可续跑；
                 # 否则开关热切换或任务详情会误把旧失败当成恢复目标。

backend/agent_executor.py

@@ -85,6 +85,8 @@ _KNOWN_TASK_ERRORS = frozenset(
         "agent_maximum_analysis_depth_exceeded",
         "agent_timeout",
         "task_interrupted",
+        "service_shutdown",
+        "agent_attempt_metadata_write_failed",
         "agent_process_failed",
         "unknown_execution",
         "agent_output_invalid_json",
@@ -329,11 +331,18 @@ class AgentExecutorClient:
             trigger_reason=trigger_reason,
         )
 
-    def _wait_for_terminal(self, response: ExecutorTaskResponse, *, task_id: str, executor_task_id: str, timeout_seconds: int) -> ExecutorTaskResponse:
+    def _wait_for_terminal(self, response: ExecutorTaskResponse, *, task_id: str, executor_task_id: str, timeout_seconds: int, shutdown_requested: Callable[[], bool] | None = None) -> ExecutorTaskResponse:
         """轮询同步提交的非终态响应，超时后只取消当前任务。"""
         deadline = time.monotonic() + max(5, int(timeout_seconds) + 10)
         poll_delay = 0.2
+        shutdown_sent = False
         while response.status in _PENDING_STATUSES:
+            # 提交和关闭可能并发；提交返回后由原线程补发中断，并继续等待终态。
+            if not shutdown_sent and shutdown_requested is not None and shutdown_requested():
+                response = self._for_task(self.cancel(executor_task_id, reason="service_shutdown"), task_id)
+                shutdown_sent = True
+                if response.status not in _PENDING_STATUSES:
+                    break
             remaining = deadline - time.monotonic()
             if remaining <= 0:
                 cancelled_response: ExecutorTaskResponse | None = None
@@ -426,6 +435,7 @@ class AgentExecutorClient:
         model_capability: str | None = None,
         visual_snapshot_sha256: str | None = None,
         analysis_policy: dict[str, object] | None = None,
+        shutdown_requested: Callable[[], bool] | None = None,
     ) -> ExecutorTaskResponse:
         """提交绑定模型 capability 的独立 executor attempt，并可按明确 session 续跑。"""
         timeout_value = int(timeout_seconds)
@@ -452,7 +462,7 @@ class AgentExecutorClient:
                     "image_relative_path": image_relative_path,
                     "reverse_image_policy": reverse_image_policy,
                     "timeout_seconds": timeout_value,
-                    "wait": True,
+                    "wait": False,
                     **({"session_id": session_id} if session_id else {}),
                     **({"resume_of_attempt_id": resume_of_attempt_id} if resume_of_attempt_id else {}),
                     **({"processing_config_hash": processing_config_hash} if processing_config_hash else {}),
@@ -501,7 +511,7 @@ class AgentExecutorClient:
         # 的轮询兼容；新协议总会返回显式 executor_attempt_id。
         executor_task_id = response.executor_attempt_id or task_id
         try:
-            response = self._wait_for_terminal(response, task_id=task_id, executor_task_id=executor_task_id, timeout_seconds=timeout_value)
+            response = self._wait_for_terminal(response, task_id=task_id, executor_task_id=executor_task_id, timeout_seconds=timeout_value, shutdown_requested=shutdown_requested)
         except AgentExecutorError as exc:
             raise AgentExecutorError(
                 exc.code,
@@ -559,9 +569,12 @@ class AgentExecutorClient:
             raise AgentExecutorError("agent_executor_invalid_response", "Agent executor 任务状态无效")
         return response
 
-    def cancel(self, task_id: str) -> ExecutorTaskResponse:
+    def cancel(self, task_id: str, *, reason: str = "task_interrupted") -> ExecutorTaskResponse:
         """取消指定任务，超时或服务关闭时调用。"""
-        _status, value = self._request("POST", f"/v1/tasks/{quote(task_id, safe='')}/cancel", timeout=min(10, self.timeout))
+        if reason not in {"task_interrupted", "service_shutdown"}:
+            raise ValueError("invalid_interruption_reason")
+        action = "interrupt-for-shutdown" if reason == "service_shutdown" else "cancel"
+        _status, value = self._request("POST", f"/v1/tasks/{quote(task_id, safe='')}/{action}", timeout=min(10, self.timeout))
         return self._for_executor_attempt(self._response(value), task_id)
 
     def status(self, task_id: str) -> ExecutorTaskResponse:

backend/agent_resume.py

@@ -41,6 +41,7 @@ RESUME_RETRYABLE_ERRORS = frozenset(
         "agent_connection_interrupted",
         "agent_process_failed",
         "agent_executor_unavailable",
+        "service_shutdown",
     }
 )
 RESUME_UNSAFE_ERRORS = frozenset(
@@ -113,6 +114,8 @@ def agent_failure_requires_unknown(
         return True
     # provider/进程错误同样发生在 Agent 调用边界之后；没有 session 就无法证明
     # 请求是否已经生效，不能因恢复标识缺失而退化为一次全新的外部调用。
+    if normalized == "service_shutdown":
+        return False
     if normalized in RESUME_RETRYABLE_ERRORS:
         # 仅有 session 字符串还不足以证明它已经和当前失败 attempt 绑定；
         # handler 必须先把带 fencing 的 attempt 事实成功写入数据库。

backend/application_lifecycle.py

@@ -650,17 +650,22 @@ async def shutdown_lifecycle(
         except BaseException as exc:  # noqa: BLE001 - 扩展失败不能跳过核心资源关闭
             errors.append(("shutdown.extension", exc))
     opencode = getattr(app.state, "opencode", None)
+    image_workers = list(getattr(app.state, "image_processing_workers", {}).values())
+    for image_worker in image_workers:
+        _close_resource(image_worker, "stop_claiming", errors, "shutdown.image_worker.stop_claiming")
+    if runtime is not None:
+        _close_resource(runtime.worker_manager, "stop_claiming", errors, "shutdown.worker_manager.stop_claiming")
     _close_resource(opencode, "shutdown", errors, "shutdown.opencode")
-    for image_worker in reversed(list(getattr(app.state, "image_processing_workers", {}).values())):
+    for image_worker in reversed(image_workers):
         _close_resource(image_worker, "shutdown", errors, "shutdown.image_worker")
     if runtime is not None:
-        _close_resource(runtime.factory, "shutdown", errors, "shutdown.factory")
-        _close_resource(runtime.worker_manager, "shutdown", errors, "shutdown.worker_manager")
         if runtime.shared_worker_executor is not None:
             try:
                 runtime.shared_worker_executor.shutdown(wait=True, cancel_futures=True)
             except BaseException as exc:  # noqa: BLE001 - 线程池失败不能阻止 Engine 收束
                 errors.append(("shutdown.executor", exc))
+        _close_resource(runtime.worker_manager, "shutdown", errors, "shutdown.worker_manager")
+        _close_resource(runtime.factory, "shutdown", errors, "shutdown.factory")
     _close_resource(setup.engine, "dispose", errors, "shutdown.engine")
     _clear_lifecycle_state(app, managed_factory=not setup.custom_factory)
     if primary_error is not None:

backend/image_processing.py

@@ -2031,6 +2031,8 @@ class ImageProcessingWorker:
         """认领 job，按阶段顺序创建或执行一个叶子 Task。"""
         reschedule = False
         try:
+            if self._stopped.is_set():
+                return
             current_job = self.jobs.get(job_id)
             now = utcnow()
             if current_job is not None and current_job.status == "running" and current_job.lease_owner == self.owner and current_job.lease_expires_at is not None and current_job.lease_expires_at > now:
@@ -2510,15 +2512,21 @@ class ImageProcessingWorker:
             except Exception as exc:  # noqa: BLE001 - 恢复扫描不能终止应用
                 logger.info("image_processing_reconcile_failed scope=%s error=%s", self.scope.scope_id, type(exc).__name__)
 
-    def shutdown(self) -> None:
-        """停止新 job 认领并等待图片叶子任务退出，再释放线程池。"""
+    def stop_claiming(self) -> None:
+        """停止安排 Job 和叶子 Task，保持正在运行任务的心跳。"""
         self._stopped.set()
-        if self._reconcile_thread is not None:
-            self._reconcile_thread.join(timeout=2)
         if self._task_runner is not None:
-            self._task_runner.shutdown()
+            self._task_runner.stop_claiming()
+        if self._reconcile_thread is not None:
+            self._reconcile_thread.join()
+
+    def shutdown(self) -> None:
+        """等待图片任务保存状态，再处理剩余租约并释放资源。"""
+        self.stop_claiming()
         # 线程仍可能持有 PostgreSQL session；数据库连接池必须在它们退出后再销毁。
         self.executor.shutdown(wait=True, cancel_futures=True)
+        if self._task_runner is not None:
+            self._task_runner.shutdown()
 
 
 class SingleImageEmbeddingService:

backend/opencode.py

@@ -6,6 +6,8 @@ sidecar 的读取、指纹复核和写入始终由上层元数据服务完成。
 
 from __future__ import annotations
 
+from loguru import logger
+
 import hashlib
 import io
 import json
@@ -625,6 +627,8 @@ class OpenCodeRunner:
             "model_broker_endpoint_invalid",
             "agent_timeout",
             "task_interrupted",
+            "service_shutdown",
+            "agent_attempt_metadata_write_failed",
             "agent_process_failed",
             "unknown_execution",
             "agent_output_invalid_json",
@@ -1801,6 +1805,7 @@ class OpenCodeRunner:
                         workspace_capability=capability,
                         model_capability=model_capability,
                         analysis_policy=analysis_policy,
+                        shutdown_requested=self._closing.is_set,
                     )
                     if response.executor_attempt_id:
                         self._remember_executor_attempt(task_id, response.executor_attempt_id)
@@ -1913,9 +1918,9 @@ class OpenCodeRunner:
                 timed_out = False
                 interrupted = False
                 try:
-                    if self._take_pre_cancelled(task_id):
+                    interrupted = self._take_pre_cancelled(task_id)
+                    if interrupted or self._closing.is_set():
                         self._terminate(process)
-                        interrupted = True
                     else:
                         # stdout/stderr 已重定向到临时文件；communicate 只等待进程。
                         process.communicate(timeout=self.settings.opencode_timeout_seconds)
@@ -1953,7 +1958,9 @@ class OpenCodeRunner:
                 if not session_id:
                     session_id = resume_session_id
                 if interrupted:
-                    raise OpenCodeError("task_interrupted", "Agent 任务已取消", session_id=session_id, executor_attempt_id=local_executor_attempt_id)
+                    raise OpenCodeError("task_interrupted", "Agent 任务已取消", session_id=session_id, executor_attempt_id=local_executor_attempt_id, process_reaped=process.poll() is not None)
+                if self._closing.is_set() and process.returncode != 0:
+                    raise OpenCodeError("service_shutdown", "服务正常关闭，Agent 执行已中断", session_id=session_id, executor_attempt_id=local_executor_attempt_id, process_reaped=process.poll() is not None)
                 if timed_out:
                     raise OpenCodeError("agent_timeout", "OpenCode 执行超时", session_id=session_id, executor_attempt_id=local_executor_attempt_id)
                 if process.returncode != 0:
@@ -2017,13 +2024,12 @@ class OpenCodeRunner:
         """终止当前受管理进程，供应用生命周期收束调用。"""
         self._closing.set()
         with self._process_lock:
-            self._cancelled_task_ids.update(self._active_task_ids)
             executor_business_ids = list(self._active_executor_task_ids)
         for task_id in executor_business_ids:
             try:
-                self.executor.cancel(self.executor.attempt_id_for(task_id) or task_id)
-            except AgentExecutorError:
-                pass
+                self.executor.cancel(self.executor.attempt_id_for(task_id) or task_id, reason="service_shutdown")
+            except AgentExecutorError as exc:
+                logger.warning("停机中断请求失败：task={} error={}", task_id, exc.code)
         with self._process_lock:
             processes = list(self._processes)
             if self._process is not None and self._process not in processes:

backend/persistence/repositories/tasks.py

@@ -1462,7 +1462,7 @@ class TaskRepository:
         self.session.flush()
         return True
 
-    def fail_fenced(self, task_id: str, claim_generation: int, owner: str, *, error: dict[str, Any], message: str, retry: bool = True, result: Any | None = None, retry_delay_seconds: int = 0, resume_available: bool | None = None, resume_reason: str | None = None, session_id: str | None = None, executor_attempt_id: str | None = None) -> tuple[bool, bool]:
+    def fail_fenced(self, task_id: str, claim_generation: int, owner: str, *, error: dict[str, Any], message: str, retry: bool = True, result: Any | None = None, retry_delay_seconds: int = 0, resume_available: bool | None = None, resume_reason: str | None = None, session_id: str | None = None, executor_attempt_id: str | None = None, resume_max_attempts: int = 0) -> tuple[bool, bool]:
         """在 fencing 条件下失败或重新排队任务，并追加有限错误历史。"""
         now = utcnow()
         task = self.session.scalar(select(Task).where(Task.scope_id == self.scope.scope_id, Task.id == task_id).with_for_update())
@@ -1491,7 +1491,13 @@ class TaskRepository:
             task.resume_reason = resume_reason
             if task.resume_available and task.resume_started_at is None:
                 task.resume_started_at = now
-        should_retry = bool(retry and task.attempt_count < task.max_attempts)
+        # 已验证 session 使用独立预算，图片阶段的 max_attempts 保持原值。
+        resumable_agent = bool(
+            task.task_type == "meme_context_generation"
+            and task.resume_available
+            and task.resume_attempt_count < resume_max_attempts
+        )
+        should_retry = bool(retry and (resumable_agent or task.attempt_count < task.max_attempts))
         if should_retry:
             task.status = "queued"
             task.available_at = now + timedelta(seconds=max(0, min(int(retry_delay_seconds), 3600)))

backend/services/tasks.py

@@ -1536,6 +1536,8 @@ class PostgresTaskService:
         """执行已认领任务并以 claim generation fencing 写回终态。"""
         claim = preclaimed
         try:
+            if claim is None and self._stopped.is_set():
+                return
             if claim is None:
                 with self.resources.environment(self.scope.scope_id) as environment:
                     queued_record = environment.tasks.get(task_id)
@@ -1740,6 +1742,7 @@ class PostgresTaskService:
                     max_seconds=self.resume_max_backoff_seconds,
                 )
                 retry = code not in {
+                    "agent_attempt_metadata_write_failed",
                     "agent_reservation_expired",
                     "agent_reservation_deadline_invalid",
                     "target_changed",
@@ -1811,6 +1814,8 @@ class PostgresTaskService:
                     "task_cancelled",
                 }
                 audit_result = self._with_reverse_image_audit(task_id, None, write_provenance=False)
+                if code == "service_shutdown":
+                    retry = self.resume_enabled and resume_available
                 self._fenced_failure(
                     task_id,
                     generation,
@@ -1841,6 +1846,7 @@ class PostgresTaskService:
                         )
             finally:
                 heartbeat_stop.set()
+                heartbeat_thread.join()
             self._maybe_finalize(task_id)
         finally:
             if self._worker_manager is not None:
@@ -1913,6 +1919,7 @@ class PostgresTaskService:
                 resume_reason=resume_reason,
                 session_id=session_id,
                 executor_attempt_id=executor_attempt_id,
+                resume_max_attempts=self.resume_max_attempts if self.resume_enabled else 0,
             )
             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)
@@ -2116,12 +2123,18 @@ class PostgresTaskService:
             except Exception:
                 pass
 
-    def shutdown(self) -> None:
-        """停止新认领并将本 Worker 仍持有的任务标记为可诊断中断。"""
+    def stop_claiming(self) -> None:
+        """停止认领新 Task，保留当前执行线程的租约和心跳。"""
         if self._worker_manager is not None:
             return
         self._stopped.set()
+
+    def shutdown(self) -> None:
+        """等待自有线程退出，再处理未完成的任务记录。"""
+        if self._worker_manager is not None:
+            return
+        self.stop_claiming()
+        if self._owns_executor:
+            self._executor.shutdown(wait=True, cancel_futures=True)
         with self.resources.environment(self.scope.scope_id) as environment:
             environment.tasks.interrupt_owner(self.owner)
-        if self._owns_executor:
-            self._executor.shutdown(wait=False, cancel_futures=True)

backend/services/worker_manager.py

@@ -336,6 +336,8 @@ class PostgresTaskWorkerManager:
         service = None
         claim = None
         try:
+            if self._stopped.is_set():
+                return
             claim = self._claim_for_task(task_id)
             if claim is None:
                 self._task_finished(task_id, claimed=False)
@@ -491,11 +493,17 @@ class PostgresTaskWorkerManager:
         for task_id in task_ids:
             self.schedule(task_id)
 
-    def shutdown(self) -> None:
-        """停止调度并等待本进程持有的任务退出后再释放线程池。"""
+    def stop_claiming(self) -> None:
+        """停止任务认领和恢复扫描，保留运行任务的写回能力。"""
         self._stopped.set()
         if self._recovery_thread is not None:
             self._recovery_thread.join()
+
+    def shutdown(self) -> None:
+        """等待自有线程退出；共享线程池由生命周期管理器提前等待。"""
+        self.stop_claiming()
+        if self._owns_executor:
+            self._executor.shutdown(wait=True, cancel_futures=True)
         now = utcnow()
         with self.resources.factory() as session:
             rows = list(session.scalars(select(Task).where(Task.status == "running", Task.lease_owner == self.owner).with_for_update(skip_locked=True)))
@@ -521,7 +529,3 @@ class PostgresTaskWorkerManager:
                 task.updated_at = now
                 self._release_slot(session, task.scope_id, task.id, owner=self.owner, claim_generation=task.claim_generation)
             session.commit()
-        if self._owns_executor:
-            # 任务线程可能仍在使用数据库连接；先等待其退出，避免应用释放连接池
-            # 后留下后台事务与下一次启动/测试清理互相死锁。
-            self._executor.shutdown(wait=True, cancel_futures=True)

backend/tasks.py

@@ -89,6 +89,8 @@ STABLE_TASK_ERRORS = {
     "agent_result_path_invalid",
     "target_changed",
     "task_interrupted",
+    "service_shutdown",
+    "agent_attempt_metadata_write_failed",
     "generation_policy_conflict",
     "processing_options_conflict",
     "invalid_auto_name",

docker-compose.yml

@@ -156,6 +156,8 @@ services:
 
   mememeow:
     profiles: ["app"]
+    # API 需要等待 Agent 进程结束和恢复信息保存，依赖服务随后关闭。
+    stop_grace_period: 120s
     depends_on:
       postgres:
         condition: service_healthy

docs/agent-session-resume.md

@@ -17,6 +17,16 @@ session 才能续跑。429、5xx、连接中断和可证明的进程级暂态错
 用户显式创建新的处理 revision。续跑使用新的 executor attempt ID，不得复用已终态
 ID，也不得使用 OpenCode 全局“继续最近会话”语义。
 
+正常关闭服务时，Worker 停止认领新任务，运行中的 Agent 收到 `service_shutdown`
+中断通知。原执行线程继续维护租约，等待进程结束和 attempt 恢复信息保存，再将
+符合上述条件的同一 Task 排入 `queued`。服务启动后按照现有等待时间继续原 session。
+父 Job、Task ID、输入和授权关联保持不变；适配层继续使用原额度预留和原期限。
+停机续跑计入已有次数和累计时间预算。用户取消和已提交的成功结果保持原状态。
+session 缺失、进程结束无法确认或恢复信息保存失败时，任务按具体原因结束。
+
+Compose 为 API 提供 120 秒关闭时间，并按照依赖关系在 API 退出后关闭 executor
+与 PostgreSQL。突然终止或超过关闭时间后被强制终止，继续使用原有租约过期规则。
+
 任务接口中的 `resume_available`、`resume_reason`、`session_id`、
 `executor_attempt_id`、`resume_attempts`、`resume_started_at`、`first_error` 和 `error_history` 只包含有限脱敏诊断。图片
 处理 Job 的阶段摘要会同步展示当前 attempt 的 session/恢复状态；不会返回 prompt、

executor/server.py

@@ -357,6 +357,7 @@ class TaskState:
     result_path: str | None = None
     process: subprocess.Popen[bytes] | None = field(default=None, repr=False)
     cancel_event: threading.Event = field(default_factory=threading.Event, repr=False)
+    cancel_reason: str | None = None
     done: threading.Event = field(default_factory=threading.Event, repr=False)
     analysis_policy: dict[str, object] | None = field(default=None, repr=False)
     observed_cost: str | None = None
@@ -787,6 +788,7 @@ class Executor:
                 continue
             error = entry.get("error") if isinstance(entry.get("error"), dict) else {}
             if error.get("error") not in {
+                "service_shutdown",
                 "agent_provider_rate_limited",
                 "agent_provider_server_error",
                 "agent_connection_interrupted",
@@ -994,6 +996,7 @@ class Executor:
             raise RuntimeError("session_not_resumable")
         source_error = (source.error or {}).get("error")
         if source_error not in {
+            "service_shutdown",
             "agent_provider_rate_limited",
             "agent_provider_server_error",
             "agent_connection_interrupted",
@@ -1307,6 +1310,9 @@ class Executor:
 
     def _run(self, task: TaskState) -> None:
         """执行 OpenCode 任务并更新状态，失败时保存脱敏后的有限诊断。"""
+        with self.lock:
+            if task.done.is_set():
+                return
         process: subprocess.Popen[bytes] | None = None
         stdout = b""
         stderr = b""
@@ -1556,16 +1562,21 @@ class Executor:
             with self.lock:
                 task.process = None
                 if task.cancel_event.is_set() and (final_error or {}).get("error") != "unknown_execution":
-                    final_status = "cancelled"
-                    final_error = _json_error("task_interrupted", "任务已取消")
+                    if task.cancel_reason != "service_shutdown":
+                        final_status = "cancelled"
+                        final_error = _json_error("task_interrupted", "任务已取消")
+                    elif (final_error or {}).get("error") == "task_interrupted":
+                        final_status = "failed"
+                        final_error = _json_error("service_shutdown", "服务正常关闭，Agent 执行已中断")
                 task.status = final_status
                 task.error = final_error
                 task.completed_at = time.time()
                 try:
                     self._persist_attempt_metadata(task)
-                except (OSError, ValueError, TypeError):
-                    # 元数据写入失败时，后续恢复会因缺少签名事实被拒绝。
-                    pass
+                except (OSError, ValueError, TypeError) as exc:
+                    task.status = "failed"
+                    task.error = _json_error("agent_attempt_metadata_write_failed", "无法保存 executor attempt 恢复信息")
+                    logger.error("attempt 恢复信息保存失败：attempt={} error={}", task.executor_attempt_id, type(exc).__name__)
                 task.done.set()
             if task.status in {"failed", "cancelled"}:
                 secrets = diagnostic_secrets + (task.session_id or "",)
@@ -1692,21 +1703,32 @@ class Executor:
         """终止任务进程组并返回父进程是否已被 waitpid 确认回收。"""
         return ProcessSupervisor().terminate(process).reaped
 
-    def cancel(self, task_id: str) -> TaskState:
+    def cancel(self, task_id: str, *, reason: str = "task_interrupted") -> TaskState:
         """取消指定任务；只终止该任务进程，不影响 executor 或其他任务。"""
+        if reason not in {"task_interrupted", "service_shutdown"}:
+            raise ValueError("invalid_interruption_reason")
         with self.lock:
             task = self.tasks.get(task_id)
             if task is None:
                 raise KeyError(task_id)
             if task.status in {"succeeded", "failed", "cancelled"}:
                 return task
+            # 用户取消优先；后续停机通知不能改变已经收到的取消原因。
+            if task.cancel_reason is None or reason == "task_interrupted":
+                task.cancel_reason = reason
             task.cancel_event.set()
             was_queued = task.status == "queued"
             process = task.process
             if process is None and was_queued:
-                task.status = "cancelled"
-                task.error = _json_error("task_interrupted", "任务已取消")
+                task.status = "failed" if task.cancel_reason == "service_shutdown" else "cancelled"
+                task.error = _json_error(task.cancel_reason, "服务正常关闭，尚未开始执行" if task.cancel_reason == "service_shutdown" else "任务已取消")
                 task.completed_at = time.time()
+                try:
+                    self._persist_attempt_metadata_locked(task)
+                except (OSError, ValueError, TypeError) as exc:
+                    task.status = "failed"
+                    task.error = _json_error("agent_attempt_metadata_write_failed", "无法保存 executor attempt 恢复信息")
+                    logger.error("attempt 恢复信息保存失败：attempt={} error={}", task.executor_attempt_id, type(exc).__name__)
                 task.done.set()
         # 运行任务由执行线程终止并完成数据库维护，再统一发布最终状态。
         return task
@@ -1809,10 +1831,11 @@ class Handler(BaseHTTPRequestHandler):
             self._send(401, _json_error("executor_unauthorized", "executor 认证失败"))
             return
         path = urlsplit(self.path).path.rstrip("/") or "/"
-        cancel_match = re.fullmatch(r"/v1/tasks/([^/]+)/cancel", path)
+        cancel_match = re.fullmatch(r"/v1/tasks/([^/]+)/(cancel|interrupt-for-shutdown)", path)
         if cancel_match:
             try:
-                task = self.server.executor.cancel(unquote(cancel_match.group(1)))
+                reason = "service_shutdown" if cancel_match.group(2) == "interrupt-for-shutdown" else "task_interrupted"
+                task = self.server.executor.cancel(unquote(cancel_match.group(1)), reason=reason)
             except KeyError:
                 self._send(404, _json_error("task_not_found", "任务不存在"))
                 return

tests/test_shutdown_resume_postgres.py

@@ -0,0 +1,112 @@
+# 在现有开发 PostgreSQL 中验证停机续跑状态转换，仅创建独立测试 scope。
+
+from concurrent.futures import ThreadPoolExecutor
+from datetime import UTC, datetime, timedelta
+import os
+from uuid import uuid4
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.engine import make_url
+
+from backend.agent_resume import classify_resume_error
+from backend.persistence.models import Scope, ScopeContext, Task
+from backend.persistence.resources import DatabaseResources
+from backend.persistence.repositories.tasks import TaskRepository
+
+
+@pytest.fixture
+def shutdown_case(tmp_path):
+    """连接明确指定的开发数据库，测试记录使用独立 lane 防止后台认领。"""
+    url = os.environ.get("MEMEMEOW_DATABASE_URL")
+    if not url:
+        pytest.skip("需要显式指定现有开发 PostgreSQL")
+    parsed = make_url(url)
+    assert parsed.host in {"127.0.0.1", "localhost"} and parsed.port in {5434, 25434}
+    engine = create_engine(url, pool_size=4, max_overflow=0)
+    resources = DatabaseResources(engine, image_root=tmp_path / "images", data_root=tmp_path / "data")
+    scope = ScopeContext(f"shutdown-test-{uuid4().hex}")
+    task_id = f"shutdown-test-{uuid4().hex}"
+    with resources.factory.begin() as session:
+        session.add(Scope(id=scope.scope_id))
+        session.flush()
+        session.add(Task(
+            id=task_id, scope_id=scope.scope_id, task_type="meme_context_generation",
+            status="running", lane="shutdown-verification", payload={},
+            lease_owner="shutdown-test", lease_expires_at=datetime.now(UTC) + timedelta(hours=1),
+            claim_generation=1, attempt_count=1, max_attempts=1,
+        ))
+    try:
+        yield resources, scope, task_id
+    finally:
+        with resources.factory.begin() as session:
+            TaskRepository(session, scope).cancel(task_id, error={"error": "test_finished"}, message="停机测试结束")
+        engine.dispose()
+
+
+def requeue(case, *, available=True, budget=4):
+    """通过真实 repository 提交停机错误和已经验证的恢复标识。"""
+    resources, scope, task_id = case
+    with resources.factory.begin() as session:
+        return TaskRepository(session, scope).fail_fenced(
+            task_id, 1, "shutdown-test", error={"error": "service_shutdown"},
+            message="等待服务启动后继续", retry=available,
+            resume_available=available, resume_reason="service_shutdown",
+            session_id="session-shutdown-test" if available else None,
+            executor_attempt_id="attempt-shutdown-test", resume_max_attempts=budget,
+            retry_delay_seconds=3600,
+        )
+
+
+def test_shutdown_requeues_single_attempt_task(shutdown_case):
+    """停机使用独立续跑预算，保留 Task 身份、max_attempts 和错误原因。"""
+    assert requeue(shutdown_case) == (True, True)
+    resources, _, task_id = shutdown_case
+    with resources.factory() as session:
+        task = session.get(Task, task_id)
+        assert task.status == "queued" and task.max_attempts == 1
+        assert task.resume_session_id == "session-shutdown-test"
+        assert task.error["error"] == "service_shutdown"
+        assert task.lease_owner is None and task.resume_started_at is not None
+
+
+@pytest.mark.parametrize("available,budget", [(False, 4), (True, 0)])
+def test_shutdown_without_session_or_budget_finishes(shutdown_case, available, budget):
+    """session 缺失或预算耗尽时终止 Task，禁止重新开始外部执行。"""
+    assert requeue(shutdown_case, available=available, budget=budget) == (True, False)
+    resources, _, task_id = shutdown_case
+    with resources.factory() as session:
+        assert session.get(Task, task_id).status == "failed"
+
+
+@pytest.mark.parametrize("success", [False, True])
+def test_shutdown_cannot_replace_committed_terminal_state(shutdown_case, success):
+    """停机线程的迟到写回不能改变已经提交的成功或用户取消结果。"""
+    resources, scope, task_id = shutdown_case
+    with resources.factory.begin() as session:
+        repository = TaskRepository(session, scope)
+        if success:
+            assert repository.update_fenced(task_id, 1, "shutdown-test", status="succeeded", result={})
+        else:
+            assert repository.cancel(task_id, error={"error": "task_cancelled"}, message="用户取消")
+    assert requeue(shutdown_case) == (False, False)
+    with resources.factory() as session:
+        task = session.get(Task, task_id)
+        assert task.status == ("succeeded" if success else "failed")
+        if not success:
+            assert task.error["error"] == "task_cancelled"
+
+
+def test_concurrent_shutdown_write_has_one_winner(shutdown_case):
+    """并发执行相同 claim 的停机写回，只有一个事务能够重新排队。"""
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        results = list(pool.map(requeue, [shutdown_case, shutdown_case]))
+    assert sorted(results) == [(False, False), (True, True)]
+
+
+def test_shutdown_error_requires_session_and_preserves_cancellation():
+    """正常停机可继续已有 session，用户取消与保存失败均禁止续跑。"""
+    assert classify_resume_error("service_shutdown", session_id="session-test").available
+    assert classify_resume_error("service_shutdown", session_id=None).reason == "session_missing"
+    for code in ("task_interrupted", "agent_attempt_metadata_write_failed"):
+        assert not classify_resume_error(code, session_id="session-test").available