diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index 14748570..7540b0ca 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -56,12 +56,36 @@ imports: - storage db: [] docduty: [] - app: [] + app: + - agent + - app/chatflow + - app/docnotify + - app/stageflow + - app/uow + - artifactvault + - auth + - authz + - bizdev + - config + - coordinator + - docgov + - intent + - notification + - observability + - projectinstance + - projectworkflow + - taskrun + - workflowdef + - workspace app/chatflow: - bizdev - coordinator - ctxmgr - projectinstance + app/docnotify: + - docgov + - docrevision + - notification app/docflow: - app/uow - docgov @@ -69,6 +93,8 @@ imports: - docrevision - events app/uow: [] + app/stageflow: + - bizdev docquery: - docgov - docrevision @@ -105,8 +131,10 @@ imports: replyexport: [] server: - agent + - app - app/chatflow - app/docflow + - app/stageflow - app/uow - auth - authz @@ -128,6 +156,7 @@ imports: - replyexport - squad - stream + - stream/transport - taskrun - upload - workflowdef @@ -142,6 +171,8 @@ imports: - workspace storage: [] stream: [] + stream/transport: + - stream taskrun: - app/uow - artifactvault @@ -256,7 +287,7 @@ event_producers: event_consumers: outbox.notify: - - server + - app/docnotify - notification unwrap_owners: @@ -443,3 +474,9 @@ baseline_counts: sql_write: 5 path: 16 event: 0 + +# U12: kinds already at zero production findings fail --mode block. +# sql_write/path keep dated exceptions until their owner milestones clear them. +block_kinds: + - import + - event diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 432cd14a..feb91fe5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -45,6 +45,9 @@ jobs: TEST_DATABASE_URL: postgres://app:ci_secret_db_pass@localhost:5432/agent_system_ci?sslmode=disable run: go test -short -timeout 20m ./... + - name: arch-check (block cleared kinds) + run: ./scripts/arch-check.sh --mode block + backup-restore-drill: runs-on: ubuntu-latest services: diff --git a/cmd/bizdev-cli/main.go b/cmd/bizdev-cli/main.go index 1b87b878..f40a8fc1 100644 --- a/cmd/bizdev-cli/main.go +++ b/cmd/bizdev-cli/main.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "chat-agent-system/internal/app/stageflow" "chat-agent-system/internal/bizdev" "chat-agent-system/internal/chat" "chat-agent-system/internal/config" @@ -240,7 +241,7 @@ func main() { return } - res, err := svc.ExecuteTask(context.Background(), bizdev.TaskRunRequest{ + res, err := stageflow.New(svc).ExecuteTask(context.Background(), bizdev.TaskRunRequest{ TaskKey: *taskKey, ProjectName: *project, CustomerName: *customer, @@ -347,7 +348,7 @@ func handleStageRun( ctx, _ = stream.WithListener(ctx, verboseMode, listener) } - stageRes, err := svc.ExecuteStage(ctx, bizdev.StageRunRequest{ + stageRes, err := stageflow.New(svc).ExecuteStage(ctx, bizdev.StageRunRequest{ Stage: bizdev.MaturityStage(stageName), ProjectName: project, CustomerName: customer, diff --git a/cmd/server/main.go b/cmd/server/main.go index 39e2a761..71b260ab 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -10,7 +10,7 @@ import ( "syscall" "time" - "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/app" "chat-agent-system/internal/config" "chat-agent-system/internal/server" ) @@ -38,12 +38,16 @@ func main() { cfg.AgentBackend = *backend } - svc, err := bizdev.NewService(cfg) + rt, err := app.New(cfg) if err != nil { fmt.Fprintf(os.Stderr, "业务服务初始化失败: %v\n", err) os.Exit(1) } - srv := server.NewServer(cfg, svc) + if err := rt.Start(context.Background()); err != nil { + fmt.Fprintf(os.Stderr, "后台 worker 启动失败: %v\n", err) + os.Exit(1) + } + srv := server.NewFromRuntime(rt) addr := fmt.Sprintf("%s:%d", cfg.HTTPHost, cfg.HTTPPort) fmt.Println("==================================================================") diff --git a/internal/app/app.go b/internal/app/app.go new file mode 100644 index 00000000..1b400912 --- /dev/null +++ b/internal/app/app.go @@ -0,0 +1,305 @@ +package app + +import ( + "context" + "fmt" + "log/slog" + "sync" + "time" + + "chat-agent-system/internal/agent" + "chat-agent-system/internal/app/chatflow" + "chat-agent-system/internal/app/docnotify" + "chat-agent-system/internal/app/stageflow" + "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" + "chat-agent-system/internal/config" + "chat-agent-system/internal/coordinator" + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/intent" + "chat-agent-system/internal/notification" + "chat-agent-system/internal/observability" + "chat-agent-system/internal/projectinstance" + "chat-agent-system/internal/projectworkflow" + "chat-agent-system/internal/taskrun" + "chat-agent-system/internal/workflowdef" + "chat-agent-system/internal/workspace" +) + +const defaultReminderPollInterval = 15 * time.Minute + +// Runtime is the unique composition root. cmd/server, tests and workers +// construct it once; HTTP adapters must not re-wire domain services. +type Runtime struct { + Cfg *config.Config + Biz *bizdev.Service + Engine *coordinator.CoordinatorEngine + Auth *auth.Service + PW *projectworkflow.Service + TaskRuns *taskrun.Manager + Authorizer *authz.Authorizer + Authz *authz.Service + WFDef *workflowdef.Service + Inst *projectinstance.Service + Notify *notification.Service + Resolver *workspace.Resolver + WSCap *workspace.Capability + Commands *taskrun.Commands + Rec intent.Recognizer + Stageflow *stageflow.UseCase + DocNotify *docnotify.Subscriber + ChatLog chatflow.ChatLog + Assembler chatflow.ContextAssembler + + mu sync.Mutex + running bool + subscribers int + starts int + notifyBound bool + hookBound bool + reminderCancel context.CancelFunc +} + +func New(cfg *config.Config) (*Runtime, error) { + return Assemble(cfg, nil) +} + +// Assemble wires domain services. svc may be injected by tests. +func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) { + if cfg == nil { + cfg, _ = config.Load("") + } + if cfg != nil { + observability.InitLogger(cfg.Env) + } + if svc == nil { + var err error + svc, err = bizdev.NewService(cfg) + if err != nil { + return nil, err + } + } + + rec := intent.NewRecognizer() + dialogueBackend := agent.NewBackend(cfg.AgentBackend, cfg) + coordEngine := coordinator.NewCoordinatorEngineWithDialogue(svc, dialogueBackend, cfg.DataDir) + pool := svc.Pool() + stores, storeErr := auth.NewStores(context.Background(), pool, cfg) + if storeErr != nil { + return nil, fmt.Errorf("auth store: %w", storeErr) + } + authSvc := auth.NewService(stores.Auth, stores.Sessions, cfg) + pwSvc := svc.ProjectWorkflow() + if pwSvc == nil { + pwSvc = projectworkflow.NewService(projectworkflow.NewStore(nil), nil) + } + + authzSvc := authz.NewService(pool) + wfDefSvc := workflowdef.NewService(pool, svc.CAS()) + instStore := projectinstance.NewStore(pool) + var instSvc *projectinstance.Service + if wfDefSvc.Available() { + docChecker := projectinstance.NewDocGovChecker(svc.DocGov()) + instSvc = projectinstance.NewService(instStore, wfDefSvc.Store(), pwSvc.Store(), docChecker) + if pool != nil { + instSvc.SetActiveMemberLookup(projectinstance.NewPGMemberLookup(pool)) + } + } + taskMgr := taskrun.NewManager(pwSvc, svc.ContextManager()) + 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) + + stages := stageflow.New(svc) + coordEngine.SetStageTaskRunner(stages) + if instSvc != nil && instSvc.Available() { + coordEngine.SetInstanceService(instSvc) + } + + rt := &Runtime{ + Cfg: cfg, + Biz: svc, + Engine: coordEngine, + Auth: authSvc, + PW: pwSvc, + TaskRuns: taskMgr, + Authorizer: authz.NewAuthorizer(pool), + Authz: authzSvc, + WFDef: wfDefSvc, + Inst: instSvc, + Notify: notification.NewService(pool), + Resolver: workspace.NewResolver(cfg.DataDir, pool), + WSCap: wsCap, + Commands: cmds, + Rec: rec, + Stageflow: stages, + } + rt.DocNotify = &docnotify.Subscriber{Notify: rt.Notify, Docs: svc.DocGov()} + rt.bindDocumentNotify() + rt.bindWorkflowApprovalHook() + return rt, nil +} + +func (r *Runtime) bindDocumentNotify() { + if r == nil || r.notifyBound { + return + } + if r.Biz != nil && r.Biz.DocRevision() != nil { + if docnotify.Bind(r.Biz.DocRevision(), r.DocNotify) { + r.subscribers++ + } + } + r.notifyBound = true +} + +func (r *Runtime) bindWorkflowApprovalHook() { + if r == nil || r.hookBound || r.Biz == nil || r.Biz.DocGov() == nil || r.Inst == nil || !r.Inst.Available() { + r.hookBound = true + return + } + r.Biz.DocGov().AddOnApprovedHook(r.onWorkflowDocumentApproved) + r.hookBound = true + r.subscribers++ +} + +func (r *Runtime) onWorkflowDocumentApproved(ctx context.Context, doc *docgov.DocumentRecord) { + if doc == nil || doc.TaskKey == "" || doc.ProjectID <= 0 || r.Inst == nil { + return + } + result, err := r.Inst.HandleDocumentApproved(ctx, doc.ProjectID, doc.TaskKey) + if err != nil || result == nil || !result.StageReady { + return + } + if r.Notify == nil || !r.Notify.Available() { + return + } + if r.PW == nil { + return + } + project, err := r.PW.GetProject(ctx, doc.ProjectID, 0) + if err != nil || project == nil || project.CreatorUserID <= 0 { + return + } + _, _ = r.Notify.EmitStageReadyToAdvance(ctx, project.CreatorUserID, notification.StageReadyPayload{ + ProjectID: doc.ProjectID, + WorkflowInstanceID: result.InstanceID, + StageInstanceID: result.StageInstanceID, + StageKey: result.StageKey, + StageName: result.StageName, + ConfigRevision: result.ConfigRevision, + RequiredDocCount: result.RequiredDocCount, + ApprovedDocCount: result.ApprovedDocCount, + }) +} + +// Start launches background workers. Repeat calls on a running runtime are no-ops. +func (r *Runtime) Start(_ context.Context) error { + if r == nil { + return fmt.Errorf("app: nil runtime") + } + r.mu.Lock() + defer r.mu.Unlock() + if r.running { + return nil + } + r.bindDocumentNotify() + r.bindWorkflowApprovalHook() + ctx, cancel := context.WithCancel(context.Background()) + r.reminderCancel = cancel + go r.runReminderLoop(ctx) + r.running = true + r.starts++ + if r.subscribers == 0 { + r.subscribers = 1 + } + return nil +} + +// Stop cancels workers. A later Start recreates the reminder loop without +// double-registering document subscribers. +func (r *Runtime) Stop(_ context.Context) error { + if r == nil { + return nil + } + r.mu.Lock() + defer r.mu.Unlock() + if !r.running { + return nil + } + if r.reminderCancel != nil { + r.reminderCancel() + r.reminderCancel = nil + } + r.running = false + return nil +} + +func (r *Runtime) SubscriberCount() int { + if r == nil { + return 0 + } + r.mu.Lock() + defer r.mu.Unlock() + return r.subscribers +} + +func (r *Runtime) StartCount() int { + if r == nil { + return 0 + } + r.mu.Lock() + defer r.mu.Unlock() + return r.starts +} + +func (r *Runtime) Running() bool { + if r == nil { + return false + } + r.mu.Lock() + defer r.mu.Unlock() + return r.running +} + +func (r *Runtime) runReminderLoop(ctx context.Context) { + r.runStageAdvanceReminders(ctx) + ticker := time.NewTicker(defaultReminderPollInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + r.runStageAdvanceReminders(ctx) + } + } +} + +func (r *Runtime) runStageAdvanceReminders(ctx context.Context) { + if r.Notify == nil || !r.Notify.Available() { + return + } + n, err := r.Notify.ProcessStageAdvanceReminders(ctx) + if err != nil { + slog.Warn("stage advance reminder scan failed", "error", err) + return + } + if n > 0 { + slog.Info("stage advance 24h reminders sent", "count", n) + } +} diff --git a/internal/app/app_test.go b/internal/app/app_test.go new file mode 100644 index 00000000..e3c281f6 --- /dev/null +++ b/internal/app/app_test.go @@ -0,0 +1,44 @@ +package app + +import ( + "context" + "testing" + "time" +) + +func TestStartStopRestartIdempotent(t *testing.T) { + rt := &Runtime{} + ctx := context.Background() + if err := rt.Start(ctx); err != nil { + t.Fatal(err) + } + if err := rt.Start(ctx); err != nil { + t.Fatal(err) + } + if rt.StartCount() != 1 { + t.Fatalf("duplicate Start spawned workers: starts=%d", rt.StartCount()) + } + if got := rt.SubscriberCount(); got != 1 { + t.Fatalf("subscribers=%d", got) + } + if !rt.Running() { + t.Fatal("expected running") + } + if err := rt.Stop(ctx); err != nil { + t.Fatal(err) + } + if rt.Running() { + t.Fatal("expected stopped") + } + if err := rt.Start(ctx); err != nil { + t.Fatal(err) + } + if rt.StartCount() != 2 { + t.Fatalf("restart did not relaunch worker: starts=%d", rt.StartCount()) + } + if rt.SubscriberCount() != 1 { + t.Fatalf("restart re-registered subscribers: %d", rt.SubscriberCount()) + } + _ = rt.Stop(ctx) + time.Sleep(10 * time.Millisecond) +} diff --git a/internal/app/chatflow/assembler.go b/internal/app/chatflow/assembler.go index e310bb7c..59254a43 100644 --- a/internal/app/chatflow/assembler.go +++ b/internal/app/chatflow/assembler.go @@ -70,16 +70,22 @@ func (a *Assembler) Assemble(ctx context.Context, p Principal, req AssembleReque if restored == nil { return userCtx, nil } - if restored.ProjectID > 0 && userCtx.Project.ID == "" { - userCtx.Project.ID = fmt.Sprintf("%d", restored.ProjectID) + upd := coordinator.ContextUpdate{ + ExpectedVersion: userCtx.Version, + BlackboardFill: restored.Facts, + SetRecentDialogue: true, + RecentDialogue: authorizedTurns(restored), } - if userCtx.BlackboardFacts == nil { - userCtx.BlackboardFacts = make(map[string]string) + if restored.ProjectID > 0 { + upd.ProjectIDIfEmpty = fmt.Sprintf("%d", restored.ProjectID) } - for k, v := range restored.Facts { - if _, exists := userCtx.BlackboardFacts[k]; !exists { - userCtx.BlackboardFacts[k] = v - } + _ = coordinator.ApplyContextUpdate(userCtx, upd) + return userCtx, nil +} + +func authorizedTurns(restored *ctxmgr.RestoredSessionContext) []coordinator.DialogueTurn { + if restored == nil { + return nil } turns := make([]coordinator.DialogueTurn, 0, len(restored.Messages)) records := make([]ctxmgr.ChatMessageRecord, 0, len(restored.Messages)) @@ -101,6 +107,5 @@ func (a *Assembler) Assemble(ctx context.Context, p Principal, req AssembleReque turns = append(turns, coordinator.DialogueTurn{Role: m.Role, Content: m.Content}) } } - userCtx.RecentDialogue = turns - return userCtx, nil + return turns } diff --git a/internal/app/docnotify/subscriber.go b/internal/app/docnotify/subscriber.go new file mode 100644 index 00000000..2025e864 --- /dev/null +++ b/internal/app/docnotify/subscriber.go @@ -0,0 +1,94 @@ +package docnotify + +import ( + "context" + "fmt" + + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/docrevision" + "chat-agent-system/internal/notification" +) + +// DocumentLookup resolves a document for recipient and project fill-in. +type DocumentLookup interface { + GetDocument(ctx context.Context, docID string) (*docgov.DocumentRecord, error) +} + +// Subscriber is the unique consumer for document OnOutbox notification events. +// Bind once from the app composition root; do not assign OnOutbox in server. +type Subscriber struct { + Notify notificationEmitter + Docs DocumentLookup +} + +type notificationEmitter interface { + Available() bool + EmitDocumentEvent(ctx context.Context, eventType, notifType, title, body, dedupeKey string, recipientUserID int64, payload notification.DocumentEventPayload) (bool, error) +} + +// Bind assigns the subscriber exactly once. Repeat calls keep the first handler. +func Bind(rev *docrevision.Service, sub *Subscriber) bool { + if rev == nil || sub == nil || rev.OnOutbox != nil { + return false + } + rev.OnOutbox = sub.Handle + return true +} + +// Handle maps one document outbox event onto a permission-safe inbox notification. +func (s *Subscriber) Handle(ctx context.Context, eventType string, payload map[string]any) error { + if s == nil || s.Notify == nil || !s.Notify.Available() { + return nil + } + docID, _ := payload["doc_id"].(string) + projectID, _ := payload["project_id"].(int64) + risk, _ := payload["risk"].(string) + var recipient int64 + if s.Docs != nil && docID != "" { + if doc, err := s.Docs.GetDocument(ctx, docID); err == nil && doc != nil && doc.AuthorID > 0 { + recipient = doc.AuthorID + projectID = doc.ProjectID + } + } + if recipient == 0 { + return nil + } + p := notification.DocumentEventPayload{ + ProjectID: projectID, DocID: docID, RiskLevel: risk, + } + switch eventType { + case notification.EventDocAgentCandidateReady: + if v, ok := payload["candidate_id"].(int64); ok { + p.CandidateID = v + } + _, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocAgentCandidateReady, + "智能体变更待确认", "文档有高风险智能体候选变更，请确认后生效。", + fmt.Sprintf("doc_candidate:%s:%d:%d", docID, p.CandidateID, recipient), recipient, p) + return err + case notification.EventDocAgentVersionCommitted: + if v, ok := payload["version_no"].(int); ok { + p.VersionNo = v + } + _, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocAgentVersionCommitted, + "智能体版本已生效", "新的智能体文档版本已提交。", + fmt.Sprintf("doc_agent_version:%s:%d:%d", docID, p.VersionNo, recipient), recipient, p) + return err + case notification.EventDocDownstreamImpact: + impacted, _ := payload["impacted_doc_id"].(string) + p.ImpactedDocID = impacted + _, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocDownstreamImpact, + "下游文档受影响", "上游文档变更，请复核关联文档。", + fmt.Sprintf("doc_impact:%s:%s:%d", docID, impacted, recipient), recipient, p) + return err + case notification.EventDocChangeReportDegraded: + if v, ok := payload["report_id"].(int64); ok { + p.ReportID = v + } + _, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocChangeReportDegraded, + "变更报告未完整评估", "变更报告降级，受控文档暂不可送审。", + fmt.Sprintf("doc_report_degraded:%s:%d:%d", docID, p.ReportID, recipient), recipient, p) + return err + default: + return nil + } +} diff --git a/internal/app/docnotify/subscriber_test.go b/internal/app/docnotify/subscriber_test.go new file mode 100644 index 00000000..16a6440f --- /dev/null +++ b/internal/app/docnotify/subscriber_test.go @@ -0,0 +1,73 @@ +package docnotify + +import ( + "context" + "testing" + + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/docrevision" + "chat-agent-system/internal/notification" +) + +type stubNotify struct { + available bool + calls int + lastType string +} + +func (s *stubNotify) Available() bool { return s != nil && s.available } + +func (s *stubNotify) EmitDocumentEvent(_ context.Context, eventType, _, _, _, _ string, _ int64, _ notification.DocumentEventPayload) (bool, error) { + s.calls++ + s.lastType = eventType + return true, nil +} + +type stubDocs struct { + doc *docgov.DocumentRecord +} + +func (s stubDocs) GetDocument(_ context.Context, _ string) (*docgov.DocumentRecord, error) { + return s.doc, nil +} + +func TestBindOnce(t *testing.T) { + rev := &docrevision.Service{} + sub := &Subscriber{Notify: &stubNotify{available: true}} + if !Bind(rev, sub) { + t.Fatal("first bind") + } + if Bind(rev, sub) { + t.Fatal("second bind must be a no-op") + } + if rev.OnOutbox == nil { + t.Fatal("hook missing") + } +} + +func TestHandleCandidateReady(t *testing.T) { + n := &stubNotify{available: true} + sub := &Subscriber{ + Notify: n, + Docs: stubDocs{doc: &docgov.DocumentRecord{AuthorID: 9, ProjectID: 4}}, + } + if err := sub.Handle(context.Background(), notification.EventDocAgentCandidateReady, map[string]any{ + "doc_id": "d1", "candidate_id": int64(3), + }); err != nil { + t.Fatal(err) + } + if n.calls != 1 || n.lastType != notification.EventDocAgentCandidateReady { + t.Fatalf("calls=%d type=%s", n.calls, n.lastType) + } +} + +func TestHandleSkipsUnknownAndNoRecipient(t *testing.T) { + n := &stubNotify{available: true} + sub := &Subscriber{Notify: n} + if err := sub.Handle(context.Background(), "document.unknown", map[string]any{"doc_id": "d1"}); err != nil { + t.Fatal(err) + } + if n.calls != 0 { + t.Fatalf("unknown event must not emit, calls=%d", n.calls) + } +} diff --git a/internal/app/stageflow/usecase.go b/internal/app/stageflow/usecase.go new file mode 100644 index 00000000..e0a34a82 --- /dev/null +++ b/internal/app/stageflow/usecase.go @@ -0,0 +1,40 @@ +package stageflow + +import ( + "context" + "errors" + + "chat-agent-system/internal/bizdev" +) + +var ErrClosed = errors.New("stageflow: executor not configured") + +// Executor is the unique application owner for task and stage runs. +// HTTP, CLI and coordinator must call this instead of bizdev directly. +type Executor interface { + ExecuteTask(ctx context.Context, req bizdev.TaskRunRequest) (*bizdev.TaskRunResult, error) + ExecuteStage(ctx context.Context, req bizdev.StageRunRequest) (*bizdev.StageRunResult, error) +} + +// UseCase adapts bizdev orchestration without moving domain tables. +type UseCase struct { + Biz Executor +} + +func New(biz Executor) *UseCase { + return &UseCase{Biz: biz} +} + +func (u *UseCase) ExecuteTask(ctx context.Context, req bizdev.TaskRunRequest) (*bizdev.TaskRunResult, error) { + if u == nil || u.Biz == nil { + return nil, ErrClosed + } + return u.Biz.ExecuteTask(ctx, req) +} + +func (u *UseCase) ExecuteStage(ctx context.Context, req bizdev.StageRunRequest) (*bizdev.StageRunResult, error) { + if u == nil || u.Biz == nil { + return nil, ErrClosed + } + return u.Biz.ExecuteStage(ctx, req) +} diff --git a/internal/app/stageflow/usecase_test.go b/internal/app/stageflow/usecase_test.go new file mode 100644 index 00000000..efe370d1 --- /dev/null +++ b/internal/app/stageflow/usecase_test.go @@ -0,0 +1,46 @@ +package stageflow + +import ( + "context" + "errors" + "testing" + + "chat-agent-system/internal/bizdev" +) + +type stubBiz struct { + tasks int + stages int +} + +func (s *stubBiz) ExecuteTask(_ context.Context, req bizdev.TaskRunRequest) (*bizdev.TaskRunResult, error) { + s.tasks++ + return &bizdev.TaskRunResult{Success: true, ArtifactTitle: req.TaskKey}, nil +} + +func (s *stubBiz) ExecuteStage(_ context.Context, req bizdev.StageRunRequest) (*bizdev.StageRunResult, error) { + s.stages++ + return &bizdev.StageRunResult{Success: true, StageTitle: string(req.Stage)}, nil +} + +func TestUseCaseDelegates(t *testing.T) { + biz := &stubBiz{} + httpUC := New(biz) + cliUC := New(biz) + if _, err := httpUC.ExecuteTask(context.Background(), bizdev.TaskRunRequest{TaskKey: "intake"}); err != nil { + t.Fatal(err) + } + if _, err := cliUC.ExecuteStage(context.Background(), bizdev.StageRunRequest{Stage: bizdev.StageV01_Triage}); err != nil { + t.Fatal(err) + } + if biz.tasks != 1 || biz.stages != 1 { + t.Fatalf("shared executor counts task=%d stage=%d", biz.tasks, biz.stages) + } +} + +func TestUseCaseClosed(t *testing.T) { + var u *UseCase + if _, err := u.ExecuteTask(context.Background(), bizdev.TaskRunRequest{}); !errors.Is(err, ErrClosed) { + t.Fatalf("got %v", err) + } +} diff --git a/internal/archtest/arch_test.go b/internal/archtest/arch_test.go index eb414381..8bce1e52 100644 --- a/internal/archtest/arch_test.go +++ b/internal/archtest/arch_test.go @@ -26,7 +26,25 @@ func TestArchitectureBoundaries(t *testing.T) { if ev.Failed() { t.Fatal(ev.Report()) } - if os.Getenv("ARCHTEST_STRICT") != "" && len(ev.Excepted) > 0 { - t.Fatalf("strict/block mode: %d baseline exceptions remain\n%s", len(ev.Excepted), ev.Report()) + if os.Getenv("ARCHTEST_STRICT") == "" { + return + } + if len(b.BlockKinds) == 0 { + if len(ev.Excepted) > 0 { + t.Fatalf("strict/block mode: %d baseline exceptions remain\n%s", len(ev.Excepted), ev.Report()) + } + return + } + for _, f := range ev.Excepted { + for _, kind := range b.BlockKinds { + if f.Kind == kind { + t.Fatalf("strict/block mode: cleared kind %s still has exception %s\n%s", kind, f.Key(), ev.Report()) + } + } + } + for _, kind := range b.BlockKinds { + if ev.Counts[kind] > 0 { + t.Fatalf("strict/block mode: cleared kind %s count=%d\n%s", kind, ev.Counts[kind], ev.Report()) + } } } diff --git a/internal/archtest/boundaries.go b/internal/archtest/boundaries.go index 8f575af3..3b8e2ecd 100644 --- a/internal/archtest/boundaries.go +++ b/internal/archtest/boundaries.go @@ -27,6 +27,7 @@ type Boundaries struct { UnwrapOwners []string `yaml:"unwrap_owners"` Exceptions []Exception `yaml:"exceptions"` BaselineCounts map[string]int `yaml:"baseline_counts"` + BlockKinds []string `yaml:"block_kinds"` } type Exception struct { diff --git a/internal/chat/session.go b/internal/chat/session.go index 85b99276..fae3289e 100644 --- a/internal/chat/session.go +++ b/internal/chat/session.go @@ -81,7 +81,11 @@ func NewSession(id string, userCtx *coordinator.UserProjectContext, rec intent.R } // 若上下文尚未挂起预期且当前已具有阶段，则初始化默认选项草稿 if userCtx != nil && userCtx.PendingExpectation == nil && userCtx.Project.CurrentStage != "" { - userCtx.PendingExpectation = coordinator.BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name) + _ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{ + ExpectedVersion: userCtx.Version, + SetPendingExpectation: true, + PendingExpectation: coordinator.BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name), + }) } sess := &Session{ @@ -115,7 +119,11 @@ func (s *Session) SyncRecentDialogue() { } turns = append(turns, coordinator.DialogueTurn{Role: m.Role, Content: m.Content}) } - s.UserCtx.RecentDialogue = turns + _ = coordinator.ApplyContextUpdate(s.UserCtx, coordinator.ContextUpdate{ + ExpectedVersion: s.UserCtx.Version, + SetRecentDialogue: true, + RecentDialogue: turns, + }) // 若发生了截断且存在早期消息，同步计算或加载滚动摘要 if truncated && len(early) > 0 && s.CtxMgr != nil { @@ -416,11 +424,17 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str } if coordRes.ClearExpectation { - s.UserCtx.PendingExpectation = nil + _ = coordinator.ApplyContextUpdate(s.UserCtx, coordinator.ContextUpdate{ + ExpectedVersion: s.UserCtx.Version, SetPendingExpectation: true, + }) } else if coordRes.NextExpectation != nil { - s.UserCtx.PendingExpectation = coordRes.NextExpectation + _ = coordinator.ApplyContextUpdate(s.UserCtx, coordinator.ContextUpdate{ + ExpectedVersion: s.UserCtx.Version, SetPendingExpectation: true, PendingExpectation: coordRes.NextExpectation, + }) } else if recRes.Branch == intent.BranchExpectationHit { - s.UserCtx.PendingExpectation = nil + _ = coordinator.ApplyContextUpdate(s.UserCtx, coordinator.ContextUpdate{ + ExpectedVersion: s.UserCtx.Version, SetPendingExpectation: true, + }) } // 若为 BranchNewIntentPreempt 且未产生新预期，原挂起预期在焦点底静默保留，不阻拦用户 diff --git a/internal/coordinator/context.go b/internal/coordinator/context.go index 5b79e019..7472cee6 100644 --- a/internal/coordinator/context.go +++ b/internal/coordinator/context.go @@ -8,6 +8,7 @@ import ( "path/filepath" "strconv" "strings" + "sync" "time" "chat-agent-system/internal/bizdev" @@ -76,6 +77,8 @@ type DocumentSummaryReader func(ctx context.Context, target string) (*ReadableDo // UserProjectContext 用户项目级全量统一上下文 (对话系统与执行系统共同消费) type UserProjectContext struct { + mu sync.Mutex + Version int64 `json:"version"` SnapshotID string `json:"snapshot_id"` AssembledAt time.Time `json:"assembled_at"` User UserInfo `json:"user"` diff --git a/internal/coordinator/context_update.go b/internal/coordinator/context_update.go new file mode 100644 index 00000000..92ff07e8 --- /dev/null +++ b/internal/coordinator/context_update.go @@ -0,0 +1,144 @@ +package coordinator + +import ( + "errors" + + "chat-agent-system/internal/intent" +) + +var ( + ErrNilContext = errors.New("coordinator: nil user project context") + ErrStaleContext = errors.New("coordinator: stale context version") + ErrInvalidUpdate = errors.New("coordinator: empty context update") +) + +// ContextUpdate is the only typed write against UserProjectContext. +// Callers must supply ExpectedVersion from a prior read; a mismatch fails closed. +type ContextUpdate struct { + ExpectedVersion int64 + + BlackboardFill map[string]string + BlackboardSet map[string]string + + SetPendingExpectation bool + PendingExpectation *intent.PendingExpectation + ExpirePending bool + + SetInteractiveCard bool + InteractiveCard *intent.InteractiveCardPayload + TickInteractive bool + DismissCardStatus string // cancelled | resolved | expired; then nil + + SetPersonalizedGuidance bool + PersonalizedGuidance string + + SetRecentDialogue bool + RecentDialogue []DialogueTurn + + ProjectIDIfEmpty string +} + +// ApplyContextUpdate applies one versioned patch. Concurrent writers with the +// same ExpectedVersion: exactly one succeeds, the rest get ErrStaleContext. +func ApplyContextUpdate(ctx *UserProjectContext, u ContextUpdate) error { + if ctx == nil { + return ErrNilContext + } + ctx.mu.Lock() + defer ctx.mu.Unlock() + if u.ExpectedVersion != ctx.Version { + return ErrStaleContext + } + applied := false + if len(u.BlackboardFill) > 0 { + if ctx.BlackboardFacts == nil { + ctx.BlackboardFacts = make(map[string]string, len(u.BlackboardFill)) + } + for k, v := range u.BlackboardFill { + if _, exists := ctx.BlackboardFacts[k]; !exists { + ctx.BlackboardFacts[k] = v + } + } + applied = true + } + if len(u.BlackboardSet) > 0 { + if ctx.BlackboardFacts == nil { + ctx.BlackboardFacts = make(map[string]string, len(u.BlackboardSet)) + } + for k, v := range u.BlackboardSet { + ctx.BlackboardFacts[k] = v + } + applied = true + } + if u.ExpirePending && ctx.PendingExpectation != nil && !ctx.PendingExpectation.ExpiresAt.IsZero() { + // caller decides freshness; apply just clears + ctx.PendingExpectation = nil + applied = true + } + if u.SetPendingExpectation { + ctx.PendingExpectation = u.PendingExpectation + applied = true + } + if u.TickInteractive && ctx.InteractiveCard != nil { + ctx.InteractiveCard.TTLTurns-- + if ctx.InteractiveCard.TTLTurns <= 0 || ctx.InteractiveCard.CurrentStatus != "active" { + ctx.InteractiveCard.CurrentStatus = "expired" + ctx.InteractiveCard = nil + } + applied = true + } + if u.DismissCardStatus != "" && ctx.InteractiveCard != nil { + ctx.InteractiveCard.CurrentStatus = u.DismissCardStatus + ctx.InteractiveCard = nil + applied = true + } + if u.SetInteractiveCard { + ctx.InteractiveCard = u.InteractiveCard + applied = true + } + if u.SetPersonalizedGuidance { + ctx.PersonalizedGuidance = u.PersonalizedGuidance + applied = true + } + if u.SetRecentDialogue { + ctx.RecentDialogue = u.RecentDialogue + applied = true + } + if u.ProjectIDIfEmpty != "" && ctx.Project.ID == "" { + ctx.Project.ID = u.ProjectIDIfEmpty + applied = true + } + if !applied { + return ErrInvalidUpdate + } + ctx.Version++ + return nil +} + +func (c *UserProjectContext) applyPending(exp *intent.PendingExpectation) { + if c == nil { + return + } + _ = ApplyContextUpdate(c, ContextUpdate{ExpectedVersion: c.Version, SetPendingExpectation: true, PendingExpectation: exp}) +} + +func (c *UserProjectContext) applyCard(card *intent.InteractiveCardPayload) { + if c == nil { + return + } + _ = ApplyContextUpdate(c, ContextUpdate{ExpectedVersion: c.Version, SetInteractiveCard: true, InteractiveCard: card}) +} + +func (c *UserProjectContext) dismissCard(status string) { + if c == nil { + return + } + _ = ApplyContextUpdate(c, ContextUpdate{ExpectedVersion: c.Version, DismissCardStatus: status}) +} + +func (c *UserProjectContext) applyBlackboard(key, value string) { + if c == nil { + return + } + _ = ApplyContextUpdate(c, ContextUpdate{ExpectedVersion: c.Version, BlackboardSet: map[string]string{key: value}}) +} diff --git a/internal/coordinator/context_update_test.go b/internal/coordinator/context_update_test.go new file mode 100644 index 00000000..cb7aa537 --- /dev/null +++ b/internal/coordinator/context_update_test.go @@ -0,0 +1,55 @@ +package coordinator + +import ( + "errors" + "sync" + "sync/atomic" + "testing" +) + +func TestContextUpdateConcurrentNoLostUpdate(t *testing.T) { + ctx := &UserProjectContext{ + Version: 1, + BlackboardFacts: map[string]string{"market": "国内"}, + } + var ok, stale int32 + var wg sync.WaitGroup + for i := 0; i < 2; i++ { + wg.Add(1) + go func(label string) { + defer wg.Done() + err := ApplyContextUpdate(ctx, ContextUpdate{ + ExpectedVersion: 1, + BlackboardSet: map[string]string{"market": label}, + }) + switch { + case err == nil: + atomic.AddInt32(&ok, 1) + case errors.Is(err, ErrStaleContext): + atomic.AddInt32(&stale, 1) + default: + t.Errorf("unexpected %v", err) + } + }("v" + string(rune('a'+i))) + } + wg.Wait() + if ok != 1 || stale != 1 { + t.Fatalf("ok=%d stale=%d version=%d facts=%v", ok, stale, ctx.Version, ctx.BlackboardFacts) + } + if ctx.Version != 2 { + t.Fatalf("version %d", ctx.Version) + } + if ctx.BlackboardFacts["market"] == "国内" { + t.Fatal("winner did not apply") + } +} + +func TestContextUpdateFillDoesNotOverwrite(t *testing.T) { + ctx := &UserProjectContext{Version: 0, BlackboardFacts: map[string]string{"a": "1"}} + if err := ApplyContextUpdate(ctx, ContextUpdate{ExpectedVersion: 0, BlackboardFill: map[string]string{"a": "x", "b": "2"}}); err != nil { + t.Fatal(err) + } + if ctx.BlackboardFacts["a"] != "1" || ctx.BlackboardFacts["b"] != "2" { + t.Fatalf("%v", ctx.BlackboardFacts) + } +} diff --git a/internal/coordinator/engine.go b/internal/coordinator/engine.go index 97301339..82a676be 100644 --- a/internal/coordinator/engine.go +++ b/internal/coordinator/engine.go @@ -73,12 +73,36 @@ type CoordinationResult struct { // CoordinatorEngine L2 协调引擎 type CoordinatorEngine struct { bizdevSvc bizdev.BizdevService + stageTasks StageTaskRunner instSvc *projectinstance.Service assembler *ContextAssembler dialogueBackend agent.Backend dataDir string } +// StageTaskRunner is the application owner for ExecuteTask / ExecuteStage. +// Wired from app/stageflow; tests may leave it nil and fall back to bizdevSvc. +type StageTaskRunner interface { + ExecuteTask(ctx context.Context, req bizdev.TaskRunRequest) (*bizdev.TaskRunResult, error) + ExecuteStage(ctx context.Context, req bizdev.StageRunRequest) (*bizdev.StageRunResult, error) +} + +func (e *CoordinatorEngine) SetStageTaskRunner(r StageTaskRunner) { + if e != nil { + e.stageTasks = r + } +} + +func (e *CoordinatorEngine) tasks() StageTaskRunner { + if e == nil { + return nil + } + if e.stageTasks != nil { + return e.stageTasks + } + return e.bizdevSvc +} + // SetInstanceService 注入项目工作流实例服务（对话推进与实例门禁合流）。 func (e *CoordinatorEngine) SetInstanceService(svc *projectinstance.Service) { if e != nil { @@ -118,14 +142,10 @@ func (e *CoordinatorEngine) ProcessIntent( // 交互卡片 TTL 轮次消退与生命周期清理 if userCtx.InteractiveCard != nil { - userCtx.InteractiveCard.TTLTurns-- - if userCtx.InteractiveCard.TTLTurns <= 0 || userCtx.InteractiveCard.CurrentStatus != "active" { - userCtx.InteractiveCard.CurrentStatus = "expired" - userCtx.InteractiveCard = nil - } + _ = ApplyContextUpdate(userCtx, ContextUpdate{ExpectedVersion: userCtx.Version, TickInteractive: true}) } if userCtx.PendingExpectation != nil && !userCtx.PendingExpectation.ExpiresAt.IsZero() && time.Now().After(userCtx.PendingExpectation.ExpiresAt) { - userCtx.PendingExpectation = nil + userCtx.applyPending(nil) } top := recRes.TopIntent @@ -355,10 +375,9 @@ func (e *CoordinatorEngine) handleDeclineExpectation(userCtx *UserProjectContext if projName == "" { projName = "当前项目" } - userCtx.PendingExpectation = nil + userCtx.applyPending(nil) if userCtx.InteractiveCard != nil { - userCtx.InteractiveCard.CurrentStatus = "cancelled" - userCtx.InteractiveCard = nil + userCtx.dismissCard("cancelled") } reply := fmt.Sprintf("收到，听您的！那我们先不急着推进。\n针对「%s」，您手头还有哪些细节、技术顾虑或商务考量想先聊聊？我随时待命为您解答。", projName) @@ -418,7 +437,7 @@ func (e *CoordinatorEngine) handleAmbiguousConsult(ctx context.Context, cand *in nextExp := BuildNextExpectation(currentStage, projName) // 路由层：复杂自由任务画像命中时，先给即时初答，再附升级建议（复用挂起期望机制，含每会话冷却） reply, nextExp = maybeAttachUpgradeHint(rawInput, userCtx, reply, nextExp) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) return &CoordinationResult{ Status: StatusAccepted, @@ -436,7 +455,7 @@ func (e *CoordinatorEngine) handleQueryProgress(ctx context.Context, cand *inten brief := userCtx.LatestAdvisorBrief topic := cand.FilledSlots[intent.SlotQueryTopic] nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) // 1. 若为测试方案/难点/验证咨询，直接调取测试经理与参谋专业知识库解答 if topic == "testing_analysis" { @@ -624,7 +643,7 @@ func (e *CoordinatorEngine) handleSupplementInfo(cand *intent.CandidateIntent, u summaryKey := "" if factKey != "" && factVal != "" { - userCtx.BlackboardFacts[factKey] = factVal + userCtx.applyBlackboard(factKey, factVal) summaryKey = fmt.Sprintf("「%s」已更新为「%s」", factKey, factVal) if factKey == "target_market" { userCtx.Project.TargetMarket = factVal @@ -657,7 +676,7 @@ func (e *CoordinatorEngine) handleSupplementInfo(cand *intent.CandidateIntent, u } nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) var sb strings.Builder sb.WriteString(fmt.Sprintf("已收到并为您记录最新信息：%s。\n", summaryKey)) @@ -757,7 +776,7 @@ func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent if len(cand.PatchSlots) > 0 { var patches []string for k, v := range cand.PatchSlots { - userCtx.BlackboardFacts[k] = v + userCtx.applyBlackboard(k, v) patches = append(patches, fmt.Sprintf("%s=%s", k, v)) if k == "target_market" { userCtx.Project.TargetMarket = v @@ -795,11 +814,11 @@ func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent }, nil } - if e.bizdevSvc == nil { + if e.tasks() == nil { userCtx.Project.CurrentStage = targetStage userCtx.AvailableActions = ComputeAvailableActions(targetStage, userCtx.User.ProjectRole) nextExp := BuildNextExpectation(targetStage, userCtx.Project.Name) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) reply := fmt.Sprintf("已为您模拟推进到阶段【%s】。", targetStage) if patchSummary != "" { @@ -816,7 +835,7 @@ func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent } // 3. 调用底层 L3 调度器执行阶段跃迁 - stageRes, err := e.bizdevSvc.ExecuteStage(ctx, bizdev.StageRunRequest{ + stageRes, err := e.tasks().ExecuteStage(ctx, bizdev.StageRunRequest{ Stage: targetStage, ProjectName: userCtx.Project.Name, CustomerName: userCtx.Project.CustomerName, @@ -850,7 +869,7 @@ func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent } nextExp := BuildNextExpectation(targetStage, userCtx.Project.Name) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) var sb strings.Builder if patchSummary != "" { @@ -889,9 +908,9 @@ func (e *CoordinatorEngine) handleTriggerTask(ctx context.Context, cand *intent. } nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) - if e.bizdevSvc == nil { + if e.tasks() == nil { return &CoordinationResult{ Status: StatusAccepted, Intent: cand.Intent, @@ -910,7 +929,7 @@ func (e *CoordinatorEngine) handleTriggerTask(ctx context.Context, cand *intent. fmt.Sscanf(userCtx.User.ID, "%d", &userID) } - taskRes, err := e.bizdevSvc.ExecuteTask(ctx, bizdev.TaskRunRequest{ + taskRes, err := e.tasks().ExecuteTask(ctx, bizdev.TaskRunRequest{ TaskKey: taskKey, ProjectName: userCtx.Project.Name, CustomerName: userCtx.Project.CustomerName, @@ -1026,9 +1045,9 @@ func (e *CoordinatorEngine) handleTriggerAdhocTask(ctx context.Context, cand *in taskKey := sanitizeAdhocTaskKey(title) nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) - if e.bizdevSvc == nil { + if e.tasks() == nil { return &CoordinationResult{ Status: StatusAccepted, Intent: cand.Intent, @@ -1058,7 +1077,7 @@ func (e *CoordinatorEngine) handleTriggerAdhocTask(ctx context.Context, cand *in Content: []byte(fmt.Sprintf("# 任务简报（用户原始诉求，权威等级：primary_evidence）\n\n%s\n", brief)), } - taskRes, err := e.bizdevSvc.ExecuteTask(ctx, bizdev.TaskRunRequest{ + taskRes, err := e.tasks().ExecuteTask(ctx, bizdev.TaskRunRequest{ TaskKey: taskKey, Title: title, ProjectName: userCtx.Project.Name, @@ -1141,7 +1160,7 @@ func (e *CoordinatorEngine) handleCasualChat(ctx context.Context, userCtx *UserP } nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) nameSuffix := "" if userCtx.User.Name != "" { @@ -1157,7 +1176,7 @@ func (e *CoordinatorEngine) handleCasualChat(ctx context.Context, userCtx *UserP // 路由层：复杂自由任务画像命中时，先给即时初答，再附升级建议（复用挂起期望机制，含每会话冷却） reply, nextExp = maybeAttachUpgradeHint(rawInput, userCtx, reply, nextExp) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) return &CoordinationResult{ Status: StatusAccepted, @@ -1254,8 +1273,8 @@ func (e *CoordinatorEngine) handleDocGovReject(ctx context.Context, cand *intent }, nil } - userCtx.InteractiveCard = resolved.Card - userCtx.PendingExpectation = resolved.Expectation + userCtx.applyCard(resolved.Card) + userCtx.applyPending(resolved.Expectation) return &CoordinationResult{ Status: StatusAccepted, @@ -1299,7 +1318,7 @@ func (e *CoordinatorEngine) handleQueryRejectReason(ctx context.Context, cand *i resolved := ResolveRejectCard(role, "", docName, docName, "", reason, RejectUtteranceQuery) if resolved.Card != nil { - userCtx.InteractiveCard = resolved.Card + userCtx.applyCard(resolved.Card) } return &CoordinationResult{ @@ -1442,8 +1461,8 @@ func (e *CoordinatorEngine) handleDocGovApprove(ctx context.Context, cand *inten }, } - userCtx.InteractiveCard = cardPayload - userCtx.PendingExpectation = nextExp + userCtx.applyCard(cardPayload) + userCtx.applyPending(nextExp) reply := fmt.Sprintf("已为您起草针对《%s》的终审批准确认单。终审批准将使文档正式归档并永久锁定，请审阅下方卡片并在确认无误后点击确认或回复‘确认’。", docName) @@ -1559,10 +1578,9 @@ func (e *CoordinatorEngine) handleDocGovProposalDecide(ctx context.Context, cand } if userCtx.InteractiveCard != nil { - userCtx.InteractiveCard.CurrentStatus = "resolved" - userCtx.InteractiveCard = nil + userCtx.dismissCard("resolved") } - userCtx.PendingExpectation = nil + userCtx.applyPending(nil) rollback := docgov.StateDrafting if userCtx.InteractiveCard != nil && strings.Contains(userCtx.InteractiveCard.ImpactRisk, "reviewing") { @@ -1587,10 +1605,9 @@ func (e *CoordinatorEngine) handleDocGovProposalDecide(ctx context.Context, cand if targetAction == "confirm_final_approve" { if userCtx.InteractiveCard != nil { - userCtx.InteractiveCard.CurrentStatus = "resolved" - userCtx.InteractiveCard = nil + userCtx.dismissCard("resolved") } - userCtx.PendingExpectation = nil + userCtx.applyPending(nil) return &CoordinationResult{ Status: StatusAccepted, @@ -1608,10 +1625,9 @@ func (e *CoordinatorEngine) handleDocGovProposalDecide(ctx context.Context, cand if targetAction == "cancel" { if userCtx.InteractiveCard != nil { - userCtx.InteractiveCard.CurrentStatus = "cancelled" - userCtx.InteractiveCard = nil + userCtx.dismissCard("cancelled") } - userCtx.PendingExpectation = nil + userCtx.applyPending(nil) return &CoordinationResult{ Status: StatusAccepted, diff --git a/internal/coordinator/engine_run.go b/internal/coordinator/engine_run.go index ddfb2cca..b52883af 100644 --- a/internal/coordinator/engine_run.go +++ b/internal/coordinator/engine_run.go @@ -50,7 +50,7 @@ func (e *CoordinatorEngine) handleRunResume(_ context.Context, cand *intent.Cand } if isRunActionConfirmed(userCtx, "run.resume") { - userCtx.PendingExpectation = nil + userCtx.applyPending(nil) return &CoordinationResult{ Status: StatusAccepted, Intent: intent.IntentRunResume, @@ -96,7 +96,7 @@ func (e *CoordinatorEngine) handleRunResume(_ context.Context, cand *intent.Cand func (e *CoordinatorEngine) handleRunRestart(_ context.Context, cand *intent.CandidateIntent, userCtx *UserProjectContext) (*CoordinationResult, error) { if isRunActionConfirmed(userCtx, "run.restart") { - userCtx.PendingExpectation = nil + userCtx.applyPending(nil) taskKey := cand.FilledSlots["task_key"] if taskKey == "" && userCtx.TaskState != nil { taskKey = userCtx.TaskState.ActiveTask diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 4de4ff31..ef183972 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -253,7 +253,7 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { // 1. 如果指定了 stage，则执行成熟度阶段编排 if req.Stage != "" { - stageRes, err := s.svc.ExecuteStage(r.Context(), bizdev.StageRunRequest{ + stageRes, err := s.taskUseCase().ExecuteStage(r.Context(), bizdev.StageRunRequest{ Stage: bizdev.MaturityStage(req.Stage), ProjectName: req.ProjectName, CustomerName: req.CustomerName, @@ -301,7 +301,7 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { if req.TaskKey == "" { req.TaskKey = "intake" } - taskRes, err := s.svc.ExecuteTask(r.Context(), bizdev.TaskRunRequest{ + taskRes, err := s.taskUseCase().ExecuteTask(r.Context(), bizdev.TaskRunRequest{ TaskKey: req.TaskKey, ProjectName: req.ProjectName, CustomerName: req.CustomerName, diff --git a/internal/server/server.go b/internal/server/server.go index 666199d0..ca1ee34c 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -10,10 +10,9 @@ import ( "sync" "time" - "chat-agent-system/internal/agent" + "chat-agent-system/internal/app" "chat-agent-system/internal/app/chatflow" - "chat-agent-system/internal/app/uow" - "chat-agent-system/internal/artifactvault" + "chat-agent-system/internal/app/stageflow" "chat-agent-system/internal/auth" "chat-agent-system/internal/authz" "chat-agent-system/internal/bizdev" @@ -23,7 +22,6 @@ import ( "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/intent" "chat-agent-system/internal/notification" - "chat-agent-system/internal/observability" "chat-agent-system/internal/projectinstance" "chat-agent-system/internal/projectworkflow" "chat-agent-system/internal/taskrun" @@ -33,6 +31,7 @@ import ( // Server 提供基于 HTTP RESTful 与 SSE 的商务与研发参谋接口服务 type Server struct { + rt *app.Runtime cfg *config.Config svc *bizdev.Service pwSvc *projectworkflow.Service @@ -52,156 +51,55 @@ type Server struct { cmds *taskrun.Commands chatLog chatflow.ChatLog assembler chatflow.ContextAssembler + stageflow *stageflow.UseCase httpServer *http.Server } -// NewServer 构造函数 -func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { - if cfg == nil { - cfg, _ = config.Load("") +// NewFromRuntime attaches the HTTP adapter to an already-wired composition root. +func NewFromRuntime(rt *app.Runtime) *Server { + if rt == nil { + panic("server: nil runtime") } - if cfg != nil { - observability.InitLogger(cfg.Env) - } - if svc == nil { - var err error - svc, err = bizdev.NewService(cfg) - if err != nil { - panic("server: bizdev service init failed: " + err.Error()) - } - } - - rec := intent.NewRecognizer() - dialogueBackend := agent.NewBackend(cfg.AgentBackend, cfg) - coordEngine := coordinator.NewCoordinatorEngineWithDialogue(svc, dialogueBackend, cfg.DataDir) - pool := svc.Pool() - stores, storeErr := auth.NewStores(context.Background(), pool, cfg) - if storeErr != nil { - panic("server: auth store init failed: " + storeErr.Error()) - } - authSvc := auth.NewService(stores.Auth, stores.Sessions, cfg) - pwSvc := svc.ProjectWorkflow() - if pwSvc == nil { - pwSvc = projectworkflow.NewService(projectworkflow.NewStore(nil), nil) - } - - authzSvc := authz.NewService(pool) - wfDefSvc := workflowdef.NewService(pool, svc.CAS()) - instStore := projectinstance.NewStore(pool) - var instSvc *projectinstance.Service - if wfDefSvc.Available() { - docChecker := projectinstance.NewDocGovChecker(svc.DocGov()) - instSvc = projectinstance.NewService(instStore, wfDefSvc.Store(), pwSvc.Store(), docChecker) - if pool != nil { - instSvc.SetActiveMemberLookup(projectinstance.NewPGMemberLookup(pool)) - } - } - taskMgr := taskrun.NewManager(pwSvc, svc.ContextManager()) - 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, - svc: svc, - pwSvc: pwSvc, - rec: rec, - engine: coordEngine, - authSvc: authSvc, + 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: taskMgr, - authorizer: authz.NewAuthorizer(pool), - authzSvc: authzSvc, - wfDefSvc: wfDefSvc, - instSvc: instSvc, - notifySvc: notification.NewService(pool), - resolver: workspace.NewResolver(cfg.DataDir, pool), - wsCap: wsCap, - cmds: cmds, + 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, } - s.chatLog = chatflow.NewAdapter(svc.ContextManager(), s.resolveStageAccess, s.failClosedAuthz()) + s.chatLog = chatflow.NewAdapter(s.svc.ContextManager(), s.resolveStageAccess, s.failClosedAuthz()) s.assembler = chatflow.NewAssembler(coordinator.NewContextAssembler(), s.chatLog) - if instSvc != nil && instSvc.Available() { - svc.SetStageAccessGate(newBizdevStageGate(instSvc, pwSvc)) - coordEngine.SetInstanceService(instSvc) + rt.ChatLog = s.chatLog + rt.Assembler = s.assembler + if s.instSvc != nil && s.instSvc.Available() && s.svc != nil { + s.svc.SetStageAccessGate(newBizdevStageGate(s.instSvc, s.pwSvc)) } - s.registerWorkflowApprovalHook() - s.startNotificationReminderLoop() s.initTaskRunExecutors() - if svc != nil && svc.DocRevision() != nil && s.notifySvc != nil { - rev := svc.DocRevision() - rev.OnOutbox = func(ctx context.Context, eventType string, payload map[string]any) error { - if !s.notifySvc.Available() { - return nil - } - docID, _ := payload["doc_id"].(string) - projectID, _ := payload["project_id"].(int64) - risk, _ := payload["risk"].(string) - var recipient int64 - if s.svc.DocGov() != nil { - if doc, err := s.svc.DocGov().GetDocument(ctx, docID); err == nil && doc != nil && doc.AuthorID > 0 { - recipient = doc.AuthorID - projectID = doc.ProjectID - } - } - if recipient == 0 { - return nil - } - p := notification.DocumentEventPayload{ - ProjectID: projectID, DocID: docID, RiskLevel: risk, - } - switch eventType { - case notification.EventDocAgentCandidateReady: - if v, ok := payload["candidate_id"].(int64); ok { - p.CandidateID = v - } - _, err := s.notifySvc.EmitDocumentEvent(ctx, eventType, notification.TypeDocAgentCandidateReady, - "智能体变更待确认", "文档有高风险智能体候选变更，请确认后生效。", - fmt.Sprintf("doc_candidate:%s:%d:%d", docID, p.CandidateID, recipient), recipient, p) - return err - case notification.EventDocAgentVersionCommitted: - if v, ok := payload["version_no"].(int); ok { - p.VersionNo = v - } - _, err := s.notifySvc.EmitDocumentEvent(ctx, eventType, notification.TypeDocAgentVersionCommitted, - "智能体版本已生效", "新的智能体文档版本已提交。", - fmt.Sprintf("doc_agent_version:%s:%d:%d", docID, p.VersionNo, recipient), recipient, p) - return err - case notification.EventDocDownstreamImpact: - impacted, _ := payload["impacted_doc_id"].(string) - p.ImpactedDocID = impacted - _, err := s.notifySvc.EmitDocumentEvent(ctx, eventType, notification.TypeDocDownstreamImpact, - "下游文档受影响", "上游文档变更，请复核关联文档。", - fmt.Sprintf("doc_impact:%s:%s:%d", docID, impacted, recipient), recipient, p) - return err - case notification.EventDocChangeReportDegraded: - if v, ok := payload["report_id"].(int64); ok { - p.ReportID = v - } - _, err := s.notifySvc.EmitDocumentEvent(ctx, eventType, notification.TypeDocChangeReportDegraded, - "变更报告未完整评估", "变更报告降级，受控文档暂不可送审。", - fmt.Sprintf("doc_report_degraded:%s:%d:%d", docID, p.ReportID, recipient), recipient, p) - return err - default: - return nil - } - } - } return s } +// NewServer 构造函数（测试兼容：内部走 app.Assemble） +func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { + rt, err := app.Assemble(cfg, svc) + if err != nil { + panic("server: runtime init failed: " + err.Error()) + } + return NewFromRuntime(rt) +} + // AuthService 获取认证服务 func (s *Server) AuthService() *auth.Service { return s.authSvc @@ -426,6 +324,9 @@ func (s *Server) serveSPA(w http.ResponseWriter, r *http.Request) { // Start 启动 HTTP 守护进程 func (s *Server) Start() error { + if s.rt != nil { + _ = s.rt.Start(context.Background()) + } addr := fmt.Sprintf("%s:%d", s.cfg.HTTPHost, s.cfg.HTTPPort) s.httpServer = &http.Server{ Addr: addr, @@ -441,6 +342,9 @@ func (s *Server) Start() error { // Shutdown 优雅停机 func (s *Server) Shutdown(ctx context.Context) error { + if s.rt != nil { + _ = s.rt.Stop(ctx) + } if s.httpServer != nil { return s.httpServer.Shutdown(ctx) } @@ -558,11 +462,18 @@ func (s *Server) getOrCreateSession( BlackboardFacts: make(map[string]string), AvailableActions: coordinator.ComputeAvailableActions(bizdev.StageV01_Triage, u.ProjectRole), } - if targetMarket != "" { - userCtx.BlackboardFacts["target_market"] = targetMarket - } - if deviceType != "" { - userCtx.BlackboardFacts["device_type"] = deviceType + if targetMarket != "" || deviceType != "" { + fill := map[string]string{} + if targetMarket != "" { + fill["target_market"] = targetMarket + } + if deviceType != "" { + fill["device_type"] = deviceType + } + _ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{ + ExpectedVersion: userCtx.Version, + BlackboardFill: fill, + }) } } @@ -586,10 +497,11 @@ func (s *Server) getOrCreateSession( if restored.ProjectID > 0 { sess.ProjectID = restored.ProjectID } - for k, v := range restored.Facts { - if _, exists := userCtx.BlackboardFacts[k]; !exists { - userCtx.BlackboardFacts[k] = v - } + if len(restored.Facts) > 0 { + _ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{ + ExpectedVersion: userCtx.Version, + BlackboardFill: restored.Facts, + }) } if s.chatLog == nil { for _, rm := range restored.Messages { @@ -607,7 +519,11 @@ func (s *Server) getOrCreateSession( if sess.UserID > 0 && sess.CtxMgr != nil { if guidance, err := sess.CtxMgr.CompilePromptGuidance(context.Background(), sess.UserID); err == nil && guidance != "" { - userCtx.PersonalizedGuidance = guidance + _ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{ + ExpectedVersion: userCtx.Version, + SetPersonalizedGuidance: true, + PersonalizedGuidance: guidance, + }) } } } @@ -641,6 +557,16 @@ func (s *Server) hydrateVisibleHistory(ctx context.Context, sess *chat.Session, sess.SyncRecentDialogue() } +func (s *Server) taskUseCase() *stageflow.UseCase { + if s == nil { + return stageflow.New(nil) + } + if s.stageflow != nil { + return s.stageflow + } + return stageflow.New(s.svc) +} + func (s *Server) corsMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { s.writeCORS(w, r) diff --git a/internal/server/sse.go b/internal/server/sse.go index 921fcad3..171dbe2b 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -13,6 +13,7 @@ import ( "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/squad" "chat-agent-system/internal/stream" + "chat-agent-system/internal/stream/transport" "chat-agent-system/internal/taskrun" ) @@ -55,7 +56,7 @@ func extendSSEWriteDeadline(w http.ResponseWriter) { func writeSSEHeartbeat(w http.ResponseWriter, flusher http.Flusher) error { extendSSEWriteDeadline(w) - if _, err := fmt.Fprintf(w, ": heartbeat %d\n\n", time.Now().Unix()); err != nil { + if err := (transport.Encoder{W: w}).Heartbeat(time.Now()); err != nil { return err } flusher.Flush() @@ -63,11 +64,7 @@ func writeSSEHeartbeat(w http.ResponseWriter, flusher http.Flusher) error { } func writeSSE(w http.ResponseWriter, flusher http.Flusher, ev stream.StreamEvent) { - data, err := json.Marshal(ev) - if err != nil { - return - } - fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, string(data)) + _ = (transport.Encoder{W: w}).Write(ev) flusher.Flush() } @@ -236,14 +233,7 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { listener := func(ev stream.StreamEvent) { writeMu.Lock() defer writeMu.Unlock() - - data, err := json.Marshal(ev) - if err != nil { - return - } - // 按照标准 SSE 协议规范发射: event: \ndata: \n\n - fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, string(data)) - flusher.Flush() + writeSSE(w, flusher, ev) } clientTurnID := req.ClientTurnID @@ -302,15 +292,12 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { if err != nil { writeMu.Lock() info := stream.ClassifyError(err, "chat_stream", requestID, sess.ID, "") - errEv := stream.StreamEvent{ + writeSSE(w, flusher, stream.StreamEvent{ Type: stream.EventError, Timestamp: time.Now(), Error: info.UserMessage, ErrorInfo: &info, - } - data, _ := json.Marshal(errEv) - fmt.Fprintf(w, "event: %s\ndata: %s\n\n", errEv.Type, string(data)) - flusher.Flush() + }) writeMu.Unlock() } } @@ -408,12 +395,7 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) { defer s.taskRuns.Unsubscribe(created.RunID, ch) writeStored := func(stored taskrun.StoredEvent) { - data, marshalErr := json.Marshal(stored.Event) - if marshalErr != nil { - return - } - fmt.Fprintf(w, "id: %d\n", stored.Seq) - fmt.Fprintf(w, "event: %s\ndata: %s\n\n", stored.Event.Type, string(data)) + _ = (transport.Encoder{W: w}).WriteWithID(stored.Seq, stored.Event) flusher.Flush() } diff --git a/internal/server/task_run_handlers.go b/internal/server/task_run_handlers.go index a25babf6..675361a2 100644 --- a/internal/server/task_run_handlers.go +++ b/internal/server/task_run_handlers.go @@ -708,7 +708,7 @@ func (s *Server) buildBizdevStageRequest(dto TaskRunRequestDTO, sessionID, clien func (s *Server) initTaskRunExecutors() { s.taskRuns.SetExecutor( func(ctx context.Context, _ string, req bizdev.TaskRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) { - return s.svc.ExecuteTask(ctx, req) + return s.taskUseCase().ExecuteTask(ctx, req) }, func(ctx context.Context, _ string, workDir string, _ *stream.Streamer) (*bizdev.TaskRunResult, error) { res, err := s.svc.ResumeOrModifyWorkspace(ctx, squad.ModifyWorkflowRequest{WorkDir: workDir}) @@ -722,7 +722,7 @@ func (s *Server) initTaskRunExecutors() { return out, err }, func(ctx context.Context, _ string, req bizdev.StageRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) { - stageRes, err := s.svc.ExecuteStage(ctx, req) + stageRes, err := s.taskUseCase().ExecuteStage(ctx, req) if stageRes == nil { return nil, err } diff --git a/internal/stream/transport/sse.go b/internal/stream/transport/sse.go new file mode 100644 index 00000000..0ad76d40 --- /dev/null +++ b/internal/stream/transport/sse.go @@ -0,0 +1,47 @@ +package transport + +import ( + "encoding/json" + "fmt" + "io" + "time" + + "chat-agent-system/internal/stream" +) + +// Encoder writes neutral SSE frames. It does not own retries, leases, or task state. +type Encoder struct { + W io.Writer +} + +func (e Encoder) Write(ev stream.StreamEvent) error { + return e.WriteWithID(0, ev) +} + +func (e Encoder) WriteWithID(id int64, ev stream.StreamEvent) error { + if e.W == nil { + return nil + } + if ev.Timestamp.IsZero() { + ev.Timestamp = time.Now() + } + data, err := json.Marshal(ev) + if err != nil { + return err + } + if id > 0 { + if _, err = fmt.Fprintf(e.W, "id: %d\n", id); err != nil { + return err + } + } + _, err = fmt.Fprintf(e.W, "event: %s\ndata: %s\n\n", ev.Type, string(data)) + return err +} + +func (e Encoder) Heartbeat(now time.Time) error { + if e.W == nil { + return nil + } + _, err := fmt.Fprintf(e.W, ": heartbeat %d\n\n", now.Unix()) + return err +} diff --git a/internal/stream/transport/sse_test.go b/internal/stream/transport/sse_test.go new file mode 100644 index 00000000..a615edff --- /dev/null +++ b/internal/stream/transport/sse_test.go @@ -0,0 +1,35 @@ +package transport + +import ( + "bytes" + "strings" + "testing" + "time" + + "chat-agent-system/internal/stream" +) + +func TestWriteNeutralFrame(t *testing.T) { + var buf bytes.Buffer + enc := Encoder{W: &buf} + if err := enc.Write(stream.StreamEvent{Type: stream.EventDone, Timestamp: time.Unix(1, 0)}); err != nil { + t.Fatal(err) + } + got := buf.String() + if !strings.Contains(got, "event: done") || !strings.Contains(got, "data: {") { + t.Fatalf("%q", got) + } + if strings.Contains(got, "retry:") { + t.Fatal("transport must not own retry") + } +} + +func TestHeartbeat(t *testing.T) { + var buf bytes.Buffer + if err := (Encoder{W: &buf}).Heartbeat(time.Unix(10, 0)); err != nil { + t.Fatal(err) + } + if !strings.Contains(buf.String(), ": heartbeat 10") { + t.Fatalf("%q", buf.String()) + } +} diff --git a/scripts/arch-check.sh b/scripts/arch-check.sh index fc82c2a2..4ceb73b0 100755 --- a/scripts/arch-check.sh +++ b/scripts/arch-check.sh @@ -31,7 +31,7 @@ fi if [[ "${MODE}" == "block" ]]; then if ! ARCHTEST_STRICT=1 go test ./internal/archtest -count=1 -run TestArchitectureBoundaries; then - echo "arch-check block: baseline exceptions remain; only cleared rules may use --mode block" >&2 + echo "arch-check block: a cleared kind in block_kinds still has findings" >&2 exit 1 fi fi