feat: async ai workbench tasks

internal/app/app.go

@@ -21,6 +21,7 @@ type App struct {
 	redis  *redis.Client
 	events events.Publisher
 	api    *httpapi.API
+	cancel context.CancelFunc
 }
 
 func New(cfg *config.Config) (*App, error) {
@@ -60,11 +61,15 @@ func New(cfg *config.Config) (*App, error) {
 		return nil, err
 	}
 	aiService := ai.NewService(ai.NewClient(cfg.AI))
+	api := httpapi.New(st, auth.New(cfg.Auth, redisClient), publisher, cfg.Auth, cfg.Manager, aiService)
+	ctx, cancel := context.WithCancel(context.Background())
+	api.StartAIWorkers(ctx)
 	return &App{
 		store:  st,
 		redis:  redisClient,
 		events: publisher,
-		api:    httpapi.New(st, auth.New(cfg.Auth, redisClient), publisher, cfg.Auth, cfg.Manager, aiService),
+		api:    api,
+		cancel: cancel,
 	}, nil
 }
 
@@ -77,6 +82,9 @@ func (a *App) Handler() http.Handler {
 }
 
 func (a *App) Close() {
+	if a.cancel != nil {
+		a.cancel()
+	}
 	if a.redis != nil {
 		_ = a.redis.Close()
 	}

internal/httpapi/ai.go

@@ -118,48 +118,28 @@ func (a *API) aiDocumentGenerate(w http.ResponseWriter, r *http.Request, p *auth
 		return
 	}
 	taskID := "aitask_" + randomToken(12)
-	started := time.Now().UTC().Round(time.Millisecond)
+	snapshot, err := aiTaskSnapshot(aiDocumentTaskSnapshot{
+		SourceText:   source,
+		TitleHint:    strings.TrimSpace(body.TitleHint),
+		DocumentType: strings.TrimSpace(body.DocumentType),
+		Language:     strings.TrimSpace(body.Language),
+	})
+	if err != nil {
+		writeStoreError(w, r, err)
+		return
+	}
 	if err := st.CreateAITask(r.Context(), store.AITask{
 		TaskID: taskID, TenantCode: p.TenantCode, UserID: strconv.FormatInt(p.UserID, 10), Feature: "smart_document",
-		Status: "processing", Progress: 10, SourceType: "text", InputHash: hashText(source),
+		Status: "queued", Progress: 0, SourceType: "text", InputHash: hashText(source),
+		InputSnapshot: snapshot,
 		PromptVersion: "smart_document_v1", Model: "",
-		StartedAt: &started,
 	}, nil); err != nil {
 		writeStoreError(w, r, err)
 		return
 	}
-	result, err := a.ai.GenerateDocument(r.Context(), source, strings.TrimSpace(body.TitleHint), strings.TrimSpace(body.DocumentType), strings.TrimSpace(body.Language))
-	if err != nil {
-		_ = st.UpdateAITaskResult(r.Context(), taskID, "failed", 100, "", nil, aiErrorReason(err), err.Error())
-		_ = st.InsertAIUsageLog(r.Context(), store.AIUsageLog{
-			TenantCode: p.TenantCode, UserID: formatUserID(p.UserID), Feature: "smart_document", TaskID: taskID,
-			Provider: "deepseek", Status: "failed", ErrorReason: aiErrorReason(err), InputHash: hashText(source),
-		})
-		a.writeAIError(w, r, err)
-		return
-	}
-	taskResult := map[string]any{
-		"title":            result.Title,
-		"summary":          result.Summary,
-		"content_markdown": result.ContentMarkdown,
-		"content_plain":    result.ContentPlain,
-		"sections":         result.Sections,
-		"source_excerpt":   excerpt(source, 200),
-		"can_save":         true,
-	}
-	if err := st.UpdateAITaskResult(r.Context(), taskID, "succeeded", 100, "", taskResult, "", ""); err != nil {
-		writeStoreError(w, r, err)
-		return
-	}
-	_ = st.InsertAIUsageLog(r.Context(), store.AIUsageLog{
-		TenantCode: p.TenantCode, UserID: formatUserID(p.UserID), Feature: "smart_document", TaskID: taskID,
-		Provider: "deepseek", Model: result.Usage.Model, ProviderRequestID: result.Usage.ProviderRequestID,
-		PromptTokens: result.Usage.PromptTokens, CompletionTokens: result.Usage.CompletionTokens, TotalTokens: result.Usage.TotalTokens,
-		LatencyMS: int(result.Usage.LatencyMS), Status: "succeeded", InputHash: hashText(source), OutputHash: hashText(result.ContentMarkdown),
-	})
 	writeSuccess(w, r, map[string]any{
 		"task_id":             taskID,
-		"status":              "succeeded",
+		"status":              "queued",
 		"polling_interval_ms": 1500,
 		"expires_at":          time.Now().UTC().Add(2 * time.Hour).Format(time.RFC3339Nano),
 	})
@@ -179,6 +159,13 @@ func (a *API) aiTaskResource(w http.ResponseWriter, r *http.Request, p *auth.Pri
 			writeStoreError(w, r, err)
 			return
 		}
+		feature := ""
+		if task, err := st.AITask(r.Context(), p.UserID, taskID); err == nil {
+			feature = task.Feature
+		}
+		a.publishAITaskEvent(r.Context(), p.TenantID, p.TenantCode, p.UserID, taskID, "ai", "AI 任务已取消", "", map[string]any{
+			"task_id": taskID, "feature": feature, "status": "canceled", "progress": 100, "result_type": "none",
+		})
 		writeSuccess(w, r, map[string]any{"task_id": taskID, "status": "canceled", "canceled_at": time.Now().UTC().Format(time.RFC3339Nano)})
 		return
 	}
@@ -370,50 +357,23 @@ func (a *API) aiTableAnalysisCreate(w http.ResponseWriter, r *http.Request, p *a
 	}
 	taskID := "aitask_tbl_" + randomToken(12)
 	recordID := "airec_tbl_" + randomToken(10)
-	started := time.Now().UTC().Round(time.Millisecond)
+	snapshot, err := aiTaskSnapshot(aiTableTaskSnapshot{
+		FileID: fileID, FileName: strings.TrimSpace(body.FileName), FileType: fileType, FileSize: fileSize,
+		Topic: topic, SheetName: strings.TrimSpace(body.SheetName), Language: strings.TrimSpace(body.Language), RecordID: recordID,
+	})
+	if err != nil {
+		writeStoreError(w, r, err)
+		return
+	}
 	if err := st.CreateAITask(r.Context(), store.AITask{
 		TaskID: taskID, TenantCode: p.TenantCode, UserID: strconv.FormatInt(p.UserID, 10), Feature: "smart_table",
-		Status: "processing", Progress: 20, SourceType: "file", SourceRefID: fileID, InputHash: hashText(fileID + ":" + topic),
+		Status: "queued", Progress: 0, SourceType: "file", SourceRefID: fileID, InputHash: hashText(fileID + ":" + topic),
-		PromptVersion: "smart_table_v1", StartedAt: &started,
+		InputSnapshot: snapshot, PromptVersion: "smart_table_v1",
 	}, nil); err != nil {
 		writeStoreError(w, r, err)
 		return
 	}
-	profile := a.tableProfileForAI(r, fileID, fileMeta, fileType)
+	writeSuccess(w, r, map[string]any{"task_id": taskID, "record_id": recordID, "status": "queued", "polling_interval_ms": 1500})
-	result, err := a.ai.AnalyzeTable(r.Context(), ai.TableAnalysisInput{
-		Topic: topic, FileID: fileID, FileName: strings.TrimSpace(body.FileName), FileType: fileType, FileSize: fileSize, Profile: profile,
-	})
-	if err != nil {
-		_ = st.UpdateAITaskResult(r.Context(), taskID, "failed", 100, "", nil, aiErrorReason(err), err.Error())
-		a.writeAIError(w, r, err)
-		return
-	}
-	metricsJSON := mustJSON(result.Metrics, "[]")
-	previewJSON := mustJSON(result.Preview, "{}")
-	alertsJSON := mustJSON(result.Alerts, "[]")
-	if err := st.SaveAITableAnalysis(r.Context(), p.TenantCode, p.UserID, store.AIRecord{
-		RecordID: recordID, RecordType: "smart_table", Title: result.Title, Summary: excerpt(firstNonEmpty(result.Summary, result.Interpretation), 160),
-		Status: "succeeded", TaskID: taskID,
-	}, store.AITableAnalysis{
-		RecordID: recordID, FileID: fileID, FileName: strings.TrimSpace(body.FileName), FileType: fileType, FileSize: fileSize,
-		Topic: topic, MetricsJSON: metricsJSON, PreviewJSON: previewJSON, AlertsJSON: alertsJSON, Interpretation: result.Interpretation,
-		PromptVersion: "smart_table_v1", Model: result.Usage.Model,
-	}); err != nil {
-		writeStoreError(w, r, err)
-		return
-	}
-	taskResult := map[string]any{"record_id": recordID, "title": result.Title, "metrics": result.Metrics, "alerts": result.Alerts, "interpretation": result.Interpretation}
-	if err := st.UpdateAITaskResult(r.Context(), taskID, "succeeded", 100, recordID, taskResult, "", ""); err != nil {
-		writeStoreError(w, r, err)
-		return
-	}
-	_ = st.InsertAIUsageLog(r.Context(), store.AIUsageLog{
-		TenantCode: p.TenantCode, UserID: formatUserID(p.UserID), Feature: "smart_table", TaskID: taskID,
-		Provider: "deepseek", Model: result.Usage.Model, ProviderRequestID: result.Usage.ProviderRequestID,
-		PromptTokens: result.Usage.PromptTokens, CompletionTokens: result.Usage.CompletionTokens, TotalTokens: result.Usage.TotalTokens,
-		LatencyMS: int(result.Usage.LatencyMS), Status: "succeeded", InputHash: hashText(fileID + ":" + topic), OutputHash: hashText(result.Interpretation),
-	})
-	writeSuccess(w, r, map[string]any{"task_id": taskID, "record_id": recordID, "status": "succeeded", "polling_interval_ms": 1500})
 }
 
 func (a *API) aiTableAnalysisResource(w http.ResponseWriter, r *http.Request, p *auth.Principal, st *store.Store) {
@@ -532,8 +492,7 @@ func (a *API) aiMeetingSummaryCreate(w http.ResponseWriter, r *http.Request, p *
 	if _, ok := a.requireAICapability(w, r, st, "meeting_summary"); !ok {
 		return
 	}
-	meeting, err := st.Meeting(r.Context(), p.UserID, meetingID)
+	if _, err := st.Meeting(r.Context(), p.UserID, meetingID); err != nil {
-	if err != nil {
 		writeStoreError(w, r, err)
 		return
 	}
@@ -542,40 +501,26 @@ func (a *API) aiMeetingSummaryCreate(w http.ResponseWriter, r *http.Request, p *
 		writeStoreError(w, r, err)
 		return
 	}
-	fileNames, fileIDs := meetingSummaryFiles(files, body.FileIDs)
+	_, fileIDs := meetingSummaryFiles(files, body.FileIDs)
 	taskID := "aitask_mtg_" + randomToken(12)
 	summaryID := "aisum_" + randomToken(10)
-	started := time.Now().UTC().Round(time.Millisecond)
+	snapshot, err := aiTaskSnapshot(aiMeetingSummaryTaskSnapshot{
+		MeetingID: meetingID, SummaryID: summaryID, FileIDs: fileIDs,
+		SummaryType: strings.TrimSpace(body.SummaryType), ForceRegenerate: body.ForceRegenerate,
+	})
+	if err != nil {
+		writeStoreError(w, r, err)
+		return
+	}
 	if err := st.CreateAITask(r.Context(), store.AITask{
 		TaskID: taskID, TenantCode: p.TenantCode, UserID: strconv.FormatInt(p.UserID, 10), Feature: "meeting_summary",
-		Status: "processing", Progress: 20, SourceType: "meeting", SourceRefID: meetingID, InputHash: hashText(meetingID + strings.Join(fileIDs, ",")),
+		Status: "queued", Progress: 0, SourceType: "meeting", SourceRefID: meetingID, InputHash: hashText(meetingID + strings.Join(fileIDs, ",")),
-		PromptVersion: "meeting_summary_v1", StartedAt: &started,
+		InputSnapshot: snapshot, PromptVersion: "meeting_summary_v1",
 	}, nil); err != nil {
 		writeStoreError(w, r, err)
 		return
 	}
-	result, err := a.ai.SummarizeMeeting(r.Context(), ai.MeetingSummaryInput{
+	writeSuccess(w, r, map[string]any{"task_id": taskID, "meeting_id": meetingID, "summary_id": summaryID, "status": "queued", "polling_interval_ms": 1500})
-		MeetingTitle: meeting.Title, StartAt: meeting.StartAt.UTC().Format(time.RFC3339Nano), EndAt: meeting.EndAt.UTC().Format(time.RFC3339Nano),
-		Participants: meetingSummaryParticipants(*meeting), Files: fileNames,
-	})
-	if err != nil {
-		_ = st.UpdateAITaskResult(r.Context(), taskID, "failed", 100, "", nil, aiErrorReason(err), err.Error())
-		a.writeAIError(w, r, err)
-		return
-	}
-	if err := st.UpsertAIMeetingSummary(r.Context(), p.TenantCode, p.UserID, store.AIMeetingSummary{
-		SummaryID: summaryID, MeetingID: meetingID, TaskID: taskID,
-		ConclusionsJSON: mustJSON(result.Conclusions, "[]"), DecisionsJSON: mustJSON(result.Decisions, "[]"),
-		ActionItemsJSON: mustJSON(result.ActionItems, "[]"), SourceFileIDsJSON: mustJSON(fileIDs, "[]"),
-		PromptVersion: "meeting_summary_v1", Model: result.Usage.Model,
-	}); err != nil {
-		writeStoreError(w, r, err)
-		return
-	}
-	_ = st.UpdateAITaskResult(r.Context(), taskID, "succeeded", 100, summaryID, map[string]any{
-		"summary_id": summaryID, "meeting_id": meetingID, "conclusions": result.Conclusions, "decisions": result.Decisions, "action_items": result.ActionItems,
-	}, "", "")
-	writeSuccess(w, r, map[string]any{"task_id": taskID, "meeting_id": meetingID, "status": "succeeded", "polling_interval_ms": 1500})
 }
 
 func (a *API) aiMeetingSummaryUpdate(w http.ResponseWriter, r *http.Request, p *auth.Principal, st *store.Store, meetingID string) {

internal/httpapi/ai_worker.go

@@ -0,0 +1,366 @@
+package httpapi
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"log"
+	"strconv"
+	"strings"
+	"time"
+
+	"git.118im.com/im_team/im_workbench/internal/ai"
+	"git.118im.com/im_team/im_workbench/internal/events"
+	"git.118im.com/im_team/im_workbench/internal/store"
+)
+
+const (
+	aiWorkerCount        = 2
+	aiWorkerPollInterval = time.Second
+	aiTaskRunTimeout     = 3 * time.Minute
+)
+
+type aiDocumentTaskSnapshot struct {
+	SourceText   string `json:"source_text"`
+	TitleHint    string `json:"title_hint"`
+	DocumentType string `json:"document_type"`
+	Language     string `json:"language"`
+}
+
+type aiTableTaskSnapshot struct {
+	FileID    string `json:"file_id"`
+	FileName  string `json:"file_name"`
+	FileType  string `json:"file_type"`
+	FileSize  int64  `json:"file_size"`
+	Topic     string `json:"topic"`
+	SheetName string `json:"sheet_name"`
+	Language  string `json:"language"`
+	RecordID  string `json:"record_id"`
+}
+
+type aiMeetingSummaryTaskSnapshot struct {
+	MeetingID       string   `json:"meeting_id"`
+	SummaryID       string   `json:"summary_id"`
+	FileIDs         []string `json:"file_ids"`
+	SummaryType     string   `json:"summary_type"`
+	ForceRegenerate bool     `json:"force_regenerate"`
+}
+
+func (a *API) StartAIWorkers(ctx context.Context) {
+	if a == nil || a.store == nil || a.ai == nil {
+		return
+	}
+	for i := 0; i < aiWorkerCount; i++ {
+		go a.aiWorkerLoop(ctx, i+1)
+	}
+}
+
+func (a *API) aiWorkerLoop(ctx context.Context, workerID int) {
+	ticker := time.NewTicker(aiWorkerPollInterval)
+	defer ticker.Stop()
+	for {
+		task, err := a.store.ClaimQueuedAITask(ctx)
+		if errors.Is(err, store.ErrNotFound) {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				continue
+			}
+		}
+		if err != nil {
+			log.Printf("ai worker claim failed: worker=%d err=%v", workerID, err)
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				continue
+			}
+		}
+		a.runAITask(ctx, task)
+	}
+}
+
+func (a *API) runAITask(parent context.Context, task *store.AITask) {
+	ctx, cancel := context.WithTimeout(parent, aiTaskRunTimeout)
+	defer cancel()
+	st, err := a.store.ForTenant(ctx, task.TenantID, task.TenantCode)
+	if err != nil {
+		log.Printf("ai task tenant store failed: task_id=%s tenant_id=%d err=%v", task.TaskID, task.TenantID, err)
+		return
+	}
+	switch task.Feature {
+	case "smart_document":
+		a.runAIDocumentTask(ctx, st, task)
+	case "smart_table":
+		a.runAITableTask(ctx, st, task)
+	case "meeting_summary":
+		a.runAIMeetingSummaryTask(ctx, st, task)
+	default:
+		a.finishAITaskFailure(ctx, st, task, "AI_TASK_UNSUPPORTED", "unsupported ai task feature")
+	}
+}
+
+func (a *API) runAIDocumentTask(ctx context.Context, st *store.Store, task *store.AITask) {
+	var input aiDocumentTaskSnapshot
+	if !decodeTaskSnapshot(task, &input) {
+		a.finishAITaskFailure(ctx, st, task, reasonMalformedRequest, "invalid task input")
+		return
+	}
+	result, err := a.ai.GenerateDocument(ctx, strings.TrimSpace(input.SourceText), strings.TrimSpace(input.TitleHint), strings.TrimSpace(input.DocumentType), strings.TrimSpace(input.Language))
+	if err != nil {
+		a.finishAITaskAIError(ctx, st, task, err, hashText(input.SourceText))
+		return
+	}
+	taskResult := map[string]any{
+		"title":            result.Title,
+		"summary":          result.Summary,
+		"content_markdown": result.ContentMarkdown,
+		"content_plain":    result.ContentPlain,
+		"sections":         result.Sections,
+		"source_excerpt":   excerpt(input.SourceText, 200),
+		"can_save":         true,
+	}
+	ok, err := st.FinishAITaskResult(ctx, task.TaskID, "succeeded", 100, "", taskResult, "", "")
+	if err != nil {
+		log.Printf("ai document task finish failed: task_id=%s err=%v", task.TaskID, err)
+		return
+	}
+	if !ok {
+		return
+	}
+	userID := taskUserID(task)
+	_ = st.InsertAIUsageLog(ctx, store.AIUsageLog{
+		TenantCode: task.TenantCode, UserID: task.UserID, Feature: "smart_document", TaskID: task.TaskID,
+		Provider: "deepseek", Model: result.Usage.Model, ProviderRequestID: result.Usage.ProviderRequestID,
+		PromptTokens: result.Usage.PromptTokens, CompletionTokens: result.Usage.CompletionTokens, TotalTokens: result.Usage.TotalTokens,
+		LatencyMS: int(result.Usage.LatencyMS), Status: "succeeded", InputHash: hashText(input.SourceText), OutputHash: hashText(result.ContentMarkdown),
+	})
+	a.publishAITaskEvent(ctx, task.TenantID, task.TenantCode, userID, task.TaskID, "smart_document", result.Title, result.Summary, map[string]any{
+		"task_id": task.TaskID, "feature": "smart_document", "status": "succeeded", "progress": 100, "result_type": "text",
+		"result":     compactAITextResult(result.Title, result.Summary, result.ContentMarkdown, result.ContentPlain),
+		"updated_at": time.Now().UTC().Format(time.RFC3339Nano),
+	})
+}
+
+func (a *API) runAITableTask(ctx context.Context, st *store.Store, task *store.AITask) {
+	var input aiTableTaskSnapshot
+	if !decodeTaskSnapshot(task, &input) {
+		a.finishAITaskFailure(ctx, st, task, reasonMalformedRequest, "invalid task input")
+		return
+	}
+	fileType := strings.ToLower(strings.TrimSpace(input.FileType))
+	fileMeta := st.AIFileMetadata(ctx, input.FileID)
+	profile := a.tableProfileForAIContext(ctx, input.FileID, fileMeta, fileType)
+	result, err := a.ai.AnalyzeTable(ctx, ai.TableAnalysisInput{
+		Topic: input.Topic, FileID: input.FileID, FileName: input.FileName, FileType: fileType, FileSize: input.FileSize, Profile: profile,
+	})
+	if err != nil {
+		a.finishAITaskAIError(ctx, st, task, err, hashText(input.FileID+":"+input.Topic))
+		return
+	}
+	recordID := strings.TrimSpace(input.RecordID)
+	if recordID == "" {
+		recordID = "airec_tbl_" + randomToken(10)
+	}
+	if err := st.SaveAITableAnalysis(ctx, task.TenantCode, taskUserID(task), store.AIRecord{
+		RecordID: recordID, RecordType: "smart_table", Title: result.Title, Summary: excerpt(firstNonEmpty(result.Summary, result.Interpretation), 160),
+		Status: "succeeded", TaskID: task.TaskID,
+	}, store.AITableAnalysis{
+		RecordID: recordID, FileID: input.FileID, FileName: input.FileName, FileType: fileType, FileSize: input.FileSize,
+		Topic: input.Topic, MetricsJSON: mustJSON(result.Metrics, "[]"), PreviewJSON: mustJSON(result.Preview, "{}"),
+		AlertsJSON: mustJSON(result.Alerts, "[]"), Interpretation: result.Interpretation,
+		PromptVersion: "smart_table_v1", Model: result.Usage.Model,
+	}); err != nil {
+		a.finishAITaskFailure(ctx, st, task, reasonAITaskFailed, err.Error())
+		return
+	}
+	taskResult := map[string]any{"record_id": recordID, "title": result.Title, "metrics": result.Metrics, "alerts": result.Alerts, "interpretation": result.Interpretation}
+	ok, err := st.FinishAITaskResult(ctx, task.TaskID, "succeeded", 100, recordID, taskResult, "", "")
+	if err != nil {
+		log.Printf("ai table task finish failed: task_id=%s err=%v", task.TaskID, err)
+		return
+	}
+	if !ok {
+		return
+	}
+	_ = st.InsertAIUsageLog(ctx, store.AIUsageLog{
+		TenantCode: task.TenantCode, UserID: task.UserID, Feature: "smart_table", TaskID: task.TaskID,
+		Provider: "deepseek", Model: result.Usage.Model, ProviderRequestID: result.Usage.ProviderRequestID,
+		PromptTokens: result.Usage.PromptTokens, CompletionTokens: result.Usage.CompletionTokens, TotalTokens: result.Usage.TotalTokens,
+		LatencyMS: int(result.Usage.LatencyMS), Status: "succeeded", InputHash: hashText(input.FileID + ":" + input.Topic), OutputHash: hashText(result.Interpretation),
+	})
+	a.publishAITaskEvent(ctx, task.TenantID, task.TenantCode, taskUserID(task), task.TaskID, "smart_table", result.Title, result.Summary, map[string]any{
+		"task_id": task.TaskID, "feature": "smart_table", "status": "succeeded", "progress": 100, "result_type": "record",
+		"record_id":  recordID,
+		"result":     map[string]any{"record_id": recordID, "title": result.Title, "summary": excerpt(firstNonEmpty(result.Summary, result.Interpretation), 160)},
+		"updated_at": time.Now().UTC().Format(time.RFC3339Nano),
+	})
+}
+
+func (a *API) runAIMeetingSummaryTask(ctx context.Context, st *store.Store, task *store.AITask) {
+	var input aiMeetingSummaryTaskSnapshot
+	if !decodeTaskSnapshot(task, &input) {
+		a.finishAITaskFailure(ctx, st, task, reasonMalformedRequest, "invalid task input")
+		return
+	}
+	userID := taskUserID(task)
+	meeting, err := st.Meeting(ctx, userID, input.MeetingID)
+	if err != nil {
+		a.finishAITaskFailure(ctx, st, task, reasonAITaskFailed, err.Error())
+		return
+	}
+	files, err := st.MeetingNoteFiles(ctx, userID, input.MeetingID)
+	if err != nil {
+		a.finishAITaskFailure(ctx, st, task, reasonAITaskFailed, err.Error())
+		return
+	}
+	fileNames, fileIDs := meetingSummaryFiles(files, input.FileIDs)
+	result, err := a.ai.SummarizeMeeting(ctx, ai.MeetingSummaryInput{
+		MeetingTitle: meeting.Title, StartAt: meeting.StartAt.UTC().Format(time.RFC3339Nano), EndAt: meeting.EndAt.UTC().Format(time.RFC3339Nano),
+		Participants: meetingSummaryParticipants(*meeting), Files: fileNames,
+	})
+	if err != nil {
+		a.finishAITaskAIError(ctx, st, task, err, hashText(input.MeetingID+strings.Join(fileIDs, ",")))
+		return
+	}
+	summaryID := strings.TrimSpace(input.SummaryID)
+	if summaryID == "" {
+		summaryID = "aisum_" + randomToken(10)
+	}
+	if err := st.UpsertAIMeetingSummary(ctx, task.TenantCode, userID, store.AIMeetingSummary{
+		SummaryID: summaryID, MeetingID: input.MeetingID, TaskID: task.TaskID,
+		ConclusionsJSON: mustJSON(result.Conclusions, "[]"), DecisionsJSON: mustJSON(result.Decisions, "[]"),
+		ActionItemsJSON: mustJSON(result.ActionItems, "[]"), SourceFileIDsJSON: mustJSON(fileIDs, "[]"),
+		PromptVersion: "meeting_summary_v1", Model: result.Usage.Model,
+	}); err != nil {
+		a.finishAITaskFailure(ctx, st, task, reasonAITaskFailed, err.Error())
+		return
+	}
+	taskResult := map[string]any{
+		"summary_id": summaryID, "meeting_id": input.MeetingID,
+		"conclusions": result.Conclusions, "decisions": result.Decisions, "action_items": result.ActionItems,
+	}
+	ok, err := st.FinishAITaskResult(ctx, task.TaskID, "succeeded", 100, summaryID, taskResult, "", "")
+	if err != nil {
+		log.Printf("ai meeting summary task finish failed: task_id=%s err=%v", task.TaskID, err)
+		return
+	}
+	if !ok {
+		return
+	}
+	_ = st.InsertAIUsageLog(ctx, store.AIUsageLog{
+		TenantCode: task.TenantCode, UserID: task.UserID, Feature: "meeting_summary", TaskID: task.TaskID,
+		Provider: "deepseek", Model: result.Usage.Model, ProviderRequestID: result.Usage.ProviderRequestID,
+		PromptTokens: result.Usage.PromptTokens, CompletionTokens: result.Usage.CompletionTokens, TotalTokens: result.Usage.TotalTokens,
+		LatencyMS: int(result.Usage.LatencyMS), Status: "succeeded", InputHash: hashText(input.MeetingID + strings.Join(fileIDs, ",")), OutputHash: hashText(mustJSON(result.Conclusions, "[]")),
+	})
+	a.publishAITaskEvent(ctx, task.TenantID, task.TenantCode, userID, task.TaskID, "meeting_summary", meeting.Title, "", map[string]any{
+		"task_id": task.TaskID, "feature": "meeting_summary", "status": "succeeded", "progress": 100, "result_type": "summary",
+		"meeting_id": input.MeetingID, "summary_id": summaryID,
+		"result":     map[string]any{"meeting_id": input.MeetingID, "summary_id": summaryID, "conclusions": result.Conclusions, "decisions": result.Decisions, "action_items": result.ActionItems},
+		"updated_at": time.Now().UTC().Format(time.RFC3339Nano),
+	})
+}
+
+func (a *API) finishAITaskAIError(ctx context.Context, st *store.Store, task *store.AITask, err error, inputHash string) {
+	reason := aiErrorReason(err)
+	a.finishAITaskFailure(ctx, st, task, reason, err.Error())
+	_ = st.InsertAIUsageLog(ctx, store.AIUsageLog{
+		TenantCode: task.TenantCode, UserID: task.UserID, Feature: task.Feature, TaskID: task.TaskID,
+		Provider: "deepseek", Status: "failed", ErrorReason: reason, InputHash: inputHash,
+	})
+}
+
+func (a *API) finishAITaskFailure(ctx context.Context, st *store.Store, task *store.AITask, reason, message string) {
+	ok, err := st.FinishAITaskResult(ctx, task.TaskID, "failed", 100, "", nil, reason, message)
+	if err != nil {
+		log.Printf("ai task failure update failed: task_id=%s reason=%s err=%v", task.TaskID, reason, err)
+		return
+	}
+	if !ok {
+		return
+	}
+	a.publishAITaskEvent(ctx, task.TenantID, task.TenantCode, taskUserID(task), task.TaskID, task.Feature, "AI 任务失败", message, map[string]any{
+		"task_id": task.TaskID, "feature": task.Feature, "status": "failed", "progress": 100, "result_type": "none",
+		"error":      map[string]any{"reason": reason, "message": message},
+		"updated_at": time.Now().UTC().Format(time.RFC3339Nano),
+	})
+}
+
+func (a *API) publishAITaskEvent(ctx context.Context, tenantID int64, tenantCode string, userID int64, taskID, feature, title, summary string, data map[string]any) {
+	if a == nil || a.events == nil || userID <= 0 || strings.TrimSpace(taskID) == "" {
+		return
+	}
+	event := events.Event{
+		EventID:       events.StableEventID(tenantID, "ai", "workbench.ai.task.updated", taskID+":"+stringValue(data["status"]), userID),
+		TenantID:      tenantID,
+		TenantCode:    tenantCode,
+		ActorUserID:   userID,
+		Module:        "ai",
+		Action:        "workbench.ai.task.updated",
+		ResourceID:    taskID,
+		TargetUserIDs: []int64{userID},
+		Title:         strings.TrimSpace(title),
+		Summary:       strings.TrimSpace(summary),
+		OccurredAt:    time.Now().UTC().Round(time.Millisecond),
+		Data:          data,
+	}
+	pubCtx, cancel := context.WithTimeout(ctx, workbenchEventPublishTimeout)
+	defer cancel()
+	if err := a.events.Publish(pubCtx, event); err != nil {
+		log.Printf("ai task event publish failed: event_id=%s task_id=%s feature=%s err=%v", event.EventID, taskID, feature, err)
+	}
+}
+
+func aiTaskSnapshot(value any) (sql.NullString, error) {
+	data, err := json.Marshal(value)
+	if err != nil {
+		return sql.NullString{}, err
+	}
+	return sql.NullString{String: string(data), Valid: true}, nil
+}
+
+func decodeTaskSnapshot(task *store.AITask, dst any) bool {
+	if task == nil || !task.InputSnapshot.Valid || strings.TrimSpace(task.InputSnapshot.String) == "" {
+		return false
+	}
+	return json.Unmarshal([]byte(task.InputSnapshot.String), dst) == nil
+}
+
+func (a *API) tableProfileForAIContext(ctx context.Context, fileID string, meta store.AIFileMetadata, fileType string) ai.TableProfile {
+	if path := resolveTableFilePath(fileID, meta.Location); path != "" {
+		return tableProfileFromPath(path, fileType)
+	}
+	if strings.TrimSpace(meta.Location) == "" {
+		return ai.TableProfile{Parsed: false, Message: "source file metadata is not available"}
+	}
+	path, cleanup, err := a.downloadOSSObjectToTemp(ctx, meta.Location, fileType)
+	if err != nil {
+		return ai.TableProfile{Parsed: false, Message: "source file is not available: " + err.Error()}
+	}
+	defer cleanup()
+	return tableProfileFromPath(path, fileType)
+}
+
+func taskUserID(task *store.AITask) int64 {
+	if task == nil {
+		return 0
+	}
+	userID, _ := strconv.ParseInt(strings.TrimSpace(task.UserID), 10, 64)
+	return userID
+}
+
+func compactAITextResult(title, summary, markdown, plain string) map[string]any {
+	result := map[string]any{"title": title, "summary": summary, "content_plain": plain}
+	if len([]rune(markdown)) <= 8000 {
+		result["content_markdown"] = markdown
+	}
+	return result
+}
+
+func stringValue(value any) string {
+	text, _ := value.(string)
+	return strings.TrimSpace(text)
+}

internal/store/ai.go

@@ -22,6 +22,7 @@ type AICapability struct {
 
 type AITask struct {
 	TaskID        string         `db:"task_id"`
+	TenantID      int64          `db:"tenant_id"`
 	TenantCode    string         `db:"tenant_code"`
 	UserID        string         `db:"user_id"`
 	Feature       string         `db:"feature"`
@@ -30,6 +31,7 @@ type AITask struct {
 	SourceType    string         `db:"source_type"`
 	SourceRefID   string         `db:"source_ref_id"`
 	InputHash     string         `db:"input_hash"`
+	InputSnapshot sql.NullString `db:"input_snapshot"`
 	PromptVersion string         `db:"prompt_version"`
 	Model         string         `db:"model"`
 	ResultRefID   string         `db:"result_ref_id"`
@@ -181,17 +183,59 @@ func (s *Store) CreateAITask(ctx context.Context, task AITask, result any) error
 	if err != nil {
 		return err
 	}
+	var inputSnapshot any
+	if task.InputSnapshot.Valid {
+		inputSnapshot = task.InputSnapshot.String
+	}
 	_, err = s.db.ExecContext(ctx, `INSERT INTO wb_ai_task
 		(task_id,tenant_id,tenant_code,user_id,feature,status,progress,source_type,source_ref_id,input_hash,
 		 input_snapshot,prompt_version,model,result_ref_id,result_json,error_reason,error_message,retry_count,
 		 started_at,finished_at,created_at,updated_at)
-		VALUES(?,?,?,?,?,?,?,?,?,?,NULL,?,?,?,?,?,?,0,?,?,?,?)`,
+		VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,0,?,?,?,?)`,
 		task.TaskID, s.tenant(), task.TenantCode, task.UserID, task.Feature, task.Status, task.Progress,
-		task.SourceType, task.SourceRefID, task.InputHash, task.PromptVersion, task.Model, task.ResultRefID,
+		task.SourceType, task.SourceRefID, task.InputHash, inputSnapshot, task.PromptVersion, task.Model, task.ResultRefID,
 		resultJSON, task.ErrorReason, task.ErrorMessage, task.StartedAt, task.FinishedAt, now, now)
 	return err
 }
 
+func (s *Store) ClaimQueuedAITask(ctx context.Context) (*AITask, error) {
+	tx, err := s.db.BeginTxx(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+	var task AITask
+	err = tx.GetContext(ctx, &task, `SELECT task_id,tenant_id,tenant_code,user_id,feature,status,progress,source_type,source_ref_id,
+			input_hash,input_snapshot,prompt_version,model,result_ref_id,result_json,error_reason,error_message,retry_count,
+			started_at,finished_at,created_at,updated_at
+		FROM wb_ai_task
+		WHERE status='queued' AND feature IN ('smart_document','smart_table','meeting_summary')
+		ORDER BY created_at ASC,id ASC LIMIT 1 FOR UPDATE SKIP LOCKED`)
+	if err == sql.ErrNoRows {
+		return nil, ErrNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+	now := time.Now().UTC().Round(time.Millisecond)
+	res, err := tx.ExecContext(ctx, `UPDATE wb_ai_task SET status='processing',progress=?,started_at=COALESCE(started_at,?),updated_at=?
+		WHERE tenant_id=? AND task_id=? AND status='queued'`, maxInt(task.Progress, 10), now, now, task.TenantID, task.TaskID)
+	if err != nil {
+		return nil, err
+	}
+	if rows, _ := res.RowsAffected(); rows == 0 {
+		return nil, ErrNotFound
+	}
+	if err := tx.Commit(); err != nil {
+		return nil, err
+	}
+	task.Status = "processing"
+	task.Progress = maxInt(task.Progress, 10)
+	task.StartedAt = &now
+	task.UpdatedAt = now
+	return &task, nil
+}
+
 func (s *Store) UpdateAITaskResult(ctx context.Context, taskID, status string, progress int, resultRefID string, result any, reason, message string) error {
 	now := time.Now().UTC().Round(time.Millisecond)
 	resultJSON, err := marshalNullable(result)
@@ -210,6 +254,25 @@ func (s *Store) UpdateAITaskResult(ctx context.Context, taskID, status string, p
 	return nil
 }
 
+func (s *Store) FinishAITaskResult(ctx context.Context, taskID, status string, progress int, resultRefID string, result any, reason, message string) (bool, error) {
+	now := time.Now().UTC().Round(time.Millisecond)
+	resultJSON, err := marshalNullable(result)
+	if err != nil {
+		return false, err
+	}
+	res, err := s.db.ExecContext(ctx, `UPDATE wb_ai_task SET status=?,progress=?,result_ref_id=?,result_json=?,
+			error_reason=?,error_message=?,finished_at=?,updated_at=? WHERE tenant_id=? AND task_id=? AND status='processing'`,
+		status, progress, resultRefID, resultJSON, reason, message, now, now, s.tenant(), taskID)
+	if err != nil {
+		return false, err
+	}
+	rows, err := res.RowsAffected()
+	if err != nil {
+		return false, err
+	}
+	return rows > 0, nil
+}
+
 func (s *Store) CancelAITask(ctx context.Context, userID int64, taskID string) error {
 	now := time.Now().UTC().Round(time.Millisecond)
 	res, err := s.db.ExecContext(ctx, `UPDATE wb_ai_task SET status='canceled',progress=100,finished_at=?,updated_at=?
@@ -226,8 +289,8 @@ func (s *Store) CancelAITask(ctx context.Context, userID int64, taskID string) e
 
 func (s *Store) AITask(ctx context.Context, userID int64, taskID string) (*AITask, error) {
 	var task AITask
-	err := s.db.GetContext(ctx, &task, `SELECT task_id,tenant_code,user_id,feature,status,progress,source_type,source_ref_id,
+	err := s.db.GetContext(ctx, &task, `SELECT task_id,tenant_id,tenant_code,user_id,feature,status,progress,source_type,source_ref_id,
-			input_hash,prompt_version,model,result_ref_id,result_json,error_reason,error_message,retry_count,
+			input_hash,input_snapshot,prompt_version,model,result_ref_id,result_json,error_reason,error_message,retry_count,
 			started_at,finished_at,created_at,updated_at
 		FROM wb_ai_task WHERE tenant_id=? AND user_id=? AND task_id=?`, s.tenant(), formatInt64(userID), taskID)
 	if err == sql.ErrNoRows {
@@ -527,3 +590,10 @@ func marshalNullable(value any) (any, error) {
 func formatInt64(value int64) string {
 	return strconv.FormatInt(value, 10)
 }
+
+func maxInt(a, b int) int {
+	if a > b {
+		return a
+	}
+	return b
+}