diff --git a/.doc/test-reports/2026-09-12-phase3-project-workflow-governance-test-report.md b/.doc/test-reports/2026-09-12-phase3-project-workflow-governance-test-report.md index 9a25c5ab..538a01dd 100644 --- a/.doc/test-reports/2026-09-12-phase3-project-workflow-governance-test-report.md +++ b/.doc/test-reports/2026-09-12-phase3-project-workflow-governance-test-report.md @@ -96,7 +96,9 @@ ok chat-agent-system/internal/server 0.792s (TestProjectWorkflow*) 2. ✅ **阶段级 ABAC MVP**（文档 list/content 按 active 阶段 + 三权过滤；`TestResolveStageAccess_ActiveStageDutyOnly` PASS） 3. ✅ **Phase 4 MVP 通知**（`00018` Outbox + 去重通知；文档批准联动 readiness；前端 15s 轮询刷新推进卡片） 4. ✅ **聊天/搜索/SSE ABAC MVP**（发送拦截 + 历史/搜索按 `metadata.stage_key` 过滤；新消息写入 stage_key） - 5. ⏳ **Phase 4 续**（24h 提醒升级；Agent 工具读取 ABAC；真实 Agent E2E） + 5. ✅ **24h 提醒升级**（后台 15min 扫描；`stage_ready_reminder_24h` 去重；推进后 `confirmed` 停止提醒） + 6. ✅ **项目上下文 ABAC**（`GET /context/project` 需认证且至少一个允许阶段） + 7. ⏳ **真实 Agent E2E**（pi-agent + MiniMax-M3 全链路验收） ### 顺序影响说明 @@ -115,3 +117,7 @@ ok chat-agent-system/internal/server 0.792s (TestProjectWorkflow*) | `TestHandleDocumentApproved_StageReady` | PASS | 文档批准后同步 task 状态并评估 stage ready | | `TestPG_EmitStageReadyDedupe` | PASS | `(stage_instance_id, readiness_revision)` 去重，重复 emit 不产生第二条通知 | | `TestPG_ActivationAndStageAdvance`（含 notify 段） | PASS | 批准 → HandleDocumentApproved → EmitStageReadyToAdvance 全链路 | +| `TestResolveStageAccess_ActiveStageDutyOnly` | PASS | 非三权用户不可访问 active 阶段文档/聊天 | +| `TestStageKeyAllowed` / `TestMessageStageKey` | PASS | 阶段键允许判定与 metadata 提取 | +| `TestServer_ChatSearchAndDigestAPIs` | PASS | 搜索 API 在认证下仍正常工作 | +| `TestPG_StageAdvanceReminder24h` | PASS | 24h 升级提醒去重；确认推进后不再扫描 | diff --git a/internal/notification/reminder.go b/internal/notification/reminder.go new file mode 100644 index 00000000..b9f524d3 --- /dev/null +++ b/internal/notification/reminder.go @@ -0,0 +1,172 @@ +package notification + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/jackc/pgx/v5" +) + +// stageReminderAfter 首次提醒后多久发送 24h 升级提醒（测试可覆盖）。 +var stageReminderAfter = 24 * time.Hour + +// SetStageReminderAfterForTest 仅供测试缩短等待窗口。 +func SetStageReminderAfterForTest(d time.Duration) { + if d > 0 { + stageReminderAfter = d + } +} + +type pendingAdvanceRow struct { + ID int64 + StageInstanceID int64 + WorkflowInstanceID int64 + StageKey string + ReadinessRevision int + ProjectID int64 + CreatorUserID int64 + StageName string + RequiredDocCount int + ApprovedDocCount int +} + +// ProcessStageAdvanceReminders 扫描 pending 推进待办，对超过 24h 未处理的创建升级提醒（去重）。 +func (s *Service) ProcessStageAdvanceReminders(ctx context.Context) (int, error) { + if !s.Available() { + return 0, ErrNotAvailable + } + + cutoff := time.Now().Add(-stageReminderAfter) + rows, err := s.pool.Query(ctx, ` + SELECT sar.id, sar.stage_instance_id, sar.workflow_instance_id, sar.stage_key, sar.readiness_revision, + pwi.project_id, COALESCE(p.creator_user_id, 0), + COALESCE(e.payload, '{}'::jsonb) + FROM stage_advance_requests sar + JOIN project_workflow_instances pwi ON pwi.id = sar.workflow_instance_id + JOIN projects p ON p.id = pwi.project_id + LEFT JOIN LATERAL ( + SELECT payload FROM outbox_events + WHERE aggregate_type = 'stage_instance' + AND aggregate_id = 'stage_instance:' || sar.stage_instance_id::text + AND event_type = $1 + ORDER BY id DESC LIMIT 1 + ) e ON true + WHERE sar.status = 'pending' AND sar.requested_at <= $2 + `, EventStageReadyToAdvance, cutoff) + if err != nil { + return 0, err + } + defer rows.Close() + + sent := 0 + for rows.Next() { + var row pendingAdvanceRow + var payloadRaw []byte + if err := rows.Scan( + &row.ID, &row.StageInstanceID, &row.WorkflowInstanceID, &row.StageKey, &row.ReadinessRevision, + &row.ProjectID, &row.CreatorUserID, &payloadRaw, + ); err != nil { + return sent, err + } + if row.CreatorUserID <= 0 { + continue + } + if len(payloadRaw) > 0 { + var payload StageReadyPayload + _ = json.Unmarshal(payloadRaw, &payload) + row.StageName = payload.StageName + row.RequiredDocCount = payload.RequiredDocCount + row.ApprovedDocCount = payload.ApprovedDocCount + } + created, err := s.emitStageReminder24(ctx, row) + if err != nil { + return sent, err + } + if created { + sent++ + } + } + return sent, rows.Err() +} + +func (s *Service) emitStageReminder24(ctx context.Context, row pendingAdvanceRow) (bool, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return false, err + } + defer tx.Rollback(ctx) + + payload := StageReadyPayload{ + ProjectID: row.ProjectID, + WorkflowInstanceID: row.WorkflowInstanceID, + StageInstanceID: row.StageInstanceID, + StageKey: row.StageKey, + StageName: row.StageName, + ConfigRevision: row.ReadinessRevision, + RequiredDocCount: row.RequiredDocCount, + ApprovedDocCount: row.ApprovedDocCount, + } + raw, err := json.Marshal(payload) + if err != nil { + return false, err + } + + dedupeKey := fmt.Sprintf("stage_ready_remind24:%d:%d", row.StageInstanceID, row.ReadinessRevision) + var existingID int64 + err = tx.QueryRow(ctx, ` + SELECT id FROM notifications WHERE user_id = $1 AND dedupe_key = $2 + `, row.CreatorUserID, dedupeKey).Scan(&existingID) + if err == nil && existingID > 0 { + return false, nil + } + if err != nil && err != pgx.ErrNoRows { + return false, err + } + + var eventID int64 + aggregateID := fmt.Sprintf("stage_instance:%d", row.StageInstanceID) + err = tx.QueryRow(ctx, ` + INSERT INTO outbox_events (aggregate_type, aggregate_id, event_type, payload, project_id, published_at) + VALUES ('stage_instance', $1, $2, $3::jsonb, $4, now()) + RETURNING id + `, aggregateID, EventStageReadyReminder24, string(raw), row.ProjectID).Scan(&eventID) + if err != nil { + return false, err + } + + stageLabel := row.StageName + if stageLabel == "" { + stageLabel = row.StageKey + } + title := fmt.Sprintf("【提醒】阶段「%s」仍可推进", stageLabel) + body := "必需文档已全部批准，已超过 24 小时尚未确认推进。请尽快查看完成清单并确认是否进入下一阶段。" + + _, err = tx.Exec(ctx, ` + INSERT INTO notifications (user_id, project_id, event_id, type, title, body, dedupe_key, status) + VALUES ($1, $2, $3, $4, $5, $6, $7, 'unread') + ON CONFLICT (user_id, dedupe_key) DO NOTHING + `, row.CreatorUserID, row.ProjectID, eventID, TypeStageReadyReminder24, title, body, dedupeKey) + if err != nil { + return false, err + } + + if err := tx.Commit(ctx); err != nil { + return false, err + } + return true, nil +} + +// ConfirmStageAdvanceRequest 阶段已确认推进后关闭 pending 待办，停止后续提醒。 +func (s *Service) ConfirmStageAdvanceRequest(ctx context.Context, stageInstanceID int64, readinessRevision int, decidedBy int64) error { + if !s.Available() || stageInstanceID <= 0 { + return nil + } + _, err := s.pool.Exec(ctx, ` + UPDATE stage_advance_requests + SET status = 'confirmed', decided_by = $3, decided_at = now() + WHERE stage_instance_id = $1 AND readiness_revision = $2 AND status = 'pending' + `, stageInstanceID, readinessRevision, decidedBy) + return err +} diff --git a/internal/notification/reminder_test.go b/internal/notification/reminder_test.go new file mode 100644 index 00000000..6fc75659 --- /dev/null +++ b/internal/notification/reminder_test.go @@ -0,0 +1,110 @@ +package notification_test + +import ( + "context" + "testing" + "time" + + "chat-agent-system/internal/notification" +) + +func TestPG_StageAdvanceReminder24h(t *testing.T) { + pool := openPool(t) + defer pool.Close() + + notification.SetStageReminderAfterForTest(1 * time.Millisecond) + t.Cleanup(func() { + notification.SetStageReminderAfterForTest(24 * time.Hour) + }) + + svc := notification.NewService(pool) + ctx := context.Background() + + var projectID, instanceID, stageInstID int64 + code := "remind-" + time.Now().Format("150405.000") + if err := pool.QueryRow(ctx, ` + INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status) + VALUES ($1, 'Reminder Test', 'active', 2, 'active') RETURNING id + `, code).Scan(&projectID); err != nil { + t.Fatalf("insert project: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID) + }) + + var wfVersionID int64 + if err := pool.QueryRow(ctx, `SELECT id FROM workflow_versions ORDER BY id DESC LIMIT 1`).Scan(&wfVersionID); err != nil { + t.Skip("no workflow version") + } + + if err := pool.QueryRow(ctx, ` + INSERT INTO project_workflow_instances (project_id, workflow_version_id, instance_no, status, config_revision, plan_hash, config_snapshot, created_by) + VALUES ($1, $2, 1, 'active', 5, 'h', '{}'::jsonb, 2) RETURNING id + `, projectID, wfVersionID).Scan(&instanceID); err != nil { + t.Fatalf("insert instance: %v", err) + } + + if err := pool.QueryRow(ctx, ` + INSERT INTO project_stage_instances (workflow_instance_id, stage_key, position, enabled, status) + VALUES ($1, 'biz_intake', 0, true, 'active') RETURNING id + `, instanceID).Scan(&stageInstID); err != nil { + t.Fatalf("insert stage: %v", err) + } + + created, err := svc.EmitStageReadyToAdvance(ctx, 2, notification.StageReadyPayload{ + ProjectID: projectID, + WorkflowInstanceID: instanceID, + StageInstanceID: stageInstID, + StageKey: "biz_intake", + StageName: "商务摸底", + ConfigRevision: 5, + RequiredDocCount: 1, + ApprovedDocCount: 1, + }) + if err != nil || !created { + t.Fatalf("initial emit: created=%v err=%v", created, err) + } + + _, err = pool.Exec(ctx, ` + UPDATE stage_advance_requests SET requested_at = now() - interval '25 hours' + WHERE stage_instance_id = $1 AND readiness_revision = 5 + `, stageInstID) + if err != nil { + t.Fatalf("backdate request: %v", err) + } + + if _, err := svc.ProcessStageAdvanceReminders(ctx); err != nil { + t.Fatalf("process reminders: %v", err) + } + + var remindCount int + err = pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM notifications + WHERE user_id = 2 AND project_id = $1 AND type = $2 + `, projectID, notification.TypeStageReadyReminder24).Scan(&remindCount) + if err != nil { + t.Fatalf("count reminders: %v", err) + } + if remindCount != 1 { + t.Fatalf("expected 1 reminder notification for test project, got %d", remindCount) + } + + if _, err := svc.ProcessStageAdvanceReminders(ctx); err != nil { + t.Fatalf("process reminders again: %v", err) + } + err = pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM notifications + WHERE user_id = 2 AND project_id = $1 AND type = $2 + `, projectID, notification.TypeStageReadyReminder24).Scan(&remindCount) + if err != nil || remindCount != 1 { + t.Fatalf("expected deduped reminder count 1, got %d err=%v", remindCount, err) + } + + if err := svc.ConfirmStageAdvanceRequest(ctx, stageInstID, 5, 2); err != nil { + t.Fatalf("confirm: %v", err) + } + + if _, err := svc.ProcessStageAdvanceReminders(ctx); err != nil { + t.Fatalf("process after confirm: %v", err) + } +} diff --git a/internal/notification/types.go b/internal/notification/types.go index 4e6b7773..6c829e83 100644 --- a/internal/notification/types.go +++ b/internal/notification/types.go @@ -3,8 +3,10 @@ package notification import "time" const ( - EventStageReadyToAdvance = "stage.ready_to_advance" - TypeStageReadyToAdvance = "stage_ready_to_advance" + EventStageReadyToAdvance = "stage.ready_to_advance" + EventStageReadyReminder24 = "stage.ready_to_advance.reminder24" + TypeStageReadyToAdvance = "stage_ready_to_advance" + TypeStageReadyReminder24 = "stage_ready_reminder_24h" ) // StageReadyPayload Outbox / 通知载荷。 diff --git a/internal/projectinstance/stage_service.go b/internal/projectinstance/stage_service.go index 88b2d5e3..de580e21 100644 --- a/internal/projectinstance/stage_service.go +++ b/internal/projectinstance/stage_service.go @@ -86,6 +86,13 @@ func (s *Service) AdvanceStage(ctx context.Context, instanceID int64, userID int return nil, err } completedStage := before.ActiveStageKey + var completedStageInstanceID int64 + for _, st := range before.Stages { + if st.StageKey == in.StageKey { + completedStageInstanceID = st.ID + break + } + } after, err := s.store.AdvanceStage(ctx, instanceID, in.StageKey, in.ExpectedConfigRevision) if err != nil { @@ -93,9 +100,11 @@ func (s *Service) AdvanceStage(ctx context.Context, instanceID int64, userID int } return &AdvanceStageResult{ - Instance: after.Instance, - CompletedStage: completedStage, - NextStageKey: after.ActiveStageKey, - InstanceDone: after.Instance.Status == InstanceCompleted, + Instance: after.Instance, + CompletedStage: completedStage, + CompletedStageInstanceID: completedStageInstanceID, + CompletedReadinessRevision: in.ExpectedConfigRevision, + NextStageKey: after.ActiveStageKey, + InstanceDone: after.Instance.Status == InstanceCompleted, }, nil } diff --git a/internal/projectinstance/types.go b/internal/projectinstance/types.go index 59d5d48d..c317c027 100644 --- a/internal/projectinstance/types.go +++ b/internal/projectinstance/types.go @@ -206,8 +206,10 @@ type AdvanceStageInput struct { // AdvanceStageResult 阶段推进结果。 type AdvanceStageResult struct { - Instance WorkflowInstance `json:"instance"` - CompletedStage string `json:"completed_stage"` - NextStageKey string `json:"next_stage_key,omitempty"` - InstanceDone bool `json:"instance_done"` + Instance WorkflowInstance `json:"instance"` + CompletedStage string `json:"completed_stage"` + CompletedStageInstanceID int64 `json:"completed_stage_instance_id,omitempty"` + CompletedReadinessRevision int `json:"completed_readiness_revision,omitempty"` + NextStageKey string `json:"next_stage_key,omitempty"` + InstanceDone bool `json:"instance_done"` } diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 92b5c75f..c9a1f84c 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -419,6 +419,11 @@ func (s *Server) handleUpdateUserPreference(w http.ResponseWriter, r *http.Reque // handleGetProjectContext 获取项目级多方协同事实黑板与里程碑决策流 func (s *Server) handleGetProjectContext(w http.ResponseWriter, r *http.Request) { + user, ok := s.requireAuthenticatedUser(w, r) + if !ok { + return + } + projectIDStr := r.URL.Query().Get("project_id") projectID, _ := strconv.ParseInt(projectIDStr, 10, 64) if projectID <= 0 { @@ -426,6 +431,11 @@ func (s *Server) handleGetProjectContext(w http.ResponseWriter, r *http.Request) return } + if !s.allowAnyStageAccess(r.Context(), projectID, user.ID) { + s.writeWorkflowAccessDenied(w) + return + } + if s.svc == nil || s.svc.ContextManager() == nil { http.Error(w, "context manager not initialized", http.StatusServiceUnavailable) return diff --git a/internal/server/notification_reminder_loop.go b/internal/server/notification_reminder_loop.go new file mode 100644 index 00000000..7210756d --- /dev/null +++ b/internal/server/notification_reminder_loop.go @@ -0,0 +1,37 @@ +package server + +import ( + "context" + "log/slog" + "time" +) + +const defaultReminderPollInterval = 15 * time.Minute + +func (s *Server) startNotificationReminderLoop() { + if s == nil || s.notifySvc == nil || !s.notifySvc.Available() { + return + } + go func() { + ticker := time.NewTicker(defaultReminderPollInterval) + defer ticker.Stop() + s.runStageAdvanceReminders(context.Background()) + for range ticker.C { + s.runStageAdvanceReminders(context.Background()) + } + }() +} + +func (s *Server) runStageAdvanceReminders(ctx context.Context) { + if s.notifySvc == nil || !s.notifySvc.Available() { + return + } + n, err := s.notifySvc.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/server/server.go b/internal/server/server.go index 8530ba81..474cc9e2 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -103,6 +103,7 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { resolver: workspace.NewResolver(cfg.DataDir, pool), } s.registerWorkflowApprovalHook() + s.startNotificationReminderLoop() s.initTaskRunExecutors() return s } diff --git a/internal/server/stage_access_helpers.go b/internal/server/stage_access_helpers.go index dfb7bdc2..c69f3d26 100644 --- a/internal/server/stage_access_helpers.go +++ b/internal/server/stage_access_helpers.go @@ -31,6 +31,14 @@ func (s *Server) allowWorkflowDocTaskAccess(ctx context.Context, projectID, user return access.TaskKeyAllowed(taskKey) } +func (s *Server) allowAnyStageAccess(ctx context.Context, projectID, userID int64) bool { + access, err := s.resolveStageAccess(ctx, projectID, userID) + if err != nil || access == nil || !access.Restricted { + return true + } + return len(access.AllowedStageKeys) > 0 +} + func (s *Server) allowActiveStageChatAccess(ctx context.Context, projectID, userID int64) bool { access, err := s.resolveStageAccess(ctx, projectID, userID) if err != nil || access == nil || !access.Restricted { diff --git a/internal/server/workflow_instance_handlers.go b/internal/server/workflow_instance_handlers.go index 4a81a242..991b3596 100644 --- a/internal/server/workflow_instance_handlers.go +++ b/internal/server/workflow_instance_handlers.go @@ -92,5 +92,13 @@ func (s *Server) handleAdvanceStage(w http.ResponseWriter, r *http.Request) { writeProjectInstanceError(w, err) return } + if s.notifySvc != nil && result.CompletedStageInstanceID > 0 { + _ = s.notifySvc.ConfirmStageAdvanceRequest( + r.Context(), + result.CompletedStageInstanceID, + result.CompletedReadinessRevision, + user.ID, + ) + } writeJSONResponse(w, http.StatusOK, result) } diff --git a/web/src/components/workflow/StageAdvanceBanner.tsx b/web/src/components/workflow/StageAdvanceBanner.tsx index fc4f5297..0fbeb03a 100644 --- a/web/src/components/workflow/StageAdvanceBanner.tsx +++ b/web/src/components/workflow/StageAdvanceBanner.tsx @@ -59,7 +59,9 @@ export const StageAdvanceBanner: React.FC = ({ try { const items = await fetchProjectNotifications(projectId, lastNotificationId); if (cancelled || items.length === 0) return; - const stageReady = items.some((n) => n.type === 'stage_ready_to_advance'); + const stageReady = items.some( + (n) => n.type === 'stage_ready_to_advance' || n.type === 'stage_ready_reminder_24h' + ); const maxId = items.reduce((max, n) => Math.max(max, n.id), lastNotificationId); setLastNotificationId(maxId); if (stageReady) {