diff --git a/deploy/postgres/migrations/2026-09-11-task-run-lineage.sql b/deploy/postgres/migrations/2026-09-11-task-run-lineage.sql new file mode 100644 index 00000000..91904cbc --- /dev/null +++ b/deploy/postgres/migrations/2026-09-11-task-run-lineage.sql @@ -0,0 +1,7 @@ +-- 2026-09-11: 长任务 run 血缘字段扩展 +ALTER TABLE workflow_runs ADD COLUMN IF NOT EXISTS session_id TEXT NOT NULL DEFAULT ''; +ALTER TABLE workflow_runs ADD COLUMN IF NOT EXISTS workspace_id TEXT NOT NULL DEFAULT ''; +ALTER TABLE workflow_runs ADD COLUMN IF NOT EXISTS client_turn_id TEXT NOT NULL DEFAULT ''; +ALTER TABLE workflow_runs ADD COLUMN IF NOT EXISTS assistant_message_id BIGINT; + +CREATE INDEX IF NOT EXISTS idx_workflow_runs_session ON workflow_runs(session_id); diff --git a/internal/bizdev/materials.go b/internal/bizdev/materials.go new file mode 100644 index 00000000..84023a2c --- /dev/null +++ b/internal/bizdev/materials.go @@ -0,0 +1,159 @@ +package bizdev + +import ( + "context" + "errors" + "fmt" + "path/filepath" + "strings" + + "chat-agent-system/internal/squad" + "chat-agent-system/internal/storage" +) + +// ResolveMaterials 将请求材料 + 项目 CAS 资产解析为可注入工作区的 squad 材料。 +// 项目内全部有效资产自动纳入（含私密文档）；显式附件优先并去重。 +func (s *Service) ResolveMaterials(ctx context.Context, req TaskRunRequest) ([]squad.MaterialItem, map[string]any, error) { + manifest := map[string]any{ + "explicit_materials": len(req.Materials), + } + byKey := make(map[string]squad.MaterialItem) + + addItem := func(item squad.MaterialItem) error { + if strings.TrimSpace(item.Filename) == "" { + return errors.New("material filename required") + } + if len(item.Content) == 0 { + return fmt.Errorf("material %q has no extractable content", item.Filename) + } + key := item.Filename + if item.AssetID > 0 { + key = fmt.Sprintf("asset:%d", item.AssetID) + } + byKey[key] = item + return nil + } + + // 1. 项目资产池（不要求用户本轮再次附加） + if s.cas != nil && req.ProjectID > 0 { + refs, err := s.cas.ListProjectRefs(ctx, req.ProjectID) + if err != nil { + return nil, nil, err + } + var assetEntries []map[string]any + for _, ref := range refs { + data, loadedRef, err := s.cas.ReadRefBytes(ctx, ref.ID) + if err != nil { + continue + } + text, err := squad.ExtractTextWithAnchors(loadedRef.FileName, data) + if err != nil { + return nil, nil, fmt.Errorf("extract project asset %q: %w", loadedRef.FileName, err) + } + if strings.TrimSpace(text) == "" { + return nil, nil, fmt.Errorf("project asset %q extracted empty text", loadedRef.FileName) + } + item := squad.MaterialItem{ + Filename: loadedRef.FileName, + Content: []byte(text), + AssetID: loadedRef.ID, + SHA256: loadedRef.BlobSHA256, + OwnerID: loadedRef.UserID, + } + _ = addItem(item) + assetEntries = append(assetEntries, map[string]any{ + "asset_id": loadedRef.ID, + "filename": loadedRef.FileName, + "sha256": loadedRef.BlobSHA256, + "owner_user_id": loadedRef.UserID, + "file_category": loadedRef.FileCategory, + "source": "project_asset_pool", + }) + } + manifest["project_assets"] = assetEntries + } + + // 2. 显式附件（asset_id 优先，其次 inline content） + for _, m := range req.Materials { + if m.AssetID > 0 && s.cas != nil { + data, ref, err := s.cas.ReadRefBytes(ctx, m.AssetID) + if err != nil { + return nil, nil, fmt.Errorf("load attachment asset_id=%d: %w", m.AssetID, err) + } + if ref.ProjectID > 0 && req.ProjectID > 0 && ref.ProjectID != req.ProjectID { + return nil, nil, fmt.Errorf("asset_id=%d belongs to another project", m.AssetID) + } + filename := m.Filename + if filename == "" { + filename = ref.FileName + } + text, err := squad.ExtractTextWithAnchors(filename, data) + if err != nil { + return nil, nil, fmt.Errorf("extract attachment %q: %w", filename, err) + } + if strings.TrimSpace(text) == "" { + return nil, nil, fmt.Errorf("attachment %q extracted empty text", filename) + } + if err := addItem(squad.MaterialItem{ + Filename: filename, + Content: []byte(text), + AssetID: ref.ID, + SHA256: ref.BlobSHA256, + OwnerID: ref.UserID, + }); err != nil { + return nil, nil, err + } + continue + } + + filename := strings.TrimSpace(m.Filename) + if filename == "" { + return nil, nil, errors.New("material filename required when asset_id absent") + } + if len(m.Content) == 0 { + return nil, nil, fmt.Errorf("material %q has empty content and no asset_id", filename) + } + text, err := squad.ExtractTextWithAnchors(filename, m.Content) + if err != nil { + return nil, nil, fmt.Errorf("extract inline material %q: %w", filename, err) + } + if strings.TrimSpace(text) == "" { + return nil, nil, fmt.Errorf("inline material %q extracted empty text", filename) + } + if err := addItem(squad.MaterialItem{ + Filename: filepath.Base(filename), + Content: []byte(text), + }); err != nil { + return nil, nil, err + } + } + + out := make([]squad.MaterialItem, 0, len(byKey)) + for _, item := range byKey { + out = append(out, item) + } + manifest["resolved_count"] = len(out) + return out, manifest, nil +} + +// ValidateTaskRunIdentity 拒绝缺失项目/会话标识的任务启动。 +func ValidateTaskRunIdentity(sessionID string, projectID int64) error { + if strings.TrimSpace(sessionID) == "" { + return errors.New("session_id required for task run") + } + if projectID <= 0 { + return errors.New("project_id required for task run") + } + return nil +} + +// MaterialInputFromRef 由 CAS 引用构造任务材料输入（供上传 API 测试）。 +func MaterialInputFromRef(ref *storage.FileReference) MaterialInput { + if ref == nil { + return MaterialInput{} + } + return MaterialInput{ + Filename: ref.FileName, + AssetID: ref.ID, + } +} diff --git a/internal/bizdev/materials_test.go b/internal/bizdev/materials_test.go new file mode 100644 index 00000000..cc28af95 --- /dev/null +++ b/internal/bizdev/materials_test.go @@ -0,0 +1,60 @@ +package bizdev_test + +import ( + "context" + "testing" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/config" +) + +func TestResolveMaterials_InlineText(t *testing.T) { + cfg := &config.Config{DataDir: t.TempDir()} + svc := bizdev.NewService(cfg) + materials, manifest, err := svc.ResolveMaterials(context.Background(), bizdev.TaskRunRequest{ + Materials: []bizdev.MaterialInput{ + {Filename: "req.txt", Content: []byte("hello project requirements")}, + }, + }) + if err != nil { + t.Fatalf("ResolveMaterials failed: %v", err) + } + if len(materials) != 1 || string(materials[0].Content) != "hello project requirements" { + t.Fatalf("unexpected materials: %+v", materials) + } + if manifest["resolved_count"] != 1 { + t.Fatalf("manifest=%v", manifest) + } +} + +func TestResolveMaterials_RejectEmptyBinary(t *testing.T) { + cfg := &config.Config{DataDir: t.TempDir()} + svc := bizdev.NewService(cfg) + _, _, err := svc.ResolveMaterials(context.Background(), bizdev.TaskRunRequest{ + Materials: []bizdev.MaterialInput{{Filename: "empty.docx", Content: nil}}, + }) + if err == nil { + t.Fatal("expected error for empty binary material") + } +} + +func TestResolveMaterials_ProjectAssetPool(t *testing.T) { + cfg := &config.Config{DataDir: t.TempDir()} + svc := bizdev.NewService(cfg) + cas := svc.CAS() + ref, err := cas.StoreFile(context.Background(), 42, 1, "pool.txt", "material", []byte("pool asset body"), "text/plain") + if err != nil { + t.Fatalf("StoreFile failed: %v", err) + } + _ = ref + materials, manifest, err := svc.ResolveMaterials(context.Background(), bizdev.TaskRunRequest{ProjectID: 42}) + if err != nil { + t.Fatalf("ResolveMaterials failed: %v", err) + } + if len(materials) != 1 || string(materials[0].Content) != "pool asset body" { + t.Fatalf("materials=%+v", materials) + } + if manifest["project_assets"] == nil { + t.Fatalf("expected project_assets in manifest, got %v", manifest) + } +} diff --git a/internal/bizdev/service.go b/internal/bizdev/service.go index 5d94e000..efc4da38 100644 --- a/internal/bizdev/service.go +++ b/internal/bizdev/service.go @@ -182,13 +182,12 @@ func (s *Service) ListStages() []StageMeta { } func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRunResult, error) { - // 转换 MaterialInput 到 squad.MaterialItem - var materials []squad.MaterialItem - for _, m := range req.Materials { - materials = append(materials, squad.MaterialItem{ - Filename: m.Filename, - Content: m.Content, - }) + materials, inputManifest, err := s.ResolveMaterials(ctx, req) + if err != nil { + return nil, err + } + if req.RunID != "" && inputManifest != nil { + inputManifest["run_id"] = req.RunID } // 针对 system_reqs（智能产品经理）：若未显式提供 materials，自动抽取项目事实黑板/集体上下文为需求输入 @@ -558,6 +557,10 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun PriorDocs: req.PriorDocs, UserContext: userCtxStr, MemoryBankFiles: memBank, + SessionID: req.SessionID, + UserID: req.UserID, + ProjectID: req.ProjectID, + RunID: req.RunID, } res, err := s.squad.ExecuteTask(ctx, squadReq) @@ -724,8 +727,10 @@ func (s *Service) executeStageInternal(ctx context.Context, req StageRunRequest) AutoRepair: req.AutoRepair, KnowledgeHits: req.KnowledgeHits, PriorDocs: accumulatedPriorDocs, + SessionID: req.SessionID, ProjectID: req.ProjectID, UserID: req.UserID, + RunID: req.RunID, AuthorID: req.AuthorID, AuthorName: req.AuthorName, ReviewerID: req.ReviewerID, diff --git a/internal/bizdev/types.go b/internal/bizdev/types.go index e001f9ac..cd0804cd 100644 --- a/internal/bizdev/types.go +++ b/internal/bizdev/types.go @@ -34,6 +34,7 @@ type StageMeta struct { type MaterialInput struct { Filename string `json:"filename"` Content []byte `json:"content"` + AssetID int64 `json:"asset_id,omitempty"` } type Attachment = squad.Attachment @@ -56,8 +57,11 @@ type TaskRunRequest struct { AutoRepair bool `json:"auto_repair,omitempty"` KnowledgeHits []squad.KnowledgeHit `json:"knowledge_hits,omitempty"` PriorDocs []squad.PriorDocItem `json:"prior_docs,omitempty"` + SessionID string `json:"session_id,omitempty"` UserID int64 `json:"user_id,omitempty"` ProjectID int64 `json:"project_id,omitempty"` + ClientTurnID string `json:"client_turn_id,omitempty"` + RunID string `json:"run_id,omitempty"` UserContext string `json:"user_context,omitempty"` MemoryBankFiles map[string]string `json:"memory_bank_files,omitempty"` DocID string `json:"doc_id,omitempty"` @@ -105,8 +109,11 @@ type StageRunRequest struct { AutoRepair bool `json:"auto_repair,omitempty"` PriorDocs []squad.PriorDocItem `json:"prior_docs,omitempty"` KnowledgeHits []squad.KnowledgeHit `json:"knowledge_hits,omitempty"` + SessionID string `json:"session_id,omitempty"` UserID int64 `json:"user_id,omitempty"` ProjectID int64 `json:"project_id,omitempty"` + ClientTurnID string `json:"client_turn_id,omitempty"` + RunID string `json:"run_id,omitempty"` DocID string `json:"doc_id,omitempty"` AuthorID int64 `json:"author_id,omitempty"` AuthorName string `json:"author_name,omitempty"` diff --git a/internal/server/asset_handlers.go b/internal/server/asset_handlers.go new file mode 100644 index 00000000..170f6476 --- /dev/null +++ b/internal/server/asset_handlers.go @@ -0,0 +1,133 @@ +package server + +import ( + "io" + "net/http" + "path/filepath" + "strconv" + "strings" +) + +type projectAssetResponse struct { + AssetID int64 `json:"asset_id"` + Filename string `json:"filename"` + SHA256 string `json:"sha256"` + MimeType string `json:"mime_type"` + SizeBytes int64 `json:"size_bytes"` + ProjectID int64 `json:"project_id"` +} + +func (s *Server) handleUploadProjectAsset(w http.ResponseWriter, r *http.Request) { + projectID, err := strconv.ParseInt(r.PathValue("id"), 10, 64) + if err != nil || projectID <= 0 { + http.Error(w, "invalid project id", http.StatusBadRequest) + return + } + userID := parseUserIDHeader(r) + if userID <= 0 { + http.Error(w, "X-User-ID required", http.StatusUnauthorized) + return + } + if err := r.ParseMultipartForm(64 << 20); err != nil { + http.Error(w, "invalid multipart form: "+err.Error(), http.StatusBadRequest) + return + } + file, header, err := r.FormFile("file") + if err != nil { + http.Error(w, "file required", http.StatusBadRequest) + return + } + defer file.Close() + + data, err := io.ReadAll(file) + if err != nil { + http.Error(w, "read file failed", http.StatusBadRequest) + return + } + if len(data) == 0 { + http.Error(w, "cannot store empty file", http.StatusBadRequest) + return + } + + filename := header.Filename + if filename == "" { + filename = "upload.bin" + } + mimeType := header.Header.Get("Content-Type") + if mimeType == "" { + mimeType = mimeFromExt(filename) + } + + cas := s.svc.CAS() + if cas == nil { + http.Error(w, "storage unavailable", http.StatusServiceUnavailable) + return + } + ref, err := cas.StoreFile(r.Context(), projectID, userID, filepath.Base(filename), "material", data, mimeType) + if err != nil { + http.Error(w, "store asset failed: "+err.Error(), http.StatusInternalServerError) + return + } + + writeJSONResponse(w, http.StatusOK, projectAssetResponse{ + AssetID: ref.ID, + Filename: ref.FileName, + SHA256: ref.BlobSHA256, + MimeType: mimeType, + SizeBytes: int64(len(data)), + ProjectID: projectID, + }) +} + +func (s *Server) handleListProjectAssets(w http.ResponseWriter, r *http.Request) { + projectID, err := strconv.ParseInt(r.PathValue("id"), 10, 64) + if err != nil || projectID <= 0 { + http.Error(w, "invalid project id", http.StatusBadRequest) + return + } + cas := s.svc.CAS() + if cas == nil { + writeJSONResponse(w, http.StatusOK, []projectAssetResponse{}) + return + } + refs, err := cas.ListProjectRefs(r.Context(), projectID) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + out := make([]projectAssetResponse, 0, len(refs)) + for _, ref := range refs { + out = append(out, projectAssetResponse{ + AssetID: ref.ID, + Filename: ref.FileName, + SHA256: ref.BlobSHA256, + ProjectID: ref.ProjectID, + }) + } + writeJSONResponse(w, http.StatusOK, out) +} + +func parseUserIDHeader(r *http.Request) int64 { + raw := strings.TrimSpace(r.Header.Get("X-User-ID")) + if raw == "" { + return 0 + } + id, _ := strconv.ParseInt(raw, 10, 64) + return id +} + +func mimeFromExt(name string) string { + ext := strings.ToLower(filepath.Ext(name)) + switch ext { + case ".txt", ".md", ".csv", ".json", ".xml", ".yaml", ".yml", ".log": + return "text/plain" + case ".pdf": + return "application/pdf" + case ".docx": + return "application/vnd.openxmlformats-officedocument.wordprocessingml.document" + case ".xlsx": + return "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + default: + return "application/octet-stream" + } +} diff --git a/internal/server/asset_handlers_test.go b/internal/server/asset_handlers_test.go new file mode 100644 index 00000000..a0e20a57 --- /dev/null +++ b/internal/server/asset_handlers_test.go @@ -0,0 +1,46 @@ +package server + +import ( + "bytes" + "encoding/json" + "mime/multipart" + "net/http" + "net/http/httptest" + "testing" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/config" +) + +func TestUploadProjectAsset(t *testing.T) { + cfg := &config.Config{Env: "test", DataDir: t.TempDir()} + svc := bizdev.NewService(cfg) + srv := NewServer(cfg, svc) + ts := httptest.NewServer(srv.Routes()) + defer ts.Close() + + var body bytes.Buffer + w := multipart.NewWriter(&body) + part, _ := w.CreateFormFile("file", "spec.txt") + _, _ = part.Write([]byte("project asset content")) + _ = w.Close() + + req, _ := http.NewRequest(http.MethodPost, ts.URL+"/api/v1/projects/7/assets", &body) + req.Header.Set("Content-Type", w.FormDataContentType()) + req.Header.Set("X-User-ID", "2") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("upload failed: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status=%d", resp.StatusCode) + } + var out projectAssetResponse + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + t.Fatalf("decode: %v", err) + } + if out.AssetID <= 0 || out.SHA256 == "" { + t.Fatalf("unexpected response: %+v", out) + } +} diff --git a/internal/squad/flex_id.go b/internal/squad/flex_id.go new file mode 100644 index 00000000..7e98de87 --- /dev/null +++ b/internal/squad/flex_id.go @@ -0,0 +1,36 @@ +package squad + +import ( + "encoding/json" + "fmt" + "strconv" +) + +// FlexSessionID 兼容历史 workspace.json 中 session_id 为数字 0 的格式。 +type FlexSessionID string + +func (f FlexSessionID) String() string { + return string(f) +} + +func (f FlexSessionID) MarshalJSON() ([]byte, error) { + return json.Marshal(string(f)) +} + +func (f *FlexSessionID) UnmarshalJSON(b []byte) error { + var s string + if err := json.Unmarshal(b, &s); err == nil { + *f = FlexSessionID(s) + return nil + } + var n int64 + if err := json.Unmarshal(b, &n); err == nil { + if n == 0 { + *f = "" + } else { + *f = FlexSessionID(strconv.FormatInt(n, 10)) + } + return nil + } + return fmt.Errorf("invalid session_id") +} diff --git a/internal/squad/models.go b/internal/squad/models.go index 085d8157..46b62103 100644 --- a/internal/squad/models.go +++ b/internal/squad/models.go @@ -5,7 +5,7 @@ import "time" // WorkspaceMetadata 记录工作区全局元数据 type WorkspaceMetadata struct { ID string `json:"id"` - SessionID int64 `json:"session_id"` + SessionID FlexSessionID `json:"session_id"` ProjectID int64 `json:"project_id"` TaskKey string `json:"task_key"` StageTitle string `json:"stage_title"` @@ -111,6 +111,9 @@ type ContextIndex struct { type MaterialItem struct { Filename string `json:"filename"` Content []byte `json:"content"` + AssetID int64 `json:"asset_id,omitempty"` + SHA256 string `json:"sha256,omitempty"` + OwnerID int64 `json:"owner_user_id,omitempty"` } // KnowledgeHit 法规知识库检索命中条目 diff --git a/internal/squad/service.go b/internal/squad/service.go index 6e889f55..0cc66180 100644 --- a/internal/squad/service.go +++ b/internal/squad/service.go @@ -48,9 +48,13 @@ type SquadRunRequest struct { AutoRepair bool `json:"auto_repair"` KnowledgeHits []KnowledgeHit `json:"knowledge_hits"` PriorDocs []PriorDocItem `json:"prior_docs"` - UserContext string `json:"user_context,omitempty"` + UserContext string `json:"user_context,omitempty"` MemoryBankFiles map[string]string `json:"memory_bank_files,omitempty"` - StepNotify func(StepEvent) `json:"-"` + SessionID string `json:"session_id,omitempty"` + UserID int64 `json:"user_id,omitempty"` + ProjectID int64 `json:"project_id,omitempty"` + RunID string `json:"run_id,omitempty"` + StepNotify func(StepEvent) `json:"-"` } // SquadRunResult 运行结果 @@ -194,6 +198,8 @@ func (s *Service) ExecuteTask(ctx context.Context, req SquadRunRequest) (*SquadR // 1. 初始化工作区目录体系 wsReq := WorkspacePrepareRequest{ WorkDir: workDir, + SessionID: FlexSessionID(req.SessionID), + ProjectID: req.ProjectID, TaskKey: req.TaskKey, StageTitle: taskDisplayTitle(req), ProjectName: req.ProjectName, diff --git a/internal/squad/workspace.go b/internal/squad/workspace.go index cff38920..d72e4ea7 100644 --- a/internal/squad/workspace.go +++ b/internal/squad/workspace.go @@ -14,7 +14,7 @@ import ( // WorkspacePrepareRequest 工作区准备请求 type WorkspacePrepareRequest struct { WorkDir string - SessionID int64 + SessionID FlexSessionID ProjectID int64 TaskKey string StageTitle string diff --git a/internal/squad/workspace_test.go b/internal/squad/workspace_test.go index 19a6e411..2964bc31 100644 --- a/internal/squad/workspace_test.go +++ b/internal/squad/workspace_test.go @@ -16,7 +16,7 @@ func TestWorkspaceEngine_PrepareWorkspace(t *testing.T) { req := squad.WorkspacePrepareRequest{ WorkDir: filepath.Join(tmpDir, "test_workspace_01"), - SessionID: 101, + SessionID: squad.FlexSessionID("101"), ProjectID: 45, TaskKey: "intake", StageTitle: "用户初始需求收集表", diff --git a/internal/storage/cas_assets.go b/internal/storage/cas_assets.go new file mode 100644 index 00000000..d08a1ee4 --- /dev/null +++ b/internal/storage/cas_assets.go @@ -0,0 +1,64 @@ +package storage + +import ( + "context" + "errors" + "fmt" + "io" +) + +// ListProjectRefs 列出项目下全部 CAS 文件引用（智能体任务上下文资产池）。 +func (c *CAS) ListProjectRefs(ctx context.Context, projectID int64) ([]*FileReference, error) { + if projectID <= 0 { + return nil, nil + } + if c.pool != nil { + rows, err := c.pool.Query(ctx, ` + SELECT id, project_id, user_id, blob_sha256, file_name, file_category, COALESCE(summary, ''), created_at + FROM file_references + WHERE project_id = $1 + ORDER BY id ASC + `, projectID) + if err != nil { + return nil, fmt.Errorf("list project refs: %w", err) + } + defer rows.Close() + var refs []*FileReference + for rows.Next() { + ref := &FileReference{} + if err := rows.Scan(&ref.ID, &ref.ProjectID, &ref.UserID, &ref.BlobSHA256, &ref.FileName, &ref.FileCategory, &ref.Summary, &ref.CreatedAt); err != nil { + continue + } + refs = append(refs, ref) + } + return refs, nil + } + + c.mu.RLock() + defer c.mu.RUnlock() + var refs []*FileReference + for _, ref := range c.memoryRefs { + if ref.ProjectID == projectID { + cpy := *ref + refs = append(refs, &cpy) + } + } + return refs, nil +} + +// ReadRefBytes 读取文件引用对应的完整二进制内容。 +func (c *CAS) ReadRefBytes(ctx context.Context, refID int64) ([]byte, *FileReference, error) { + rc, ref, err := c.OpenFileByRef(ctx, refID) + if err != nil { + return nil, nil, err + } + defer rc.Close() + data, err := io.ReadAll(rc) + if err != nil { + return nil, ref, err + } + if len(data) == 0 { + return nil, ref, errors.New("asset file is empty") + } + return data, ref, nil +}