feat(api): expose the trace index, reopen, and recheck routes

Use the record tables when migration 00059 is present, and hide statements from documents the caller cannot see.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-20 15:35:07 +00:00
co-authored by Cursor
parent 5e05d966b4
commit e4b4e0dde8
6 changed files with 416 additions and 0 deletions
+19
View File
@@ -88,6 +88,7 @@ imports:
- projectworkflow
- storage
- taskrun
- traceimpact
- workflowdef
- workspace
- sysadmin
@@ -200,6 +201,7 @@ imports:
- stream/transport
- swboot
- taskrun
- traceimpact
- upload
- workflowdef
- workspace
@@ -251,10 +253,27 @@ imports:
- app/uow
sysadmin:
- config
traceimpact: []
table_owners:
document_records:
- docgov
project_trace_items:
- traceimpact
project_trace_item_revisions:
- traceimpact
project_trace_edges:
- traceimpact
project_trace_edge_decisions:
- traceimpact
project_trace_index_state:
- traceimpact
stage_reopen_requests:
- traceimpact
stage_recheck_marks:
- traceimpact
trace_revision_proposals:
- traceimpact
chat_messages:
- ctxmgr
chat_turns:
+25
View File
@@ -36,6 +36,7 @@ import (
"chat-agent-system/internal/storage"
"chat-agent-system/internal/sysadmin"
"chat-agent-system/internal/taskrun"
"chat-agent-system/internal/traceimpact"
"chat-agent-system/internal/workflowdef"
"chat-agent-system/internal/workspace"
@@ -71,6 +72,7 @@ type Runtime struct {
Pool *pgxpool.Pool
Gov *docgov.GovernanceEngine
Rev *docrevision.Service
Trace *traceimpact.Service
Ctx *ctxmgr.ContextManager
CAS *storage.CAS
FreeDoc *freedoc.Service
@@ -191,6 +193,28 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) {
if instSvc != nil && instSvc.Available() {
coordEngine.SetInstanceService(instSvc)
}
traceSvc := traceimpact.Open(pool)
traceSvc.SetStageReopener(traceStage{inst: instSvc})
traceSvc.SetDocUnlocker(traceUnlock{gov: parts.Gov})
if instSvc != nil {
instSvc.SetRecheckGate(traceRecheck{svc: traceSvc})
}
coordEngine.SetImpactActions(traceChat{svc: traceSvc})
if pool != nil {
access := authz.NewAuthorizer(pool)
coordEngine.SetProjectAccess(func(ctx context.Context, projectID, userID int64) error {
return access.RequireProjectAccess(ctx, projectID, userID)
})
}
if parts.Rev != nil {
parts.Rev.CoverageResolver = func(ctx context.Context, docID string) string {
projectID := projectIDFromControlledDoc(docID)
if projectID == 0 {
return traceimpact.ImpactIndexPending
}
return traceSvc.CoverageForDoc(ctx, projectID, docID)
}
}
rt := &Runtime{
Cfg: cfg,
@@ -212,6 +236,7 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) {
Pool: pool,
Gov: parts.Gov,
Rev: parts.Rev,
Trace: traceSvc,
Ctx: parts.Ctx,
CAS: parts.CAS,
FreeDoc: parts.FreeDoc,
+98
View File
@@ -0,0 +1,98 @@
package app
import (
"context"
"errors"
"fmt"
"strconv"
"strings"
"chat-agent-system/internal/coordinator"
"chat-agent-system/internal/docgov"
"chat-agent-system/internal/projectinstance"
"chat-agent-system/internal/traceimpact"
)
type traceRecheck struct{ svc *traceimpact.Service }
func (t traceRecheck) StageNeedsAck(ctx context.Context, projectID int64, stageKey string) (bool, error) {
if t.svc == nil {
return false, nil
}
return t.svc.NeedsRecheckAck(ctx, projectID, stageKey)
}
func (t traceRecheck) Acknowledge(ctx context.Context, projectID int64, stageKey string, userID int64) error {
if t.svc == nil {
return nil
}
return t.svc.AcknowledgeRecheck(ctx, projectID, stageKey, userID)
}
type traceStage struct{ inst *projectinstance.Service }
func (t traceStage) ReopenCompletedStage(ctx context.Context, instanceID int64, stageKey string, docTaskKeys []string, operatorID int64) error {
if t.inst == nil {
return fmt.Errorf("trace: instance service unavailable")
}
return t.inst.ReopenStage(ctx, instanceID, stageKey, docTaskKeys, operatorID)
}
type traceUnlock struct{ gov *docgov.GovernanceEngine }
func (t traceUnlock) ReopenApprovedDocs(ctx context.Context, projectID int64, docTaskKeys []string, actorID int64, reason string) error {
if t.gov == nil {
return nil
}
for _, key := range docTaskKeys {
docID := fmt.Sprintf("doc_proj%d_%s", projectID, key)
_, err := t.gov.ReopenForRevision(ctx, docID, actorID, "creator", reason)
if err == nil || errors.Is(err, docgov.ErrDocumentNotFound) {
continue
}
if errors.Is(err, docgov.ErrInvalidStateTransition) {
continue
}
return err
}
return nil
}
type traceChat struct{ svc *traceimpact.Service }
func (t traceChat) AnalyzeImpact(ctx context.Context, projectID, _ int64, text string) (string, error) {
if t.svc == nil {
return "", fmt.Errorf("trace: unavailable")
}
rep, err := t.svc.Analyze(ctx, projectID, traceimpact.DocRef{DocID: "chat", Markdown: text, Visible: true}, nil)
if err != nil {
return "", err
}
return rep.Summary, nil
}
func projectIDFromControlledDoc(docID string) int64 {
rest := strings.TrimPrefix(docID, "doc_proj")
if rest == docID {
return 0
}
i := strings.Index(rest, "_")
if i <= 0 {
return 0
}
n, err := strconv.ParseInt(rest[:i], 10, 64)
if err != nil {
return 0
}
return n
}
func (t traceChat) ProposeRevision(ctx context.Context, projectID, userID int64, text string, confirmed bool) (string, error) {
if t.svc == nil {
return "", fmt.Errorf("trace: unavailable")
}
id, err := t.svc.ProposeRevision(ctx, projectID, userID, text, confirmed)
if errors.Is(err, traceimpact.ErrNeedsConfirm) {
return "", coordinator.ErrImpactNeedsConfirm
}
return id, err
}
@@ -194,6 +194,10 @@ func writeProjectInstanceError(w http.ResponseWriter, err error) {
writeSafeError(w, nil, http.StatusConflict, `{"error":"STAGE_GATE_NOT_READY","message":"阶段门禁未满足,必需文档尚未全部批准"}`)
case errors.Is(err, projectinstance.ErrStageNotActive):
writeSafeError(w, nil, http.StatusConflict, `{"error":"STAGE_NOT_ACTIVE","message":"目标阶段不是当前活动阶段"}`)
case errors.Is(err, projectinstance.ErrStageNotCompleted):
writeSafeError(w, nil, http.StatusConflict, `{"error":"STAGE_NOT_COMPLETED","message":"只能再打开已完成的阶段"}`)
case errors.Is(err, projectinstance.ErrRecheckAckRequired):
writeSafeError(w, nil, http.StatusConflict, `{"error":"RECHECK_ACK_REQUIRED","message":"该阶段有待复核标记,推进前须确认已知悉"}`)
case errors.Is(err, projectinstance.ErrInvalidStage):
writeSafeError(w, nil, http.StatusBadRequest, `{"error":"ERR_INVALID_STAGE","message":"无效的阶段"}`)
case errors.Is(err, projectinstance.ErrInstanceRevisionConflict):
+7
View File
@@ -211,6 +211,13 @@ func (s *Server) Routes() http.Handler {
mux.HandleFunc("POST /api/v1/projects/{id}/workflows/{workflow_key}/run", s.handleRunProjectWorkflow)
mux.HandleFunc("GET /api/v1/projects/{id}/workflows/{workflow_key}/runs", s.handleListProjectWorkflowRuns)
mux.HandleFunc("GET /api/v1/projects/{id}/lineage", s.handleGetProjectLineage)
mux.HandleFunc("GET /api/v1/projects/{id}/trace", s.handleProjectTrace)
mux.HandleFunc("POST /api/v1/projects/{id}/trace/index", s.handleProjectTraceIndex)
mux.HandleFunc("POST /api/v1/projects/{id}/trace/analyze", s.handleProjectTraceAnalyze)
mux.HandleFunc("POST /api/v1/projects/{id}/trace/edges/{edgeID}/decision", s.handleProjectTraceDecision)
mux.HandleFunc("POST /api/v1/projects/{id}/trace/reopen", s.handleProjectTraceReopen)
mux.HandleFunc("POST /api/v1/projects/{id}/trace/reopen/{requestID}/approve", s.handleProjectTraceReopenApprove)
mux.HandleFunc("POST /api/v1/projects/{id}/trace/recheck/{markID}/clear", s.handleProjectTraceRecheckClear)
mux.HandleFunc("GET /api/v1/projects/{id}/documents", s.handleListProjectWorkflowDocuments)
mux.HandleFunc("GET /api/v1/projects/{id}/notifications", s.handleListProjectNotifications)
+263
View File
@@ -0,0 +1,263 @@
package server
import (
"encoding/json"
"io"
"net/http"
"strconv"
"chat-agent-system/internal/auth"
"chat-agent-system/internal/traceimpact"
)
func (s *Server) handleProjectTrace(w http.ResponseWriter, r *http.Request) {
projectID, user, ok := s.traceProject(w, r)
if !ok {
return
}
items, err := s.rt.Trace.ListItems(r.Context(), projectID)
if err != nil {
writeSafeError(w, r, http.StatusInternalServerError, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`)
return
}
visible, hidden := traceimpact.RedactItems(items, func(docID string) bool {
return s.traceDocVisible(r, projectID, user.ID, docID)
})
index, err := s.rt.Trace.ListIndex(r.Context(), projectID)
if err != nil {
writeSafeError(w, r, http.StatusInternalServerError, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`)
return
}
marks, err := s.rt.Trace.ListOpenMarks(r.Context(), projectID, "")
if err != nil {
writeSafeError(w, r, http.StatusInternalServerError, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`)
return
}
writeJSONResponse(w, http.StatusOK, map[string]any{
"items": visible,
"hidden_count": hidden,
"index": index,
"recheck_marks": marks,
})
}
func (s *Server) handleProjectTraceIndex(w http.ResponseWriter, r *http.Request) {
projectID, _, ok := s.traceProject(w, r)
if !ok {
return
}
var body traceimpact.DocRef
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"无效的 JSON 请求体"}`)
return
}
st, err := s.rt.Trace.IndexDocument(r.Context(), projectID, body)
if err != nil {
writeJSONResponse(w, http.StatusBadRequest, map[string]any{"error": "TRACE_INDEX_FAILED", "message": err.Error(), "index": st})
return
}
writeJSONResponse(w, http.StatusOK, st)
}
func (s *Server) handleProjectTraceAnalyze(w http.ResponseWriter, r *http.Request) {
projectID, _, ok := s.traceProject(w, r)
if !ok {
return
}
var body struct {
traceimpact.DocRef
Others []traceimpact.DocRef `json:"others"`
}
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"无效的 JSON 请求体"}`)
return
}
rep, err := s.rt.Trace.Analyze(r.Context(), projectID, body.DocRef, body.Others)
if err != nil {
writeSafeError(w, r, http.StatusInternalServerError, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`)
return
}
writeJSONResponse(w, http.StatusOK, rep)
}
func (s *Server) handleProjectTraceDecision(w http.ResponseWriter, r *http.Request) {
projectID, user, ok := s.traceProject(w, r)
if !ok {
return
}
edgeID, err := strconv.ParseInt(r.PathValue("edgeID"), 10, 64)
if err != nil || edgeID <= 0 {
writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"invalid edge id"}`)
return
}
var body struct {
DownstreamDoc string `json:"downstream_doc"`
DownstreamAuthor int64 `json:"downstream_author"`
Decision string `json:"decision"`
Reason string `json:"reason"`
}
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"无效的 JSON 请求体"}`)
return
}
if !s.traceDocVisible(r, projectID, user.ID, body.DownstreamDoc) {
writeSafeError(w, r, http.StatusForbidden, `{"error":"ERR_FORBIDDEN","message":"看不见下游文档,不能裁决"}`)
return
}
creator, _ := s.pwSvc.IsProjectCreator(r.Context(), projectID, user.ID)
if err := s.rt.Trace.DecideCandidate(r.Context(), edgeID, user.ID, creator, body.DownstreamDoc, body.DownstreamAuthor, body.Decision, body.Reason); err != nil {
writeJSONResponse(w, http.StatusForbidden, map[string]string{"error": "TRACE_DECISION_DENIED", "message": err.Error()})
return
}
writeJSONResponse(w, http.StatusOK, map[string]string{"status": "recorded"})
}
func (s *Server) handleProjectTraceReopen(w http.ResponseWriter, r *http.Request) {
projectID, user, ok := s.traceProject(w, r)
if !ok {
return
}
var req traceimpact.ReopenRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"无效的 JSON 请求体"}`)
return
}
req.ProjectID = projectID
req.RequestedBy = user.ID
creator, _ := s.pwSvc.IsProjectCreator(r.Context(), projectID, user.ID)
if !s.callerMayRequestReopen(r, projectID, user.ID, creator, req) {
writeSafeError(w, r, http.StatusForbidden, `{"error":"ERR_FORBIDDEN","message":"只有该阶段文档职责人或创建人可以申请再打开"}`)
return
}
stored, err := s.rt.Trace.RequestReopen(r.Context(), req, creator)
if err != nil {
writeJSONResponse(w, http.StatusBadRequest, map[string]string{"error": "TRACE_REOPEN_FAILED", "message": err.Error()})
return
}
writeJSONResponse(w, http.StatusOK, stored)
}
func (s *Server) handleProjectTraceReopenApprove(w http.ResponseWriter, r *http.Request) {
projectID, user, ok := s.traceProject(w, r)
if !ok {
return
}
creator, _ := s.pwSvc.IsProjectCreator(r.Context(), projectID, user.ID)
if !creator {
writeSafeError(w, r, http.StatusForbidden, `{"error":"ERR_NOT_PROJECT_CREATOR","message":"仅项目创建人可批准再打开"}`)
return
}
requestID, err := strconv.ParseInt(r.PathValue("requestID"), 10, 64)
if err != nil || requestID <= 0 {
writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"invalid request id"}`)
return
}
existing, err := s.rt.Trace.GetReopen(r.Context(), requestID)
if err != nil || existing.ProjectID != projectID {
writeSafeError(w, r, http.StatusNotFound, `{"error":"ERR_NOT_FOUND","message":"申请不属于该项目"}`)
return
}
stored, err := s.rt.Trace.ApproveReopen(r.Context(), requestID, user.ID)
if err != nil {
writeJSONResponse(w, http.StatusConflict, map[string]string{"error": "TRACE_REOPEN_FAILED", "message": err.Error()})
return
}
writeJSONResponse(w, http.StatusOK, stored)
}
func (s *Server) handleProjectTraceRecheckClear(w http.ResponseWriter, r *http.Request) {
projectID, _, ok := s.traceProject(w, r)
if !ok {
return
}
markID, err := strconv.ParseInt(r.PathValue("markID"), 10, 64)
if err != nil || markID <= 0 {
writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"invalid mark id"}`)
return
}
var body struct {
Reason string `json:"reason"`
}
if err := json.NewDecoder(r.Body).Decode(&body); err != nil && err != io.EOF {
writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"无效的 JSON 请求体"}`)
return
}
marks, err := s.rt.Trace.ListOpenMarks(r.Context(), projectID, "")
if err != nil {
writeSafeError(w, r, http.StatusInternalServerError, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`)
return
}
found := false
for _, m := range marks {
if m.ID == markID {
found = true
break
}
}
if !found {
writeSafeError(w, r, http.StatusNotFound, `{"error":"ERR_NOT_FOUND","message":"未找到该项目的打开标记"}`)
return
}
if err := s.rt.Trace.ClearRecheck(r.Context(), markID, body.Reason); err != nil {
writeJSONResponse(w, http.StatusBadRequest, map[string]string{"error": "TRACE_RECHECK_CLEAR_FAILED", "message": err.Error()})
return
}
writeJSONResponse(w, http.StatusOK, map[string]string{"status": "cleared"})
}
func (s *Server) traceProject(w http.ResponseWriter, r *http.Request) (int64, *auth.UserAccount, bool) {
projectID, err := strconv.ParseInt(r.PathValue("id"), 10, 64)
if err != nil || projectID <= 0 {
writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"invalid project id"}`)
return 0, nil, false
}
if s.rt == nil || s.rt.Trace == nil {
writeSafeError(w, r, http.StatusServiceUnavailable, `{"error":"ERR_NOT_AVAILABLE","message":"追踪索引不可用"}`)
return 0, nil, false
}
user, ok := s.requireProjectAccess(w, r, projectID)
if !ok {
return 0, nil, false
}
return projectID, user, true
}
func (s *Server) callerMayRequestReopen(r *http.Request, projectID, userID int64, creator bool, req traceimpact.ReopenRequest) bool {
if creator {
return true
}
if s.instSvc == nil {
return false
}
settings, err := s.instSvc.GetSettings(r.Context(), projectID, userID, false)
if err != nil || settings == nil || settings.Plan == nil {
return false
}
var seats []traceimpact.DutySeat
for _, stage := range settings.Plan.Stages {
if stage.StageKey != req.StageKey {
continue
}
for _, task := range stage.DocTasks {
seats = append(seats, traceimpact.DutySeat{
TaskKey: task.TaskKey, AuthorID: task.Duties.AuthorUserID,
ReviewerID: task.Duties.ReviewerUserID, ApproverID: task.Duties.ApproverUserID,
})
}
}
return traceimpact.MayRequestReopen(userID, false, req.DocTasks, seats)
}
func (s *Server) traceDocVisible(r *http.Request, projectID, userID int64, docID string) bool {
if docID == "" || docID == "chat" {
return true
}
parsedProject, taskKey, ok := parseCanonicalWorkflowDocID(docID)
if !ok {
return false
}
if parsedProject != projectID {
return false
}
return s.allowWorkflowDocTaskAccess(r.Context(), projectID, userID, taskKey)
}