Merge pull request #3230 from DaydreamCoding/feat/openai-cyber-policy-passthrough

feat(openai): cyber_policy 硬阻断全链路透传、审计与计费

backend/internal/repository/content_moderation_repo.go

@@ -177,10 +177,11 @@ LIMIT $`+fmt.Sprint(len(queryArgs)-1)+` OFFSET $`+fmt.Sprint(len(queryArgs)),
 	return items, paginationResultFromTotal(total, params), nil
 }
 
-func (r *contentModerationRepository) CountFlaggedByUserSince(ctx context.Context, userID int64, since time.Time) (int, error) {
+func (r *contentModerationRepository) CountFlaggedByUserSince(ctx context.Context, userID int64, since time.Time, excludeCyberPolicy bool) (int, error) {
 	if userID <= 0 {
 		return 0, nil
 	}
+	// SQL 中的 'cyber_policy' 字面量须与 service.ContentModerationActionCyberPolicy 保持一致。
 	var count int
 	err := r.db.QueryRowContext(ctx, `
 WITH last_auto_ban AS (
@@ -193,15 +194,24 @@ FROM content_moderation_logs
 WHERE user_id = $1
   AND flagged = TRUE
   AND action <> 'hash_block'
+  AND ($3::bool IS FALSE OR action <> 'cyber_policy')
   AND created_at >= $2
   AND created_at > COALESCE((SELECT at FROM last_auto_ban), '-infinity'::timestamptz)
-`, userID, since).Scan(&count)
+`, userID, since, excludeCyberPolicy).Scan(&count)
 	if err != nil {
 		return 0, fmt.Errorf("count user content moderation flagged logs: %w", err)
 	}
 	return count, nil
 }
 
+func (r *contentModerationRepository) UpdateLogEmailSent(ctx context.Context, id int64, sent bool) error {
+	_, err := r.db.ExecContext(ctx, `UPDATE content_moderation_logs SET email_sent = $1 WHERE id = $2`, sent, id)
+	if err != nil {
+		return fmt.Errorf("update content moderation log email_sent: %w", err)
+	}
+	return nil
+}
+
 func (r *contentModerationRepository) CleanupExpiredLogs(ctx context.Context, hitBefore time.Time, nonHitBefore time.Time) (*service.ContentModerationCleanupResult, error) {
 	result := &service.ContentModerationCleanupResult{FinishedAt: time.Now()}
 	if r == nil || r.db == nil {

backend/internal/repository/content_moderation_repo_test.go

@@ -29,12 +29,30 @@ func TestContentModerationRepositoryCountFlaggedByUserSince_ExcludesHashBlock(t
 	repo := NewContentModerationRepository(db)
 	since := time.Now().Add(-time.Hour)
 	mock.ExpectQuery(regexp.QuoteMeta("AND action <> 'hash_block'")).
-		WithArgs(int64(1001), since).
+		WithArgs(int64(1001), since, false).
 		WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(2))
 
-	count, err := repo.CountFlaggedByUserSince(context.Background(), 1001, since)
+	count, err := repo.CountFlaggedByUserSince(context.Background(), 1001, since, false)
 
 	require.NoError(t, err)
 	require.Equal(t, 2, count)
 	require.NoError(t, mock.ExpectationsWereMet())
 }
+
+func TestContentModerationRepositoryCountFlaggedByUserSince_ExcludesCyberPolicyWhenRequested(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	require.NoError(t, err)
+	defer func() { _ = db.Close() }()
+
+	repo := NewContentModerationRepository(db)
+	since := time.Now().Add(-time.Hour)
+	mock.ExpectQuery(regexp.QuoteMeta("AND ($3::bool IS FALSE OR action <> 'cyber_policy')")).
+		WithArgs(int64(1001), since, true).
+		WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(3))
+
+	count, err := repo.CountFlaggedByUserSince(context.Background(), 1001, since, true)
+
+	require.NoError(t, err)
+	require.Equal(t, 3, count)
+	require.NoError(t, mock.ExpectationsWereMet())
+}

backend/internal/repository/gateway_cache.go

@@ -51,3 +51,23 @@ func (c *gatewayCache) DeleteSessionAccountID(ctx context.Context, groupID int64
 	key := buildSessionKey(groupID, sessionHash)
 	return c.rdb.Del(ctx, key).Err()
 }
+
+// Compile-time assertion: gatewayCache must implement CyberSessionBlockStore.
+var _ service.CyberSessionBlockStore = (*gatewayCache)(nil)
+
+const cyberSessionBlockPrefix = "cyber_session_block:"
+
+// SetCyberSessionBlocked 把被 cyber_policy 命中的会话写入屏蔽表（TTL 自动过期）。
+// 存储值 "1" 作为存在标记（IsCyberSessionBlocked 只检查 key 是否存在，不读值）。
+func (c *gatewayCache) SetCyberSessionBlocked(ctx context.Context, key string, ttl time.Duration) error {
+	return c.rdb.Set(ctx, cyberSessionBlockPrefix+key, "1", ttl).Err()
+}
+
+// IsCyberSessionBlocked 查询会话是否在屏蔽表中。
+func (c *gatewayCache) IsCyberSessionBlocked(ctx context.Context, key string) (bool, error) {
+	n, err := c.rdb.Exists(ctx, cyberSessionBlockPrefix+key).Result()
+	if err != nil {
+		return false, err
+	}
+	return n > 0, nil
+}

backend/internal/repository/ops_error_where_test.go

@@ -70,6 +70,28 @@ func TestBuildOpsErrorLogsWhere_ModelFuzzy(t *testing.T) {
 	}
 }
 
+// TestBuildOpsErrorLogsWhere_CyberPolicyStatusExemption verifies that streaming
+// cyber_policy hits (status_code=200) remain visible in admin + user error-request
+// lists.  The repository filter must emit an OR exemption for error_type='cyber_policy'
+// so that stream-path cyber rows (upstream delivers 200 with a failed SSE event) are
+// not silently excluded by the COALESCE(status_code,0) >= 400 guard.
+func TestBuildOpsErrorLogsWhere_CyberPolicyStatusExemption(t *testing.T) {
+	// Default filter (no phase) must include the cyber_policy exemption.
+	where, _ := buildOpsErrorLogsWhere(&service.OpsErrorLogFilter{})
+	if !strings.Contains(where, "e.error_type = 'cyber_policy'") {
+		t.Fatalf("default filter must exempt cyber_policy from status >= 400 guard\nfull: %s", where)
+	}
+	if !strings.Contains(where, "COALESCE(e.status_code, 0) >= 400") {
+		t.Fatalf("default filter must still include the status >= 400 guard for non-cyber rows\nfull: %s", where)
+	}
+
+	// phase=upstream skips the status guard entirely — exemption is irrelevant there.
+	whereUpstream, _ := buildOpsErrorLogsWhere(&service.OpsErrorLogFilter{Phase: "upstream"})
+	if strings.Contains(whereUpstream, "status_code") {
+		t.Fatalf("upstream phase filter must not add any status_code clause\nfull: %s", whereUpstream)
+	}
+}
+
 func TestBuildOpsErrorLogsWhere_MatchDeletedKeyOwner(t *testing.T) {
 	uid := int64(42)
 

backend/internal/repository/ops_repo.go

@@ -919,8 +919,12 @@ func buildOpsErrorLogsWhere(filter *service.OpsErrorLogFilter) (string, []any) {
 		resolvedFilter = filter.Resolved
 	}
 	// Keep list endpoints scoped to client errors unless explicitly filtering upstream phase.
+	// cyber_policy is exempt from the status >= 400 guard: streaming cyber hits arrive with
+	// status 200 (the SSE stream opened successfully before upstream returned response.failed),
+	// but they are always client-visible blocked requests that belong in admin + user error
+	// lists.  Without the exemption the entire streaming-path cyber sink would be invisible.
 	if phaseFilter != "upstream" {
-		clauses = append(clauses, "COALESCE(e.status_code, 0) >= 400")
+		clauses = append(clauses, "(COALESCE(e.status_code, 0) >= 400 OR e.error_type = 'cyber_policy')")
 	}
 
 	if filter.StartTime != nil && !filter.StartTime.IsZero() {