fix(taskrun): reject busy documents before run creation

2026-09-13 04:43:50 +00:00
parent f17781adb6
commit 6ab8b52113
11 changed files with 245 additions and 5 deletions
@@ -37,7 +37,7 @@ dev PostgreSQL 的 `sess-p143-u9` 在 03:17–03:32 记录了 `debug ping`、`te

 ### 5. 文档忙被错误分型为系统内部错误

-并发请求因有效租约拒绝时，事件被分类为 `INTERNAL_ERROR/fatal/actions=[restart]`，用户看到“系统内部错误，请重新开始会话”。实际是可预期的 `DOC_BUSY` 状态，应禁止重复启动并展示持有人、进度与预计状态，而不是建议重启会话。
+并发请求因有效租约拒绝时，事件被分类为 `INTERNAL_ERROR/fatal/actions=[restart]`，用户看到“系统内部错误，请重新开始会话”。实际是可预期的 `DOC_BUSY` 状态；除正确分型外，还必须在创建 `task_runs` 之前做服务端 busy 预检，持久化“未创建新任务”的只读状态卡，且不提供重试动作。

 ### 6. dev restart 不会加载新代码，健康检查可对旧镜像误报通过

@@ -36,6 +36,7 @@ mutex: intent
 5. [x] 实现命名文档的服务端鉴权只读总结（“总结”不止是不写入）
 6. [ ] 执行前端、Go、PG 与真实 PI 双标签页验收
   - [x] Go、前端、构建、dev 重建与真实 PI 只读总结
+   - [x] 服务端 busy 预检不创建 run，并持久化无重试状态卡
   - [ ] 浏览器刷新恢复与双标签页 busy 可见性

 ## 验收标准
@@ -26,7 +26,7 @@ updated: 2026-09-13
 | Wave 1 | 将客户端模板匹配收紧为“动作 + 目标”；“总结”由服务端按项目/阶段权限读取正文并在隔离对话目录生成只读摘要 | 明确“编写/生成/更新”触发；总结返回正文摘要且不创建 run/租约 |
 | Wave 2 | 持久化/重建 `error_info`、task request、进度与恢复动作 | 刷新前后失败卡片语义和按钮一致 |
 | Wave 3 | 增加运行态/租约查询与统一 busy UI，消费 governance 事件 | 运行中刷新仍显示文档、持有人、阶段、开始时间和禁止编辑提示 |
-| Wave 4 | 将 busy 分型为策略状态，执行真实 PI + 刷新 + 双标签页 E2E | 重复请求不会创建第二 run；状态可恢复；完成后自动解锁 |
+| Wave 4 | 将 busy 分型为策略状态，并在 `task_runs`/聊天占位创建前做服务端预检；执行真实 PI + 刷新 + 双标签页 E2E | 重复请求不会创建第二 run；busy 卡刷新可恢复；完成后自动解锁 |

 ## 验收标准

@@ -87,6 +87,12 @@
 - **Actual Output**: restart 后仍返回旧版 `CASUAL_CHAT`；重建后返回新版 `QUERY_DOC_STATUS/doc_summary`
 - **Evaluation**: 首次 FAIL、整改后 PASS；`./scripts/env-service.sh dev rebuild` 已实跑通过，集成冒烟统一改用该命令

+### 用例 J：文档 busy 服务端预检
+
+- **Input**: 文档持有有效人工租约时请求 `/api/v1/tasks/runs`，携带明确写入语句
+- **Actual Output**: HTTP 409 / `DOC_BUSY`；响应不含 `run_id`；服务端持久化 user + assistant 两条消息，assistant 含 `error_info.actions=[]` 与原 `task_request`
+- **Evaluation**: PASS；刷新后仍可恢复“当前请求未创建新任务”的状态卡，且不会显示重试按钮
+
 ## 六、 自动化测试套件汇总 (Test Suite Execution)

 ```text
@@ -0,0 +1,71 @@
+package server
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/docgov"
+)
+
+func TestCreateTaskRunBusyPreflightDoesNotPersistRunOrMessages(t *testing.T) {
+	srv := newTestServer(t)
+	ctx := t.Context()
+	docID := "doc_proj143_intake"
+	if _, err := srv.svc.DocGov().CreateOrGetDocument(
+		ctx, docID, 143, "intake", "用户初始需求收集表", "V0.1_triage", "",
+		1, "编制人", 1, "审核人", 1, "批准人",
+	); err != nil {
+		t.Fatalf("create document: %v", err)
+	}
+	if _, err := srv.svc.DocGov().AcquireLease(ctx, docgov.LeaseAcquireRequest{
+		DocID: docID, HolderType: docgov.LeaseHolderHuman, HolderID: "editor-9", TTL: time.Minute,
+	}); err != nil {
+		t.Fatalf("acquire fixture lease: %v", err)
+	}
+	defer func() {
+		_, _ = srv.svc.DocGov().ReleaseLease(t.Context(), docID, string(docgov.LeaseHolderHuman), "editor-9")
+	}()
+
+	requestBody, _ := json.Marshal(createTaskRunRequest{
+		TaskRunRequestDTO: TaskRunRequestDTO{TaskKey: "intake", ProjectID: 143, ProjectName: "隔离项目"},
+		SessionID:         "fixture-busy-preflight", ClientTurnID: "turn-busy-1", UserMessage: "开始编写用户初始需求表",
+	})
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/tasks/runs", bytes.NewReader(requestBody))
+	withBearerAuth(req, loginTestAdmin(t, srv))
+	w := httptest.NewRecorder()
+	srv.Routes().ServeHTTP(w, req)
+	if w.Code != http.StatusConflict {
+		t.Fatalf("status=%d body=%s", w.Code, w.Body.String())
+	}
+	var payload map[string]any
+	if err := json.NewDecoder(w.Body).Decode(&payload); err != nil {
+		t.Fatalf("decode response: %v", err)
+	}
+	if payload["error_code"] != "DOC_BUSY" {
+		t.Fatalf("payload=%+v", payload)
+	}
+	if _, exists := payload["run_id"]; exists {
+		t.Fatalf("busy preflight must not return run_id: %+v", payload)
+	}
+	history, err := srv.svc.ContextManager().GetChatHistory(ctx, "fixture-busy-preflight", 20, 0)
+	if err != nil {
+		t.Fatalf("get busy history: %v", err)
+	}
+	if len(history) != 2 {
+		t.Fatalf("busy preflight history=%d want user+assistant", len(history))
+	}
+	if history[1].Metadata["task_status"] != "failed" {
+		t.Fatalf("assistant metadata=%+v", history[1].Metadata)
+	}
+	errorInfo, ok := history[1].Metadata["error_info"].(map[string]any)
+	if !ok || errorInfo["code"] != "DOC_BUSY" {
+		t.Fatalf("error_info=%+v", history[1].Metadata["error_info"])
+	}
+	if actions, ok := errorInfo["actions"].([]string); !ok || len(actions) != 0 {
+		t.Fatalf("busy actions=%+v", errorInfo["actions"])
+	}
+}
@@ -13,6 +13,8 @@ import (

 	"chat-agent-system/internal/agent"
 	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/ctxmgr"
+	"chat-agent-system/internal/docgov"
 	"chat-agent-system/internal/projectinstance"
 	"chat-agent-system/internal/squad"
 	"chat-agent-system/internal/stream"
@@ -87,6 +89,19 @@ func (s *Server) handleCreateTaskRun(w http.ResponseWriter, r *http.Request) {
 		})
 		return
 	}
+	if busy, preflightErr := s.preflightTaskDocumentBusy(r.Context(), req); preflightErr != nil {
+		http.Error(w, preflightErr.Error(), http.StatusBadRequest)
+		return
+	} else if busy != nil {
+		s.recordBusyPreflightMessages(r.Context(), ctxMgr, req, taskScope, busy)
+		writeJSONResponse(w, http.StatusConflict, map[string]any{
+			"error":      "document_busy",
+			"error_code": "DOC_BUSY",
+			"message":    busy.Reason,
+			"detail":     busy,
+		})
+		return
+	}

 	runID := fmt.Sprintf("run_%d", time.Now().UnixNano())
 	var assistantMsgID int64
@@ -154,6 +169,102 @@ func (s *Server) handleCreateTaskRun(w http.ResponseWriter, r *http.Request) {
 	})
 }

+func (s *Server) recordBusyPreflightMessages(
+	ctx context.Context,
+	ctxMgr *ctxmgr.ContextManager,
+	req createTaskRunRequest,
+	taskScope *projectinstance.MessageScope,
+	busy *docgov.DocumentBusyError,
+) {
+	if ctxMgr == nil || busy == nil || req.UserMessage == "" {
+		return
+	}
+	userMeta := map[string]any{
+		"project_id":     req.ProjectID,
+		"client_turn_id": req.ClientTurnID,
+		"task_key":       req.TaskKey,
+	}
+	projectinstance.ApplyScopeToMetadata(userMeta, taskScope)
+	_, _ = ctxMgr.RecordUserMessageWithID(ctx, req.SessionID, req.UserID, "user", req.UserMessage, "TRIGGER_TASK", "", userMeta)
+
+	taskRequest := map[string]any{
+		"task_key":       req.TaskKey,
+		"project_name":   req.ProjectName,
+		"customer_name":  req.CustomerName,
+		"device_type":    req.DeviceType,
+		"intended_use":   req.IntendedUse,
+		"target_market":  req.TargetMarket,
+		"session_id":     req.SessionID,
+		"user_id":        req.UserID,
+		"project_id":     req.ProjectID,
+		"client_turn_id": req.ClientTurnID,
+		"user_message":   req.UserMessage,
+		"doc_id":         req.DocID,
+	}
+	errorInfo := map[string]any{
+		"code":         "DOC_BUSY",
+		"class":        "fatal",
+		"user_message": busy.Reason,
+		"actions":      []string{},
+		"session_id":   req.SessionID,
+	}
+	asstMeta := map[string]any{
+		"project_id":     req.ProjectID,
+		"client_turn_id": req.ClientTurnID,
+		"task_key":       req.TaskKey,
+		"task_status":    "failed",
+		"task_request":   taskRequest,
+		"error_info":     errorInfo,
+	}
+	projectinstance.ApplyScopeToMetadata(asstMeta, taskScope)
+	_, _ = ctxMgr.RecordUserMessageWithID(ctx, req.SessionID, 0, "assistant", busy.Reason, "TRIGGER_TASK", "", asstMeta)
+}
+
+// preflightTaskDocumentBusy 在写入聊天占位与 task_runs 之前拦截已知租约占用。
+// 真正的租约获取仍由 docgov 原子执行；本检查用于避免用户重复点击产生必然失败的 run。
+func (s *Server) preflightTaskDocumentBusy(ctx context.Context, req createTaskRunRequest) (*docgov.DocumentBusyError, error) {
+	if req.Stage != "" || req.ProjectID <= 0 || req.TaskKey == "" || s.svc == nil || s.svc.DocGov() == nil {
+		return nil, nil
+	}
+	taskReq := s.buildBizdevTaskRequest(req.TaskRunRequestDTO, req.SessionID, req.ClientTurnID, "")
+	docID, err := s.svc.ResolveAndValidateDocID(ctx, taskReq)
+	if err != nil {
+		return nil, err
+	}
+	if docID == "" {
+		return nil, nil
+	}
+	doc, err := s.svc.DocGov().GetDocument(ctx, docID)
+	if errors.Is(err, docgov.ErrDocumentNotFound) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	if !doc.HasActiveLease(time.Now()) {
+		return nil, nil
+	}
+	// 已结束的小队 run 由 AcquireLease 的孤儿回收逻辑处理，不在这里误拦截。
+	if doc.LeaseHolderType == docgov.LeaseHolderSquad && s.taskRuns != nil {
+		if run, runErr := s.taskRuns.GetRun(doc.LeaseHolderID); runErr == nil && run.Status != "running" {
+			return nil, nil
+		}
+	}
+	expiresAt := time.Time{}
+	if doc.LeaseExpiresAt != nil {
+		expiresAt = *doc.LeaseExpiresAt
+	}
+	reason := fmt.Sprintf("文档【%s】正在由 %s(%s) 编辑中，当前请求未创建新任务", doc.Title, doc.LeaseHolderType, doc.LeaseHolderID)
+	return &docgov.DocumentBusyError{
+		DocID:      doc.DocID,
+		HolderType: doc.LeaseHolderType,
+		HolderID:   doc.LeaseHolderID,
+		WorkDir:    doc.LeaseWorkDir,
+		ExpiresAt:  expiresAt,
+		Reason:     reason,
+	}, nil
+}
+
 func (s *Server) handleTaskRunEvents(w http.ResponseWriter, r *http.Request) {
 	user, ok := s.requirePrincipal(w, r)
 	if !ok {
@@ -2139,6 +2139,14 @@ export default function App() {
  ): ChatErrorInfo => {
    const code = opts?.errorCode || '';
    const detail = opts?.detail || '';
+    if (code === 'DOC_BUSY' || code === 'document_busy') {
+      return {
+        userMessage: detail || '该文档正在被其他用户或项目助手编辑，当前请求未创建新任务',
+        code: 'DOC_BUSY',
+        class: 'fatal',
+        actions: [],
+      };
+    }
    if (code === 'DOC_TASK_NOT_IN_ACTIVE_STAGE') {
      return {
        userMessage: detail || '该文档任务不在当前活动阶段，请先完成当前阶段工作流后再试',
@@ -25,6 +25,10 @@ test('shouldAutoRetryTask: transient only, max 2', () => {
  );
 });

+test('shouldAutoRetryTask: document busy never retries', () => {
+  assert.equal(shouldAutoRetryTask({ class: 'transient', code: 'DOC_BUSY' }, 0), false);
+});
+
 test('shouldAutoRetry alias delegates to task path', () => {
  assert.equal(shouldAutoRetry({ class: 'transient', code: 'NETWORK_ERROR' }, 0), true);
 });
@@ -16,6 +16,7 @@ export function shouldAutoRetryTask(
  if (
    info.code === 'DOC_TASK_NOT_IN_ACTIVE_STAGE' ||
    info.code === 'DOC_DUTY_FORBIDDEN' ||
+    info.code === 'DOC_BUSY' ||
    info.code === 'ERR_FORBIDDEN'
  ) {
    return false;
@@ -0,0 +1,31 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { createTaskRun } from './taskRuns';
+
+test('createTaskRun exposes DOC_BUSY detail without a retryable run', async () => {
+  const originalFetch = globalThis.fetch;
+  globalThis.fetch = async () =>
+    new Response(
+      JSON.stringify({
+        error: 'document_busy',
+        error_code: 'DOC_BUSY',
+        message: '文档正在由其他用户编辑中，当前请求未创建新任务',
+        detail: { reason: '文档正在由其他用户编辑中' },
+      }),
+      { status: 409, headers: { 'Content-Type': 'application/json' } },
+    );
+  try {
+    await assert.rejects(
+      () => createTaskRun({ session_id: 'fixture', project_id: 143 }),
+      (err: unknown) => {
+        const typed = err as { httpStatus?: number; errorCode?: string; detail?: string };
+        assert.equal(typed.httpStatus, 409);
+        assert.equal(typed.errorCode, 'DOC_BUSY');
+        assert.match(typed.detail || '', /未创建新任务/);
+        return true;
+      },
+    );
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
@@ -25,9 +25,16 @@ export async function createTaskRun(body: CreateTaskRunBody): Promise<CreateTask
    let detail = '';
    let errorCode = '';
    try {
-      const payload = (await res.json()) as { message?: string; error?: string; error_code?: string };
-      errorCode = payload.error || payload.error_code || '';
-      detail = payload.message || errorCode || '';
+      const payload = (await res.json()) as {
+        message?: string;
+        error?: string;
+        error_code?: string;
+        detail?: string | { reason?: string };
+      };
+      const rawCode = payload.error_code || payload.error || '';
+      errorCode = rawCode === 'document_busy' ? 'DOC_BUSY' : rawCode;
+      const structuredDetail = typeof payload.detail === 'object' ? payload.detail?.reason : payload.detail;
+      detail = payload.message || structuredDetail || errorCode || '';
    } catch {
      // ignore non-json body
    }