feat(agent): project asset upload and squad materials assembly

Register uploads via CAS, resolve project asset pools into task materials
with server-side extraction, and propagate asset_id through the bizdev/squad chain.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
WeiCHEN
2026-09-13 02:50:22 +00:00
committed by weichen
co-authored by Cursor
parent 6f3a4d4c80
commit 908d4bdb2e
13 changed files with 538 additions and 12 deletions
@@ -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);
+159
View File
@@ -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,
}
}
+60
View File
@@ -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)
}
}
+12 -7
View File
@@ -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,
+7
View File
@@ -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"`
+133
View File
@@ -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"
}
}
+46
View File
@@ -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)
}
}
+36
View File
@@ -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")
}
+4 -1
View File
@@ -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 法规知识库检索命中条目
+8 -2
View File
@@ -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,
+1 -1
View File
@@ -14,7 +14,7 @@ import (
// WorkspacePrepareRequest 工作区准备请求
type WorkspacePrepareRequest struct {
WorkDir string
SessionID int64
SessionID FlexSessionID
ProjectID int64
TaskKey string
StageTitle string
+1 -1
View File
@@ -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: "用户初始需求收集表",
+64
View File
@@ -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
}