diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 37eccceb..7f9873e8 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -259,7 +259,7 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { IntendedUse: req.IntendedUse, TargetMarket: req.TargetMarket, Fields: req.Fields, - Backend: req.Backend, + Backend: resolveTaskBackend(s, req.Backend), AutoRepair: req.AutoRepair, UserID: req.UserID, ProjectID: req.ProjectID, @@ -308,7 +308,7 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { TargetMarket: req.TargetMarket, Fields: req.Fields, Materials: req.Materials, - Backend: req.Backend, + Backend: resolveTaskBackend(s, req.Backend), AutoRepair: req.AutoRepair, UserID: req.UserID, ProjectID: req.ProjectID, diff --git a/internal/server/server.go b/internal/server/server.go index e9731e9f..f0e703d4 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -11,6 +11,8 @@ import ( "time" "chat-agent-system/internal/agent" + "chat-agent-system/internal/app/uow" + "chat-agent-system/internal/artifactvault" "chat-agent-system/internal/auth" "chat-agent-system/internal/authz" "chat-agent-system/internal/bizdev" @@ -45,6 +47,8 @@ type Server struct { instSvc *projectinstance.Service notifySvc *notification.Service resolver *workspace.Resolver + wsCap *workspace.Capability + cmds *taskrun.Commands httpServer *http.Server } @@ -93,6 +97,18 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { if pool != nil { taskMgr.SetStore(taskrun.NewPGStore(pool)) } + wsCap := workspace.NewCapability(cfg.DataDir, workspace.NewLeaseTable()) + cmds := &taskrun.Commands{ + Ctx: svc.ContextManager(), + PW: pwSvc, + Env: cfg.Env, + StorageIdentity: "env:" + cfg.Env, + } + if pool != nil { + cmds.UoW = uow.NewPG(pool) + cmds.Fav = &artifactvault.Ingest{Env: cfg.Env} + } + taskMgr.SetCommands(cmds) s := &Server{ cfg: cfg, @@ -109,6 +125,8 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { instSvc: instSvc, notifySvc: notification.NewService(pool), resolver: workspace.NewResolver(cfg.DataDir, pool), + wsCap: wsCap, + cmds: cmds, } if instSvc != nil && instSvc.Available() { svc.SetStageAccessGate(newBizdevStageGate(instSvc, pwSvc)) diff --git a/internal/server/sse.go b/internal/server/sse.go index d19cbc8d..ffad0f19 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -334,7 +334,7 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) { req.DeviceType = q.Get("device_type") req.IntendedUse = q.Get("intended_use") req.TargetMarket = q.Get("target_market") - req.Backend = q.Get("backend") + req.Backend = resolveTaskBackend(s, q.Get("backend")) req.WorkDir = q.Get("work_dir") req.AutoRepair, _ = strconv.ParseBool(q.Get("auto_repair")) if pid := q.Get("project_id"); pid != "" { diff --git a/internal/server/task_finalize.go b/internal/server/task_finalize.go index 1d933c43..1a84ed13 100644 --- a/internal/server/task_finalize.go +++ b/internal/server/task_finalize.go @@ -5,12 +5,11 @@ import ( "crypto/sha256" "encoding/hex" "fmt" - "os" - "path/filepath" "chat-agent-system/internal/bizdev" "chat-agent-system/internal/squad" "chat-agent-system/internal/taskrun" + "chat-agent-system/internal/workspace" ) // FinalizeTaskRunProjection 将任务执行结果投影到 workflow_runs 与聊天助手消息（供回填/管理命令使用）。 @@ -18,101 +17,69 @@ func (s *Server) FinalizeTaskRunProjection(ctx context.Context, record *taskrun. return s.finalizeTaskRun(ctx, record, result) } -// finalizeTaskRun 将任务执行结果投影到 workflow_runs 与聊天助手消息。 +// finalizeTaskRun 是薄适配：产物哈希在事务外经 workspace.Capability 读取，终态命令交给 taskrun.Commands。 func (s *Server) finalizeTaskRun(ctx context.Context, record *taskrun.RunRecord, result *bizdev.TaskRunResult) (string, error) { if record == nil || result == nil { return "failed", fmt.Errorf("finalize: missing record or result") } + if result.DocRecord != nil && result.DocRecord.ContentSHA256 == "" { + if hexSum := s.hashDeliverable(ctx, record, result); hexSum != "" { + result.DocRecord.ContentSHA256 = hexSum + } + } + cmds := s.taskCommands() + if s != nil { + cmds.PW = s.pwSvc + if s.svc != nil { + cmds.Ctx = s.svc.ContextManager() + } + } + return cmds.Finalize(ctx, record, result) +} - workspaceID := result.WorkDir - if workspaceID == "" { - workspaceID = record.WorkspaceID +func (s *Server) taskCommands() *taskrun.Commands { + if s != nil && s.cmds != nil { + return s.cmds } + cmds := &taskrun.Commands{ + PW: s.pwSvc, + Env: "test", + } + if s != nil && s.cfg != nil { + cmds.Env = s.cfg.Env + cmds.StorageIdentity = "env:" + s.cfg.Env + } + if s != nil && s.svc != nil { + cmds.Ctx = s.svc.ContextManager() + } + return cmds +} - terminalStatus := result.TerminalStatus() - summaryContent := result.CompletionSummary() - - docID := "" - artifactTitle := result.ArtifactTitle - sha256Hex := "" - if result.DocRecord != nil { - docID = result.DocRecord.DocID - if result.DocRecord.ContentSHA256 != "" { - sha256Hex = result.DocRecord.ContentSHA256 - } +func (s *Server) hashDeliverable(ctx context.Context, record *taskrun.RunRecord, result *bizdev.TaskRunResult) string { + if s == nil || s.wsCap == nil || record == nil { + return "" } - if docID == "" && record.ProjectID > 0 && record.TaskKey != "" { - docID = fmt.Sprintf("doc_proj%d_%s", record.ProjectID, record.TaskKey) + id := workspace.OpaqueID(result.WorkDir, record.WorkspaceID) + if id == "" { + return "" } - if artifactTitle == "" && record.TaskKey != "" { - artifactTitle = squad.TaskArtifactTitle(record.TaskKey) + env := "" + if s.cfg != nil { + env = s.cfg.Env } - - if sha256Hex == "" && workspaceID != "" { - if computed, err := hashOutputFile(workspaceID); err == nil { - sha256Hex = computed - } + ref := workspace.WorkspaceRef{ + ID: id, + SchemaVersion: workspace.SchemaVersion, + ProjectScope: record.ProjectID, + RunID: record.RunID, + Environment: env, } - - meta := map[string]any{ - "run_id": record.RunID, - "workspace_id": workspaceID, - "task_status": terminalStatus, - "task_key": record.TaskKey, + body, err := s.wsCap.ReadArtifact(ctx, ref, workspace.DeliverableKey) + if err != nil || len(body) == 0 { + return "" } - if record.ProjectID > 0 { - meta["project_id"] = record.ProjectID - } - if art := result.ArtifactMetadata(); art != nil { - if workspaceID != "" { - art["workspace_id"] = workspaceID - } - if sha256Hex != "" { - art["sha256"] = sha256Hex - } - if artifactTitle != "" { - art["title"] = artifactTitle - } - if record.TaskKey != "" { - art["task_key"] = record.TaskKey - } - if docID != "" { - art["doc_id"] = docID - } - meta["artifact"] = art - } else if docID != "" { - artifact := map[string]any{ - "doc_id": docID, - "title": artifactTitle, - "task_key": record.TaskKey, - "workspace_id": workspaceID, - } - if sha256Hex != "" { - artifact["sha256"] = sha256Hex - } - if result.QualityReport != nil { - artifact["quality_passed"] = result.QualityReport.Passed - artifact["quality_score"] = result.QualityReport.Score - if len(result.QualityReport.Violations) > 0 { - artifact["quality_violations"] = result.QualityReport.Violations - } - } - meta["artifact"] = artifact - } - - if s.pwSvc != nil && record.ProjectID > 0 && record.RunID != "" { - if err := s.pwSvc.UpdateWorkflowRun(ctx, record.RunID, terminalStatus, docID, artifactTitle, ""); err != nil { - return "finalizing_failed", fmt.Errorf("update workflow run: %w", err) - } - } - - if s.svc != nil && s.svc.ContextManager() != nil && record.AssistantMsgID > 0 { - if err := s.svc.ContextManager().FinalizeAssistantTaskMessage(ctx, record.AssistantMsgID, summaryContent, meta, terminalStatus); err != nil { - return "finalizing_failed", fmt.Errorf("finalize assistant message: %w", err) - } - } - - return terminalStatus, nil + sum := sha256.Sum256(body) + return hex.EncodeToString(sum[:]) } func resolveTerminalStatus(result *bizdev.TaskRunResult) string { @@ -124,19 +91,6 @@ func buildTaskCompletionSummary(result *bizdev.TaskRunResult, terminalStatus str return result.CompletionSummary() } -func hashOutputFile(workDir string) (string, error) { - path := filepath.Join(workDir, "output.md") - data, err := os.ReadFile(path) - if err != nil { - return "", err - } - if len(data) == 0 { - return "", fmt.Errorf("output.md is empty") - } - sum := sha256.Sum256(data) - return hex.EncodeToString(sum[:]), nil -} - func squadResultToTaskRun(res *squad.SquadRunResult, taskKey string) *bizdev.TaskRunResult { if res == nil { return nil diff --git a/internal/server/task_run_handlers.go b/internal/server/task_run_handlers.go index 5a6e6b9e..a25babf6 100644 --- a/internal/server/task_run_handlers.go +++ b/internal/server/task_run_handlers.go @@ -645,7 +645,7 @@ func (s *Server) buildBizdevTaskRequest(dto TaskRunRequestDTO, sessionID, client TargetMarket: dto.TargetMarket, Fields: dto.Fields, Materials: dto.Materials, - Backend: dto.Backend, + Backend: resolveTaskBackend(s, dto.Backend), AutoRepair: true, // 产品契约：受控小队任务服务端启用修复，不依赖浏览器字段 SessionID: sessionID, UserID: dto.UserID, @@ -664,6 +664,19 @@ func (s *Server) buildBizdevTaskRequest(dto TaskRunRequestDTO, sessionID, client } } +func resolveTaskBackend(s *Server, requested string) string { + configured := "" + if s != nil && s.cfg != nil { + configured = s.cfg.AgentBackend + } + resolved, err := taskrun.ResolveBackend(requested, configured) + if err != nil { + fallback, _ := taskrun.ResolveBackend("", configured) + return fallback + } + return resolved +} + func (s *Server) buildBizdevStageRequest(dto TaskRunRequestDTO, sessionID, clientTurnID, runID string) bizdev.StageRunRequest { return bizdev.StageRunRequest{ Stage: bizdev.MaturityStage(dto.Stage), @@ -674,7 +687,7 @@ func (s *Server) buildBizdevStageRequest(dto TaskRunRequestDTO, sessionID, clien TargetMarket: dto.TargetMarket, Fields: dto.Fields, Materials: dto.Materials, - Backend: dto.Backend, + Backend: resolveTaskBackend(s, dto.Backend), AutoRepair: true, // 产品契约：阶段任务同样启用修复预算 SessionID: sessionID, UserID: dto.UserID, diff --git a/internal/workspace/cleanup.go b/internal/workspace/cleanup.go new file mode 100644 index 00000000..57a49c7f --- /dev/null +++ b/internal/workspace/cleanup.go @@ -0,0 +1,111 @@ +package workspace + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" +) + +// OpaqueID turns a stored work dir or already-opaque id into a WorkspaceRef.ID. +// It never returns a physical path. +func OpaqueID(parts ...string) string { + for _, p := range parts { + p = strings.TrimSpace(p) + if p == "" { + continue + } + slash := filepath.ToSlash(p) + if i := strings.LastIndex(slash, "/workspaces/"); i >= 0 { + rest := slash[i+len("/workspaces/"):] + if rest == "" { + continue + } + return strings.Split(rest, "/")[0] + } + if !strings.Contains(slash, "/") { + return p + } + return filepath.Base(filepath.Clean(p)) + } + return "" +} + +type RetentionReceipt struct { + LogicalID string + Digest string + StorageIdentity string + Verified bool +} + +type FrozenManifest struct { + SHA256 string + StorageIdentity string + WorkspaceID string + Members []FrozenMember +} + +type FrozenMember struct { + LogicalID string + Digest string + Kind string +} + +func memberNeedsRetention(kind string) bool { + switch kind { + case "document", "assistant_turn", "run_manifest", "attachment": + return true + default: + return false + } +} + +var ErrCleanupBlocked = fmt.Errorf("workspace: cleanup blocked until verified receipts match the frozen manifest") + +func AssertCleanupAllowed(frozen FrozenManifest, receipts []RetentionReceipt) error { + if frozen.SHA256 == "" || frozen.StorageIdentity == "" { + return ErrCleanupBlocked + } + byLogical := map[string]RetentionReceipt{} + for _, r := range receipts { + byLogical[r.LogicalID] = r + } + for _, mem := range frozen.Members { + if !memberNeedsRetention(mem.Kind) { + continue + } + r, ok := byLogical[mem.LogicalID] + if !ok || !r.Verified { + return ErrCleanupBlocked + } + if r.Digest != mem.Digest || r.StorageIdentity != frozen.StorageIdentity { + return ErrCleanupBlocked + } + } + return nil +} + +// Cleanup removes the workspace directory only after verified receipts match the frozen manifest. +func (c *Capability) Cleanup(ctx context.Context, ref WorkspaceRef, frozen FrozenManifest, receipts []RetentionReceipt) error { + if err := AssertCleanupAllowed(frozen, receipts); err != nil { + return err + } + if err := c.guard(ctx, ref, false); err != nil { + return err + } + root := filepath.Join(c.dataRoot, "workspaces", ref.ID) + abs, err := filepath.Abs(root) + if err != nil { + return err + } + dataRoot, err := filepath.Abs(c.dataRoot) + if err != nil { + return err + } + rel, err := filepath.Rel(dataRoot, abs) + if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(os.PathSeparator)) { + return ErrPathEscape + } + return os.RemoveAll(abs) +} diff --git a/internal/workspace/cleanup_test.go b/internal/workspace/cleanup_test.go new file mode 100644 index 00000000..381fa673 --- /dev/null +++ b/internal/workspace/cleanup_test.go @@ -0,0 +1,63 @@ +package workspace + +import ( + "context" + "os" + "path/filepath" + "testing" +) + +func TestOpaqueID_StripsPhysicalPath(t *testing.T) { + got := OpaqueID("/data/dev/workspaces/run_abc/output.md", "") + if got != "run_abc" { + t.Fatalf("got %q", got) + } + if OpaqueID("already-opaque") != "already-opaque" { + t.Fatal("opaque id changed") + } +} + +func TestCleanup_RequiresVerifiedReceipts(t *testing.T) { + root := t.TempDir() + cap := NewCapability(root, nil) + ctx := context.Background() + ref, err := cap.Issue(WorkspaceRef{ID: "ws_clean", ProjectScope: 1, RunID: "run_clean"}) + if err != nil { + t.Fatal(err) + } + if err := cap.WriteArtifact(ctx, ref, DeliverableKey, []byte("# body\n")); err != nil { + t.Fatal(err) + } + frozen := FrozenManifest{ + SHA256: "deadbeef", StorageIdentity: "env:dev", WorkspaceID: ref.ID, + Members: []FrozenMember{{LogicalID: "doc_a", Digest: "abc", Kind: "document"}}, + } + if err := cap.Cleanup(ctx, ref, frozen, nil); err == nil { + t.Fatal("cleanup without receipts must fail") + } + if _, err := os.Stat(filepath.Join(root, "workspaces", ref.ID, DeliverableKey)); err != nil { + t.Fatalf("workspace removed without receipts: %v", err) + } + receipts := []RetentionReceipt{{ + LogicalID: "doc_a", Digest: "abc", StorageIdentity: "env:dev", Verified: true, + }} + if err := cap.Cleanup(ctx, ref, frozen, receipts); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(filepath.Join(root, "workspaces", ref.ID)); !os.IsNotExist(err) { + t.Fatalf("workspace still present: %v", err) + } +} + +func TestCleanup_MismatchDigestRejected(t *testing.T) { + frozen := FrozenManifest{ + SHA256: "x", StorageIdentity: "env:dev", + Members: []FrozenMember{{LogicalID: "doc_a", Digest: "abc", Kind: "document"}}, + } + err := AssertCleanupAllowed(frozen, []RetentionReceipt{{ + LogicalID: "doc_a", Digest: "other", StorageIdentity: "env:dev", Verified: true, + }}) + if err == nil { + t.Fatal("mismatch must fail") + } +} diff --git a/internal/workspace/keys.go b/internal/workspace/keys.go new file mode 100644 index 00000000..abc4e8e2 --- /dev/null +++ b/internal/workspace/keys.go @@ -0,0 +1,4 @@ +package workspace + +// DeliverableKey is the only artifact key for the task run markdown deliverable. +const DeliverableKey = "output.md"