diff --git a/internal/server/intent_plan_handlers.go b/internal/server/intent_plan_handlers.go new file mode 100644 index 00000000..70ced247 --- /dev/null +++ b/internal/server/intent_plan_handlers.go @@ -0,0 +1,185 @@ +package server + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "strconv" + "strings" + + "chat-agent-system/internal/app/intentflow" + "chat-agent-system/internal/authz" +) + +type intentPlanDecisionBody struct { + ExpectedRevision int64 `json:"expected_revision"` + DecisionNonce string `json:"decision_nonce"` + DecisionDigest string `json:"decision_digest"` + Kind string `json:"kind"` + StepIDs []string `json:"step_ids"` + ExcludeUnselected bool `json:"exclude_unselected"` + UserID int64 `json:"user_id"` + ProjectID int64 `json:"project_id"` + SessionID string `json:"session_id"` +} + +func (s *Server) handleGetActiveIntentPlan(w http.ResponseWriter, r *http.Request) { + plans, prin, ok := s.planRequestScope(w, r) + if !ok { + return + } + view, err := plans.GetActivePlan(r.Context(), prin) + if err != nil { + writeIntentPlanError(w, r, plans, prin, "", err) + return + } + writeJSONResponse(w, http.StatusOK, map[string]any{"plan": view}) +} + +func (s *Server) handleGetIntentPlan(w http.ResponseWriter, r *http.Request) { + plans, prin, ok := s.planRequestScope(w, r) + if !ok { + return + } + planID := r.PathValue("plan_id") + view, err := plans.GetPlan(r.Context(), prin, planID) + if err != nil { + writeIntentPlanError(w, r, plans, prin, planID, err) + return + } + writeJSONResponse(w, http.StatusOK, map[string]any{"plan": view}) +} + +func (s *Server) handleIntentPlanDecision(w http.ResponseWriter, r *http.Request) { + plans, prin, ok := s.planRequestScope(w, r) + if !ok { + return + } + var body intentPlanDecisionBody + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + writeJSONResponse(w, http.StatusBadRequest, map[string]string{ + "error_code": "PLAN_DEPENDENCY_INVALID", + "message": "无效的决策请求", + }) + return + } + kind := intentflow.DecisionKind(strings.TrimSpace(body.Kind)) + view, err := plans.DecidePlan(r.Context(), intentflow.DecisionCommand{ + Principal: prin, + PlanID: r.PathValue("plan_id"), + ExpectedRevision: body.ExpectedRevision, + DecisionNonce: body.DecisionNonce, + DecisionDigest: body.DecisionDigest, + Kind: kind, + StepIDs: body.StepIDs, + ExcludeUnselected: body.ExcludeUnselected, + }) + if err != nil { + writeIntentPlanError(w, r, plans, prin, r.PathValue("plan_id"), err) + return + } + writeJSONResponse(w, http.StatusOK, map[string]any{"plan": view}) +} + +func (s *Server) planRequestScope(w http.ResponseWriter, r *http.Request) (intentflow.Service, intentflow.Principal, bool) { + if s == nil || s.intentPlans == nil { + writeJSONResponse(w, http.StatusServiceUnavailable, map[string]string{ + "error_code": "PLAN_NOT_FOUND_OR_HIDDEN", + "message": "计划服务不可用", + }) + return nil, intentflow.Principal{}, false + } + user, ok := s.requireAuthenticatedUser(w, r) + if !ok { + return nil, intentflow.Principal{}, false + } + projectID, err := strconv.ParseInt(strings.TrimSpace(r.URL.Query().Get("project_id")), 10, 64) + sessionID := strings.TrimSpace(r.URL.Query().Get("session_id")) + if err != nil || projectID <= 0 || sessionID == "" { + writeJSONResponse(w, http.StatusBadRequest, map[string]string{ + "error_code": "PLAN_DEPENDENCY_INVALID", + "message": "缺少 project_id 或 session_id", + }) + return nil, intentflow.Principal{}, false + } + if err := s.planProjectAccess(r.Context(), projectID, user.ID); err != nil { + if errors.Is(err, authz.ErrUnauthenticated) { + writeJSONResponse(w, http.StatusUnauthorized, map[string]string{ + "error_code": "ERR_UNAUTHENTICATED", + "message": "请先登录", + }) + return nil, intentflow.Principal{}, false + } + writeJSONResponse(w, http.StatusForbidden, map[string]string{ + "error_code": "ERR_FORBIDDEN", + "message": "无权访问该项目", + }) + return nil, intentflow.Principal{}, false + } + ownerID, ownerErr := s.planSessionOwner(r.Context(), sessionID) + if ownerErr != nil || ownerID <= 0 || ownerID != user.ID { + writeJSONResponse(w, http.StatusNotFound, map[string]string{ + "error_code": "PLAN_NOT_FOUND_OR_HIDDEN", + "message": "计划不存在或不可见", + }) + return nil, intentflow.Principal{}, false + } + env := "dev" + if s.cfg != nil && strings.TrimSpace(s.cfg.Env) != "" { + env = s.cfg.Env + } + return s.intentPlans, intentflow.Principal{ + UserID: user.ID, + Environment: env, + ProjectID: projectID, + SessionID: sessionID, + }, true +} + +func (s *Server) planProjectAccess(ctx context.Context, projectID, userID int64) error { + if s.planAccessFn != nil { + return s.planAccessFn(ctx, projectID, userID) + } + if s.authorizer == nil { + return authz.ErrForbidden + } + return s.authorizer.RequireProjectAccess(ctx, projectID, userID) +} + +func (s *Server) planSessionOwner(ctx context.Context, sessionID string) (int64, error) { + if s.planSessionOwnerFn != nil { + return s.planSessionOwnerFn(ctx, sessionID) + } + if s.ctxMgr == nil { + return 0, intentflow.ErrPlanNotFoundOrHidden + } + return s.ctxMgr.GetSessionOwnerID(ctx, sessionID) +} + +func writeIntentPlanError(w http.ResponseWriter, r *http.Request, plans intentflow.Service, prin intentflow.Principal, planID string, err error) { + code := intentflow.ErrorCode(err) + status := http.StatusInternalServerError + switch { + case errors.Is(err, intentflow.ErrUnauthenticated): + status = http.StatusUnauthorized + case errors.Is(err, intentflow.ErrPlanNotFoundOrHidden): + status = http.StatusNotFound + case errors.Is(err, intentflow.ErrPlanActionForbidden): + status = http.StatusForbidden + case errors.Is(err, intentflow.ErrPlanStaleRevision), + errors.Is(err, intentflow.ErrPlanExpired), + errors.Is(err, intentflow.ErrPlanDecisionConflict), + errors.Is(err, intentflow.ErrPlanActionVersionConflict): + status = http.StatusConflict + case errors.Is(err, intentflow.ErrPlanDependencyInvalid): + status = http.StatusBadRequest + } + body := map[string]any{"error_code": code, "message": code} + if plans != nil && (errors.Is(err, intentflow.ErrPlanStaleRevision) || errors.Is(err, intentflow.ErrPlanDecisionConflict)) && planID != "" { + if latest, getErr := plans.GetPlan(r.Context(), prin, planID); getErr == nil && latest != nil { + body["plan"] = latest + } + } + writeJSONResponse(w, status, body) +} diff --git a/internal/server/intent_plan_handlers_test.go b/internal/server/intent_plan_handlers_test.go new file mode 100644 index 00000000..82d8ad61 --- /dev/null +++ b/internal/server/intent_plan_handlers_test.go @@ -0,0 +1,196 @@ +package server + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "chat-agent-system/internal/app/intentflow" + "chat-agent-system/internal/auth" + "chat-agent-system/internal/authz" + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/config" +) + +func u5Server(t *testing.T) (*Server, *intentflow.Plans, *intentflow.FakeActionOwner, string, int64) { + t.Helper() + cfg := &config.Config{ + DataDir: t.TempDir(), AgentBackend: "stub", AllowInMemory: true, Env: "dev", + ContextDbHost: "127.0.0.1", AdminUsername: "admin", AdminPassword: "admin-secret-password-123", + } + srv := NewServer(cfg, bizdev.MustNewService(t, &config.Config{DataDir: t.TempDir(), AgentBackend: "stub"})) + owner := intentflow.NewFakeActionOwner() + plans := intentflow.NewPlans(&intentflow.SerialUoW{}, intentflow.NewMemoryStore(), intentflow.Options{Owner: owner}) + srv.intentPlans = plans + token, userID := u5Login(t, srv) + srv.planAccessFn = func(ctx context.Context, projectID, userID int64) error { + if projectID == 99 { + return authz.ErrForbidden + } + return nil + } + srv.planSessionOwnerFn = func(ctx context.Context, sessionID string) (int64, error) { + switch sessionID { + case "sess-u5", "sess-other": + return userID, nil + case "sess-foreign": + return userID + 99, nil + default: + return 0, intentflow.ErrPlanNotFoundOrHidden + } + } + return srv, plans, owner, token, userID +} + +func u5Login(t *testing.T, srv *Server) (string, int64) { + t.Helper() + body, _ := json.Marshal(map[string]string{"username": "admin", "password": "admin-secret-password-123"}) + req := httptest.NewRequest(http.MethodPost, "/api/v1/auth/login", bytes.NewReader(body)) + rec := httptest.NewRecorder() + srv.Routes().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("login %d %s", rec.Code, rec.Body.String()) + } + var sess auth.SessionInfo + if err := json.NewDecoder(rec.Body).Decode(&sess); err != nil || sess.Token == "" || sess.User == nil { + t.Fatalf("login decode: %v body=%s", err, rec.Body.String()) + } + return sess.Token, sess.User.ID +} + +func u5Do(srv *Server, method, path, token string, body any) *httptest.ResponseRecorder { + var r *http.Request + if body != nil { + raw, _ := json.Marshal(body) + r = httptest.NewRequest(method, path, bytes.NewReader(raw)) + r.Header.Set("Content-Type", "application/json") + } else { + r = httptest.NewRequest(method, path, nil) + } + if token != "" { + r.Header.Set("Authorization", "Bearer "+token) + } + rec := httptest.NewRecorder() + srv.Routes().ServeHTTP(rec, r) + return rec +} + +func u5Create(t *testing.T, plans *intentflow.Plans, userID int64, projectID int64, sessionID, turn string) *intentflow.PlanView { + t.Helper() + view, err := plans.CreatePlan(context.Background(), intentflow.CreateCommand{ + Principal: intentflow.Principal{UserID: userID, Environment: "dev", ProjectID: projectID, SessionID: sessionID}, + ClientTurnID: turn, + RequestDigest: turn + "-digest", + WritePlan: true, + SafeSummary: "query then generate", + Steps: []intentflow.StepDraft{ + {ActionKey: "regulation.query", ActionDigest: "q1"}, + {ActionKey: "task.run", ActionDigest: "w1"}, + }, + }) + if err != nil { + t.Fatal(err) + } + return view +} + +func TestIntentPlanHTTP_Unauthenticated(t *testing.T) { + srv, _, _, _, _ := u5Server(t) + rec := u5Do(srv, http.MethodGet, "/api/v1/intent-plans/active?project_id=42&session_id=sess-u5", "", nil) + if rec.Code != http.StatusUnauthorized { + t.Fatalf("want 401, got %d %s", rec.Code, rec.Body.String()) + } +} + +func TestIntentPlanHTTP_ActiveWithoutProjection(t *testing.T) { + srv, plans, _, token, userID := u5Server(t) + created := u5Create(t, plans, userID, 42, "sess-u5", "turn-active") + rec := u5Do(srv, http.MethodGet, "/api/v1/intent-plans/active?project_id=42&session_id=sess-u5", token, nil) + if rec.Code != http.StatusOK { + t.Fatalf("active %d %s", rec.Code, rec.Body.String()) + } + var out struct { + Plan *intentflow.PlanView `json:"plan"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil || out.Plan == nil || out.Plan.PlanID != created.PlanID { + t.Fatalf("want stored plan, got %s err=%v", rec.Body.String(), err) + } +} + +func TestIntentPlanHTTP_ScopeNegatives(t *testing.T) { + srv, plans, _, token, userID := u5Server(t) + created := u5Create(t, plans, userID, 42, "sess-u5", "turn-scope") + + if rec := u5Do(srv, http.MethodGet, "/api/v1/intent-plans/active?project_id=99&session_id=sess-u5", token, nil); rec.Code != http.StatusForbidden { + t.Fatalf("non-member want 403, got %d %s", rec.Code, rec.Body.String()) + } + if rec := u5Do(srv, http.MethodGet, "/api/v1/intent-plans/"+created.PlanID+"?project_id=7&session_id=sess-u5", token, nil); rec.Code != http.StatusNotFound { + t.Fatalf("cross-project want 404, got %d %s", rec.Code, rec.Body.String()) + } + if rec := u5Do(srv, http.MethodGet, "/api/v1/intent-plans/"+created.PlanID+"?project_id=42&session_id=sess-other", token, nil); rec.Code != http.StatusNotFound { + t.Fatalf("cross-session want 404, got %d %s", rec.Code, rec.Body.String()) + } + if rec := u5Do(srv, http.MethodGet, "/api/v1/intent-plans/active?project_id=42&session_id=sess-foreign", token, nil); rec.Code != http.StatusNotFound { + t.Fatalf("foreign session want 404, got %d %s", rec.Code, rec.Body.String()) + } +} + +func TestIntentPlanHTTP_IgnoresBodyPrincipalAndSameReceipt(t *testing.T) { + srv, plans, _, token, userID := u5Server(t) + created := u5Create(t, plans, userID, 42, "sess-u5", "turn-dec") + path := "/api/v1/intent-plans/" + created.PlanID + "/decisions?project_id=42&session_id=sess-u5" + body := map[string]any{ + "expected_revision": created.Revision, + "decision_nonce": "nonce-same", + "decision_digest": "digest-same", + "kind": "confirm", + "user_id": userID + 8, + "project_id": 99, + "session_id": "injected", + } + first := u5Do(srv, http.MethodPost, path, token, body) + if first.Code != http.StatusOK { + t.Fatalf("confirm %d %s", first.Code, first.Body.String()) + } + again := u5Do(srv, http.MethodPost, path, token, body) + if again.Code != http.StatusOK { + t.Fatalf("replay %d %s", again.Code, again.Body.String()) + } + var a, b struct { + Plan *intentflow.PlanView `json:"plan"` + } + _ = json.Unmarshal(first.Body.Bytes(), &a) + _ = json.Unmarshal(again.Body.Bytes(), &b) + if a.Plan == nil || b.Plan == nil || a.Plan.Revision != b.Plan.Revision || a.Plan.Status != intentflow.PlanExecuting { + t.Fatalf("same nonce must share receipt a=%+v b=%+v", a.Plan, b.Plan) + } +} + +func TestIntentPlanHTTP_StaleRevision(t *testing.T) { + srv, plans, _, token, userID := u5Server(t) + created := u5Create(t, plans, userID, 42, "sess-u5", "turn-stale") + path := "/api/v1/intent-plans/" + created.PlanID + "/decisions?project_id=42&session_id=sess-u5" + rec := u5Do(srv, http.MethodPost, path, token, map[string]any{ + "expected_revision": created.Revision + 3, + "decision_nonce": "n-stale", + "decision_digest": "d-stale", + "kind": "confirm", + }) + if rec.Code != http.StatusConflict { + t.Fatalf("stale want 409, got %d %s", rec.Code, rec.Body.String()) + } + if !bytes.Contains(rec.Body.Bytes(), []byte("PLAN_STALE_REVISION")) { + t.Fatalf("want stale code, got %s", rec.Body.String()) + } +} + +func TestIntentPlanHTTP_AdminDoesNotBypassMembership(t *testing.T) { + srv, _, _, token, _ := u5Server(t) + rec := u5Do(srv, http.MethodGet, "/api/v1/intent-plans/active?project_id=99&session_id=sess-u5", token, nil) + if rec.Code != http.StatusForbidden { + t.Fatalf("admin must not bypass membership, got %d %s", rec.Code, rec.Body.String()) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 68570335..5411cf6b 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -12,6 +12,7 @@ import ( "chat-agent-system/internal/app" "chat-agent-system/internal/app/chatflow" + "chat-agent-system/internal/app/intentflow" "chat-agent-system/internal/app/stageflow" "chat-agent-system/internal/auth" "chat-agent-system/internal/authz" @@ -38,35 +39,38 @@ import ( // Server 提供基于 HTTP RESTful 与 SSE 的商务与研发参谋接口服务 type Server struct { - rt *app.Runtime - cfg *config.Config - svc *bizdev.Service - pwSvc *projectworkflow.Service - rec intent.Recognizer - engine *coordinator.CoordinatorEngine - authSvc *auth.Service - sessions map[string]*chat.Session - sessionsMu sync.RWMutex - taskRuns *taskrun.Manager - authorizer *authz.Authorizer - authzSvc *authz.Service - wfDefSvc *workflowdef.Service - instSvc *projectinstance.Service - notifySvc *notification.Service - resolver *workspace.Resolver - wsCap *workspace.Capability - cmds *taskrun.Commands - chatLog chatflow.ChatLog - assembler chatflow.ContextAssembler - stageflow *stageflow.UseCase - gov *docgov.GovernanceEngine - rev *docrevision.Service - ctxMgr *ctxmgr.ContextManager - cas *storage.CAS - pool *pgxpool.Pool - freeDoc *freedoc.Service - kb *kb.Client - httpServer *http.Server + rt *app.Runtime + cfg *config.Config + svc *bizdev.Service + pwSvc *projectworkflow.Service + rec intent.Recognizer + engine *coordinator.CoordinatorEngine + authSvc *auth.Service + sessions map[string]*chat.Session + sessionsMu sync.RWMutex + taskRuns *taskrun.Manager + authorizer *authz.Authorizer + authzSvc *authz.Service + wfDefSvc *workflowdef.Service + instSvc *projectinstance.Service + notifySvc *notification.Service + resolver *workspace.Resolver + wsCap *workspace.Capability + cmds *taskrun.Commands + chatLog chatflow.ChatLog + assembler chatflow.ContextAssembler + stageflow *stageflow.UseCase + gov *docgov.GovernanceEngine + rev *docrevision.Service + ctxMgr *ctxmgr.ContextManager + cas *storage.CAS + pool *pgxpool.Pool + freeDoc *freedoc.Service + kb *kb.Client + httpServer *http.Server + intentPlans intentflow.Service + planAccessFn func(ctx context.Context, projectID, userID int64) error + planSessionOwnerFn func(ctx context.Context, sessionID string) (int64, error) } // NewFromRuntime attaches the HTTP adapter to an already-wired composition root. @@ -75,31 +79,32 @@ func NewFromRuntime(rt *app.Runtime) *Server { panic("server: nil runtime") } s := &Server{ - rt: rt, - cfg: rt.Cfg, - svc: rt.Biz, - pwSvc: rt.PW, - rec: rt.Rec, - engine: rt.Engine, - authSvc: rt.Auth, - sessions: make(map[string]*chat.Session), - taskRuns: rt.TaskRuns, - authorizer: rt.Authorizer, - authzSvc: rt.Authz, - wfDefSvc: rt.WFDef, - instSvc: rt.Inst, - notifySvc: rt.Notify, - resolver: rt.Resolver, - wsCap: rt.WSCap, - cmds: rt.Commands, - stageflow: rt.Stageflow, - gov: rt.Gov, - rev: rt.Rev, - ctxMgr: rt.Ctx, - cas: rt.CAS, - pool: rt.Pool, - freeDoc: rt.FreeDoc, - kb: rt.KB, + rt: rt, + cfg: rt.Cfg, + svc: rt.Biz, + pwSvc: rt.PW, + rec: rt.Rec, + engine: rt.Engine, + authSvc: rt.Auth, + sessions: make(map[string]*chat.Session), + taskRuns: rt.TaskRuns, + authorizer: rt.Authorizer, + authzSvc: rt.Authz, + wfDefSvc: rt.WFDef, + instSvc: rt.Inst, + notifySvc: rt.Notify, + resolver: rt.Resolver, + wsCap: rt.WSCap, + cmds: rt.Commands, + stageflow: rt.Stageflow, + gov: rt.Gov, + rev: rt.Rev, + ctxMgr: rt.Ctx, + cas: rt.CAS, + pool: rt.Pool, + freeDoc: rt.FreeDoc, + kb: rt.KB, + intentPlans: rt.IntentPlans, } s.chatLog = chatflow.NewAdapter(s.ctxMgr, s.resolveStageAccess, s.failClosedAuthz()) s.assembler = chatflow.NewAssembler(coordinator.NewContextAssembler(), s.chatLog) @@ -187,6 +192,10 @@ func (s *Server) Routes() http.Handler { mux.HandleFunc("GET /api/v1/chat/stream", s.handleChatStream) mux.HandleFunc("GET /api/v1/chat/turns/{client_turn_id}", s.handleGetChatTurn) + mux.HandleFunc("GET "+intentflow.PathActivePlan, s.handleGetActiveIntentPlan) + mux.HandleFunc("GET "+intentflow.PathPlanByID, s.handleGetIntentPlan) + mux.HandleFunc("POST "+intentflow.PathPlanDecisions, s.handleIntentPlanDecision) + // 5. 独立任务 / 阶段执行 mux.HandleFunc("POST /api/v1/tasks/run", s.handleRunTask) mux.HandleFunc("POST /api/v1/tasks/run/stream", s.handleRunTaskStream)