fix: harden controlled AI session consistency

2026-07-21 19:54:25 +08:00
parent 82829645c2
commit b7de3c28b6
16 changed files with 497 additions and 69 deletions
--- a/apps/web_v1/scripts/structure-check.mjs
+++ b/apps/web_v1/scripts/structure-check.mjs
@@ -117,12 +117,17 @@ const aiApiSource = existsSync('src/api/ai.ts') ? readFileSync('src/api/ai.ts',
 for (const required of ['/api/v1/projects/', '/ai-sessions', 'listAISessions', 'createAISession']) {
  if (!aiApiSource.includes(required)) failures.push(`AI API must include ${required}`)
 }
+if (!aiApiSource.includes('signal?: AbortSignal')) failures.push('AI API requests must accept an AbortSignal')
+if (!aiApiSource.includes('signal,')) failures.push('AI API requests must pass the AbortSignal to apiRequest')
 if (/\b(?:task|note|source)Id\b/.test(aiApiSource)) failures.push('AI session responses must not expose automatic formal object IDs')

 const aiPageSource = existsSync('src/pages/projects/project-ai.tsx') ? readFileSync('src/pages/projects/project-ai.tsx', 'utf8') : ''
 for (const required of ['AI 助手', '创建会话', 'loading', 'error']) {
  if (!aiPageSource.includes(required)) failures.push(`project AI page must include ${required}`)
 }
+for (const required of ['AbortController', 'generationRef', 'projectRef', 'setSessions([])']) {
+  if (!aiPageSource.includes(required)) failures.push(`project AI page must gate stale requests with ${required}`)
+}
 for (const forbidden of ['DeepSeek V4.0 Flash', '给 DeepSeek 发送消息', 'IconAttachment', 'agent-send-button']) {
  if (aiPageSource.includes(forbidden)) failures.push(`project AI page contains unsupported chat control ${forbidden}`)
 }
@@ -140,7 +145,7 @@ const unsupportedControls = [
  },
  {
    file: 'src/pages/projects/project-statusbar.tsx',
-    forbidden: ['升级', '支付', 'billing-plan', '152GB', 'AI 空闲'],
+    forbidden: ['升级', '支付', 'billing-plan', '152GB', 'AI 空闲', '服务已连接', 'IconCheckCircle'],
  },
  {
    file: 'src/pages/projects/project-topbar.tsx',
--- a/apps/web_v1/scripts/visual-check.mjs
+++ b/apps/web_v1/scripts/visual-check.mjs
@@ -43,6 +43,12 @@ const projectPatchRequests = []
 const inboxAnalyzeRequests = []
 const inboxConfirmRequests = []
 const aiSessionRequests = []
+let delayedAIListsRemaining = 0
+let delayNextAICreate = false
+let delayedAICreateCompleted = false
+let failNextSecondProjectAIList = false
+let expectingSecondProjectAIListError = false
+let expectedSecondProjectAIListConsoleErrorCount = 0
 let expectingProjectPatchError = false
 let expectedProjectPatchConsoleErrorCount = 0
 let expectingInboxConfirmError = false
@@ -71,6 +77,13 @@ const controlledAISessions = [
 ]
 page.on('console', (message) => {
  if (message.type() !== 'error') return
+  if (
+    expectingSecondProjectAIListError &&
+    message.text().includes('Failed to load resource')
+  ) {
+    expectedSecondProjectAIListConsoleErrorCount += 1
+    return
+  }
  if (
    expectingProjectPatchError &&
    expectedProjectPatchConsoleErrorCount === 0 &&
@@ -188,6 +201,7 @@ const secondVisualCheckWorkspace = {
  },
  channels: [
    { id: 'second-overview', projectId: secondProjectId, type: 'overview', title: '概况', icon: 'home', count: 0, url: '', sortOrder: 0 },
+    { id: 'second-ai', projectId: secondProjectId, type: 'ai_sessions', title: 'AI 助手', icon: 'robot', count: 0, url: '', sortOrder: 1 },
  ],
  tags: [],
  recentSessions: [],
@@ -232,8 +246,11 @@ await page.route('http://localhost:9150/api/v1/**', async (route) => {
    return
  }
  if (url.pathname === `/api/v1/projects/${projectId}/ai-sessions` && method === 'GET') {
-    await new Promise((resolve) => setTimeout(resolve, 120))
-    await route.fulfill({ json: controlledAISessions })
+    const responseSnapshot = structuredClone(controlledAISessions)
+    const delay = delayedAIListsRemaining > 0 ? 700 : 120
+    if (delayedAIListsRemaining > 0) delayedAIListsRemaining -= 1
+    await new Promise((resolve) => setTimeout(resolve, delay))
+    await route.fulfill({ json: responseSnapshot }).catch(() => {})
    return
  }
  if (url.pathname === `/api/v1/projects/${projectId}/ai-sessions` && method === 'POST') {
@@ -248,8 +265,24 @@ await page.route('http://localhost:9150/api/v1/**', async (route) => {
      createdAt: '2026-07-21T03:10:00Z',
      updatedAt: '2026-07-21T03:10:00Z',
    }
+    if (delayNextAICreate) {
+      delayNextAICreate = false
+      await new Promise((resolve) => setTimeout(resolve, 700))
+      delayedAICreateCompleted = true
+    }
    controlledAISessions.unshift(created)
-    await route.fulfill({ status: 201, json: created })
+    await route.fulfill({ status: 201, json: created }).catch(() => {})
+    return
+  }
+  if (url.pathname === `/api/v1/projects/${secondProjectId}/ai-sessions` && method === 'GET') {
+    if (failNextSecondProjectAIList) {
+      await route.fulfill({
+        status: 500,
+        json: { error: { code: 'ai_list_failed', message: '新项目会话加载失败' } },
+      })
+      return
+    }
+    await route.fulfill({ json: [] })
    return
  }
  if (url.pathname === `/api/v1/inbox/${inboxItemId}/analyze` && method === 'POST') {
@@ -990,6 +1023,57 @@ for (const channel of [
      activeChannel,
    }
  }, channel))
+
+  if (channel.label === 'AI 助手') {
+    await page.locator('.channel-button', { hasText: '工作计划' }).click()
+    delayedAIListsRemaining = 2
+    await page.locator('.channel-button', { hasText: 'AI 助手' }).click()
+    const sameProjectRacePage = page.locator('.project-ai-page')
+    await sameProjectRacePage.locator('input').fill('乱序请求保留的新会话')
+    await sameProjectRacePage.locator('textarea').fill('POST 完成后，旧 GET 不得覆盖结果')
+    await sameProjectRacePage.getByRole('button', { name: '创建会话', exact: true }).click()
+    await sameProjectRacePage.getByText('乱序请求保留的新会话', { exact: true }).waitFor({ state: 'visible', timeout: 2000 })
+    await page.waitForTimeout(850)
+    if (await sameProjectRacePage.getByText('乱序请求保留的新会话', { exact: true }).count() !== 1) {
+      failures.push('a stale AI session GET must not overwrite a newer POST result in the same project')
+    }
+    delayedAIListsRemaining = 0
+
+    delayNextAICreate = true
+    await sameProjectRacePage.locator('input').fill('旧项目延迟会话')
+    await sameProjectRacePage.locator('textarea').fill('切换项目后不得回流')
+    await sameProjectRacePage.getByRole('button', { name: '创建会话', exact: true }).click()
+    failNextSecondProjectAIList = true
+    expectingSecondProjectAIListError = true
+    await page.locator('.project-button[title="并行项目"]').click()
+    await page.locator('.project-title h5').getByText('并行项目', { exact: true }).waitFor({ state: 'visible', timeout: 2000 })
+    await page.locator('.channel-button', { hasText: 'AI 助手' }).click()
+    const secondProjectAIPage = page.locator('.project-ai-page')
+    await secondProjectAIPage.waitFor({ state: 'visible', timeout: 2000 })
+    await page.waitForTimeout(500)
+    const secondProjectAIText = await secondProjectAIPage.textContent()
+    if (!secondProjectAIText?.includes('新项目会话加载失败')) {
+      failures.push(`the new project AI list failure must be visible, got ${JSON.stringify(secondProjectAIText)}`)
+    }
+    if (!secondProjectAIText?.includes('暂无 AI 会话')) {
+      failures.push(`the new project AI list failure must retain an empty list, got ${JSON.stringify(secondProjectAIText)}`)
+    }
+    await page.waitForTimeout(850)
+    if (!delayedAICreateCompleted) failures.push('the simulated old-project AI POST did not complete')
+    if (await secondProjectAIPage.getByText('旧项目延迟会话', { exact: true }).count() !== 0) {
+      failures.push('an old-project AI POST must not mutate the newly selected project')
+    }
+    if (await secondProjectAIPage.getByText('新项目会话加载失败', { exact: true }).count() !== 1) {
+      failures.push('an old-project AI POST completion must not clear the new project load error')
+    }
+    failNextSecondProjectAIList = false
+    expectingSecondProjectAIListError = false
+    if (expectedSecondProjectAIListConsoleErrorCount < 1) {
+      failures.push(`expected a simulated second-project AI list console error, got ${expectedSecondProjectAIListConsoleErrorCount}`)
+    }
+    await page.locator('.project-button[title="森林项目已更新"]').click()
+    await page.waitForTimeout(250)
+  }
 }

 const unsupportedControlMetrics = await page.evaluate(() => ({
@@ -1170,9 +1254,9 @@ for (const check of channelPageChecks) {
  }
 }

-if (aiSessionRequests.length !== 1) failures.push(`expected one controlled AI session request, got ${aiSessionRequests.length}`)
-if (aiSessionRequests.length === 1) {
-  const requestKeys = Object.keys(aiSessionRequests[0]).sort()
+if (aiSessionRequests.length !== 3) failures.push(`expected three controlled AI session requests, got ${aiSessionRequests.length}`)
+for (const request of aiSessionRequests) {
+  const requestKeys = Object.keys(request).sort()
  if (JSON.stringify(requestKeys) !== JSON.stringify(['context', 'title'])) {
    failures.push(`AI session create must send only title/context, got ${JSON.stringify(requestKeys)}`)
  }
@@ -1180,7 +1264,7 @@ if (aiSessionRequests.length === 1) {
 if (unsupportedControlMetrics.topbarLabels.some((label) => label?.includes('停靠'))) {
  failures.push(`topbar must not expose dock controls, got ${JSON.stringify(unsupportedControlMetrics.topbarLabels)}`)
 }
-if (/升级|支付|AI 空闲|GB 可用/.test(unsupportedControlMetrics.statusbarText)) {
+if (/升级|支付|AI 空闲|GB 可用|服务已连接/.test(unsupportedControlMetrics.statusbarText)) {
  failures.push(`statusbar contains unsupported product state: ${unsupportedControlMetrics.statusbarText}`)
 }
 if (!unsupportedControlMetrics.newChannelText.includes('暂未开放') || unsupportedControlMetrics.newChannelButtonCount !== 0) {
--- a/apps/web_v1/src/api/ai.ts
+++ b/apps/web_v1/src/api/ai.ts
@@ -15,16 +15,18 @@ export type CreateAISessionInput = {
  context: string
 }

-export async function listAISessions(session: ApiSession, projectId: string) {
+export async function listAISessions(session: ApiSession, projectId: string, signal?: AbortSignal) {
  return apiRequest<AISessionDTO[]>(`/api/v1/projects/${projectId}/ai-sessions`, {
    token: session.token,
+    signal,
  })
 }

-export async function createAISession(session: ApiSession, projectId: string, input: CreateAISessionInput) {
+export async function createAISession(session: ApiSession, projectId: string, input: CreateAISessionInput, signal?: AbortSignal) {
  return apiRequest<AISessionDTO>(`/api/v1/projects/${projectId}/ai-sessions`, {
    method: 'POST',
    token: session.token,
    body: input,
+    signal,
  })
 }
--- a/apps/web_v1/src/app/App.tsx
+++ b/apps/web_v1/src/app/App.tsx
@@ -44,14 +44,14 @@ function App() {
  const [searchResultPreview, setSearchResultPreview] = useState<SearchResultDTO | null>(null)
  const workspaceSearch = useWorkbenchSearch(session)

-  const handleListAISessions = useCallback((projectId: string) => {
+  const handleListAISessions = useCallback((projectId: string, signal?: AbortSignal) => {
    if (!session) return Promise.reject(new Error('未登录'))
-    return listAISessions(session, projectId)
+    return listAISessions(session, projectId, signal)
  }, [session])

-  const handleCreateAISession = useCallback((projectId: string, input: CreateAISessionInput) => {
+  const handleCreateAISession = useCallback((projectId: string, input: CreateAISessionInput, signal?: AbortSignal) => {
    if (!session) return Promise.reject(new Error('未登录'))
-    return createAISession(session, projectId, input)
+    return createAISession(session, projectId, input, signal)
  }, [session])

  const dark = theme === 'dark'
--- a/apps/web_v1/src/pages/projects/project-ai.tsx
+++ b/apps/web_v1/src/pages/projects/project-ai.tsx
@@ -1,4 +1,4 @@
-import { useEffect, useState } from 'react'
+import { useEffect, useRef, useState } from 'react'
 import { Alert, Button, Card, Empty, Input, Space, Spin, Tag, Typography } from '@arco-design/web-react'
 import { IconPlusCircle, IconRobot } from '@arco-design/web-react/icon'
 import type { AISessionDTO, CreateAISessionInput } from '../../api/ai'
@@ -14,8 +14,8 @@ export function ProjectAi({
 }: {
  activeWorkspace: ProjectWorkspace
  onSelectItem: (title: string) => void
-  onListSessions: (projectId: string) => Promise<AISessionDTO[]>
-  onCreateSession: (projectId: string, input: CreateAISessionInput) => Promise<AISessionDTO>
+  onListSessions: (projectId: string, signal?: AbortSignal) => Promise<AISessionDTO[]>
+  onCreateSession: (projectId: string, input: CreateAISessionInput, signal?: AbortSignal) => Promise<AISessionDTO>
 }) {
  const [sessions, setSessions] = useState<AISessionDTO[]>([])
  const [title, setTitle] = useState('')
@@ -24,23 +24,47 @@ export function ProjectAi({
  const [creating, setCreating] = useState(false)
  const [error, setError] = useState('')
  const projectId = activeWorkspace.project.id
+  const projectRef = useRef(projectId)
+  const generationRef = useRef(0)
+  const listControllerRef = useRef<AbortController | null>(null)
+  const createControllerRef = useRef<AbortController | null>(null)
+  projectRef.current = projectId

  useEffect(() => {
-    let current = true
+    const generation = ++generationRef.current
+    const controller = new AbortController()
+    listControllerRef.current?.abort()
+    createControllerRef.current?.abort()
+    listControllerRef.current = controller
+    createControllerRef.current = null
+    setSessions([])
+    setTitle('')
+    setContext('')
    setLoading(true)
+    setCreating(false)
    setError('')
-    void onListSessions(projectId)
+    const isCurrent = () => projectRef.current === projectId && generationRef.current === generation && !controller.signal.aborted
+
+    void onListSessions(projectId, controller.signal)
      .then((items) => {
-        if (current) setSessions(items)
+        if (isCurrent()) setSessions(items)
      })
      .catch((requestError: unknown) => {
-        if (current) setError(requestError instanceof Error ? requestError.message : 'AI 会话加载失败，请稍后重试')
+        if (isCurrent()) setError(requestError instanceof Error ? requestError.message : 'AI 会话加载失败，请稍后重试')
      })
      .finally(() => {
-        if (current) setLoading(false)
+        if (isCurrent()) {
+          setLoading(false)
+          listControllerRef.current = null
+        }
      })
+
    return () => {
-      current = false
+      generationRef.current += 1
+      controller.abort()
+      if (listControllerRef.current === controller) listControllerRef.current = null
+      createControllerRef.current?.abort()
+      createControllerRef.current = null
    }
  }, [onListSessions, projectId])

@@ -50,21 +74,35 @@ export function ProjectAi({
      setError('请输入 AI 会话标题')
      return
    }
+    const generation = ++generationRef.current
+    listControllerRef.current?.abort()
+    listControllerRef.current = null
+    createControllerRef.current?.abort()
+    const controller = new AbortController()
+    createControllerRef.current = controller
+    const isCurrent = () => projectRef.current === projectId && generationRef.current === generation && !controller.signal.aborted
+
+    setLoading(false)
    setCreating(true)
    setError('')
    try {
      const created = await onCreateSession(projectId, {
        title: trimmedTitle,
        context: context.trim(),
-      })
+      }, controller.signal)
+      if (!isCurrent()) return
      setSessions((current) => [created, ...current.filter((session) => session.id !== created.id)])
      setTitle('')
      setContext('')
      onSelectItem(created.title)
    } catch (requestError) {
+      if (!isCurrent()) return
      setError(requestError instanceof Error ? requestError.message : 'AI 会话创建失败，请稍后重试')
    } finally {
-      setCreating(false)
+      if (isCurrent()) {
+        setCreating(false)
+        createControllerRef.current = null
+      }
    }
  }

--- a/apps/web_v1/src/pages/projects/project-channel-page.tsx
+++ b/apps/web_v1/src/pages/projects/project-channel-page.tsx
@@ -40,8 +40,8 @@ export function ProjectChannelPage({
  onUpdateTask: (update: ProjectTaskUpdate) => void
  onAnalyzeInbox: (inboxId: string) => Promise<InboxSuggestionDTO[]>
  onConfirmInbox: (inboxId: string, suggestionIds: string[]) => Promise<InboxConfirmationOutcome>
-  onListAISessions: (projectId: string) => Promise<AISessionDTO[]>
-  onCreateAISession: (projectId: string, input: CreateAISessionInput) => Promise<AISessionDTO>
+  onListAISessions: (projectId: string, signal?: AbortSignal) => Promise<AISessionDTO[]>
+  onCreateAISession: (projectId: string, input: CreateAISessionInput, signal?: AbortSignal) => Promise<AISessionDTO>
 }) {
  switch (activeChannel) {
    case 'inbox':
@@ -49,7 +49,7 @@ export function ProjectChannelPage({
    case 'tasks':
      return <ProjectTasks activeWorkspace={activeWorkspace} activeTaskID={activeTaskID} onOpenTask={onOpenTask} onCloseTask={onCloseTask} onSelectItem={onSelectItem} onCreateTask={onCreateTask} onCreateProjectTag={onCreateProjectTag} onUpdateTask={onUpdateTask} />
    case 'ai':
-      return <ProjectAi activeWorkspace={activeWorkspace} onSelectItem={onSelectItem} onListSessions={onListAISessions} onCreateSession={onCreateAISession} />
+      return <ProjectAi key={activeWorkspace.project.id} activeWorkspace={activeWorkspace} onSelectItem={onSelectItem} onListSessions={onListAISessions} onCreateSession={onCreateAISession} />
    case 'notes':
      return <ProjectNotes activeWorkspace={activeWorkspace} onSelectItem={onSelectItem} onUploadSource={onUploadSource} />
    case 'cron':
--- a/apps/web_v1/src/pages/projects/project-statusbar.tsx
+++ b/apps/web_v1/src/pages/projects/project-statusbar.tsx
@@ -1,5 +1,4 @@
-import { Layout, Space } from '@arco-design/web-react'
-import { IconCheckCircle } from '@arco-design/web-react/icon'
+import { Layout } from '@arco-design/web-react'

 const { Footer } = Layout

@@ -10,10 +9,6 @@ export function ProjectStatusbar() {
        <span className="status-avatar" aria-hidden="true">森</span>
        <span className="status-name">已登录</span>
      </div>
-      <Space className="status-system" size={8}>
-        <IconCheckCircle />
-        <span>服务已连接</span>
-      </Space>
    </Footer>
  )
 }
--- a/apps/web_v1/src/pages/workspace-home.tsx
+++ b/apps/web_v1/src/pages/workspace-home.tsx
@@ -79,8 +79,8 @@ export function ProjectPage({
  onSelectSearchResult: (result: SearchResultDTO) => void
  onAnalyzeInbox: (inboxId: string) => Promise<InboxSuggestionDTO[]>
  onConfirmInbox: (inboxId: string, suggestionIds: string[]) => Promise<InboxConfirmationOutcome>
-  onListAISessions: (projectId: string) => Promise<AISessionDTO[]>
-  onCreateAISession: (projectId: string, input: CreateAISessionInput) => Promise<AISessionDTO>
+  onListAISessions: (projectId: string, signal?: AbortSignal) => Promise<AISessionDTO[]>
+  onCreateAISession: (projectId: string, input: CreateAISessionInput, signal?: AbortSignal) => Promise<AISessionDTO>
 }) {
  const isProject = activeView === 'project'
  const [navOpen, setNavOpen] = useState<'projects' | 'channels' | null>(null)
--- a/backend/internal/logic/ai/gateway.go
+++ b/backend/internal/logic/ai/gateway.go
@@ -18,6 +18,7 @@ import (
 type Gateway struct {
 	systemKey        string
 	encryptionSecret string
+	now              func() time.Time
 }

 type SelectedKey struct {
@@ -36,7 +37,7 @@ func NewGateway(systemKey string) *Gateway {
 }

 func NewGatewayWithSecret(systemKey string, encryptionSecret string) *Gateway {
-	return &Gateway{systemKey: systemKey, encryptionSecret: encryptionSecret}
+	return &Gateway{systemKey: systemKey, encryptionSecret: encryptionSecret, now: time.Now}
 }

 func (g *Gateway) SaveUserKey(userID uint, provider string, apiKey string) error {
@@ -55,11 +56,13 @@ func (g *Gateway) SelectKey(userID uint) (SelectedKey, error) {
 	var userKey models.SenlinAgentAIKey
 	err := models.DBService.Where("user_id = ?", userID).First(&userKey).Error
 	if err == nil {
+		selected := SelectedKey{Provider: userKey.Provider, KeyType: "user"}
 		apiKey, err := decryptAPIKey(userKey.EncryptedAPIKey, g.encryptionSecret)
 		if err != nil {
-			return SelectedKey{}, err
+			return selected, err
 		}
-		return SelectedKey{Provider: userKey.Provider, APIKey: apiKey, KeyType: "user"}, nil
+		selected.APIKey = apiKey
+		return selected, nil
 	}
 	if !errors.Is(err, gorm.ErrRecordNotFound) {
 		return SelectedKey{}, err
@@ -70,8 +73,11 @@ func (g *Gateway) SelectKey(userID uint) (SelectedKey, error) {
 	return SelectedKey{Provider: "openai", APIKey: g.systemKey, KeyType: "system"}, nil
 }

-func (g *Gateway) RecordCall(userID uint, provider string, usedKeyType string, action string, status string, errText string) error {
-	return models.DBService.Create(&models.SenlinAgentAICallLog{
+func (g *Gateway) RecordCall(database *gorm.DB, userID uint, provider string, usedKeyType string, action string, status string, errText string) error {
+	if database == nil {
+		database = models.DBService
+	}
+	return database.Create(&models.SenlinAgentAICallLog{
 		UserID:      userID,
 		Provider:    provider,
 		UsedKeyType: usedKeyType,
@@ -81,17 +87,34 @@ func (g *Gateway) RecordCall(userID uint, provider string, usedKeyType string, a
 	}).Error
 }

-func (g *Gateway) CheckRateLimit(userID uint, action string, limit int, window time.Duration) error {
+// ReserveRateLimit 以数据库单条 UPSERT 原子占用固定窗口配额。
+// 配额在 provider/key 选择前占用，后续缺 key 或 provider 失败同样计入该窗口的尝试次数。
+func (g *Gateway) ReserveRateLimit(userID uint, action string, limit int, window time.Duration) error {
 	if limit <= 0 {
 		return nil
 	}
-	var count int64
-	if err := models.DBService.Model(&models.SenlinAgentAICallLog{}).
-		Where("user_id = ? AND action = ? AND created_at >= ?", userID, action, time.Now().Add(-window)).
-		Count(&count).Error; err != nil {
-		return err
+	currentTime := time.Now().UTC()
+	if g.now != nil {
+		currentTime = g.now().UTC()
 	}
-	if count >= int64(limit) {
+	windowStart := currentTime.Truncate(window)
+	bucket := models.SenlinAgentAIRateBucket{
+		UserID: userID, Action: action, WindowStart: windowStart, Count: 1,
+	}
+	result := models.DBService.Clauses(clause.OnConflict{
+		Columns: []clause.Column{{Name: "user_id"}, {Name: "action"}, {Name: "window_start"}},
+		DoUpdates: clause.Assignments(map[string]any{
+			"count":      gorm.Expr("senlin_agent_ai_rate_buckets.count + 1"),
+			"updated_at": currentTime,
+		}),
+		Where: clause.Where{Exprs: []clause.Expression{
+			clause.Lt{Column: clause.Column{Table: "senlin_agent_ai_rate_buckets", Name: "count"}, Value: limit},
+		}},
+	}).Create(&bucket)
+	if result.Error != nil {
+		return result.Error
+	}
+	if result.RowsAffected == 0 {
 		return ErrAIRateLimited
 	}
 	return nil
--- a/backend/internal/logic/ai/gateway_test.go
+++ b/backend/internal/logic/ai/gateway_test.go
@@ -63,7 +63,7 @@ func TestRecordCallStoresAuditFields(t *testing.T) {
 	database := newTestDB(t)
 	gateway := NewGateway("system-key")

-	require.NoError(t, gateway.RecordCall(3, "openai", "system", "inbox_analyze", "failed", "rate limited"))
+	require.NoError(t, gateway.RecordCall(database, 3, "openai", "system", "inbox_analyze", "failed", "rate limited"))

 	var log models.SenlinAgentAICallLog
 	require.NoError(t, database.First(&log).Error)
@@ -75,12 +75,13 @@ func TestRecordCallStoresAuditFields(t *testing.T) {
 	require.Equal(t, "rate limited", log.Error)
 }

-func TestCheckRateLimitRejectsCallsOverWindow(t *testing.T) {
-	newTestDB(t)
+func TestReserveRateLimitRejectsCallsOverWindow(t *testing.T) {
+	database := newTestDB(t)
+	require.NoError(t, database.Create(&models.SenlinAgentUser{Email: "rate@example.com", DisplayName: "Rate", PasswordHash: "hash"}).Error)
 	gateway := NewGateway("system-key")
-	require.NoError(t, gateway.RecordCall(3, "openai", "system", "inbox_analyze", "succeeded", ""))
+	require.NoError(t, gateway.ReserveRateLimit(1, "inbox_analyze", 1, time.Hour))

-	err := gateway.CheckRateLimit(3, "inbox_analyze", 1, time.Hour)
+	err := gateway.ReserveRateLimit(1, "inbox_analyze", 1, time.Hour)

 	require.ErrorContains(t, err, "ai rate limit exceeded")
 }
--- a/backend/internal/logic/ai/handlers_test.go
+++ b/backend/internal/logic/ai/handlers_test.go
@@ -3,6 +3,7 @@ package ai
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
@@ -73,7 +74,7 @@ func TestCreateAISessionReturnsRateLimitBeforeMissingKeyAndAuditsFailure(t *test
 	project := createAIHandlerProject(t, database, owner.ID, "LIMITED")
 	gateway := NewGatewayWithSecret("", "test-encryption-secret")
 	for range aiSessionCreateLimit {
-		require.NoError(t, gateway.RecordCall(owner.ID, "openai", "system", "ai_session_create", "ready", ""))
+		require.NoError(t, gateway.ReserveRateLimit(owner.ID, aiSessionCreateAction, aiSessionCreateLimit, time.Hour))
 	}
 	router := aiHandlerTestRouter(owner.ID, gateway)
 	recorder := httptest.NewRecorder()
@@ -130,6 +131,69 @@ func TestCreateAISessionWithoutKeyReturnsAuditedErrorAndCreatesNoFormalObjects(t
 	require.Equal(t, "ai_session_create", call.Action)
 	require.Equal(t, "failed", call.Status)
 	require.Equal(t, "ai_key_missing", call.Error)
+	var bucket models.SenlinAgentAIRateBucket
+	require.NoError(t, database.Where("user_id = ? AND action = ?", owner.ID, aiSessionCreateAction).First(&bucket).Error)
+	require.Equal(t, 1, bucket.Count)
+}
+
+func TestCreateAISessionRollsBackSessionWhenReadyAuditWriteFails(t *testing.T) {
+	database := newAIHandlerTestDB(t)
+	owner := createAIHandlerUser(t, database, "audit-failure@example.com")
+	project := createAIHandlerProject(t, database, owner.ID, "AUDIT_FAILURE")
+	injectedError := errors.New("injected ready audit failure")
+	callbackName := "test:fail_ready_ai_audit"
+	require.NoError(t, database.Callback().Create().Before("gorm:create").Register(callbackName, func(tx *gorm.DB) {
+		call, ok := tx.Statement.Dest.(*models.SenlinAgentAICallLog)
+		if ok && call.Status == defaultSessionStatus {
+			tx.AddError(injectedError)
+		}
+	}))
+	t.Cleanup(func() { database.Callback().Create().Remove(callbackName) })
+	router := aiHandlerTestRouter(owner.ID, NewGatewayWithSecret("system-key", "test-encryption-secret"))
+	recorder := httptest.NewRecorder()
+
+	router.ServeHTTP(recorder, authenticatedAIRequest(t, http.MethodPost, "/api/v1/projects/"+project.Identity+"/ai-sessions", map[string]any{
+		"title": "必须回滚", "context": "成功审计失败时不能残留会话",
+	}))
+
+	require.Equal(t, http.StatusInternalServerError, recorder.Code)
+	var sessionCount int64
+	require.NoError(t, database.Model(&models.SenlinAgentAISession{}).Where("project_id = ?", project.ID).Count(&sessionCount).Error)
+	require.Zero(t, sessionCount)
+	var calls []models.SenlinAgentAICallLog
+	require.NoError(t, database.Where("user_id = ? AND action = ?", owner.ID, aiSessionCreateAction).Find(&calls).Error)
+	require.Len(t, calls, 1)
+	require.Equal(t, "openai", calls[0].Provider)
+	require.Equal(t, "system", calls[0].UsedKeyType)
+	require.Equal(t, "failed", calls[0].Status)
+	require.Equal(t, "audit_write_failed", calls[0].Error)
+}
+
+func TestCreateAISessionAuditsKnownProviderMetadataWhenUserKeyDecryptFails(t *testing.T) {
+	database := newAIHandlerTestDB(t)
+	owner := createAIHandlerUser(t, database, "decrypt-failure@example.com")
+	project := createAIHandlerProject(t, database, owner.ID, "DECRYPT_FAILURE")
+	require.NoError(t, database.Create(&models.SenlinAgentAIKey{
+		UserID: owner.ID, Provider: "deepseek", EncryptedAPIKey: "v1:not-valid-base64",
+	}).Error)
+	router := aiHandlerTestRouter(owner.ID, NewGatewayWithSecret("system-key", "test-encryption-secret"))
+	recorder := httptest.NewRecorder()
+
+	router.ServeHTTP(recorder, authenticatedAIRequest(t, http.MethodPost, "/api/v1/projects/"+project.Identity+"/ai-sessions", map[string]any{
+		"title": "解密失败", "context": "审计不得丢失已知元数据",
+	}))
+
+	require.Equal(t, http.StatusInternalServerError, recorder.Code)
+	var call models.SenlinAgentAICallLog
+	require.NoError(t, database.Where("user_id = ? AND action = ?", owner.ID, aiSessionCreateAction).First(&call).Error)
+	require.Equal(t, "deepseek", call.Provider)
+	require.Equal(t, "user", call.UsedKeyType)
+	require.Equal(t, "failed", call.Status)
+	require.Equal(t, "provider_selection_failed", call.Error)
+	require.NotContains(t, call.Error, "not-valid-base64")
+	var sessionCount int64
+	require.NoError(t, database.Model(&models.SenlinAgentAISession{}).Count(&sessionCount).Error)
+	require.Zero(t, sessionCount)
 }

 func TestCreateAISessionReturnsIdentityDTOAndCompleteAuditWithoutAutomaticObjectIDs(t *testing.T) {
@@ -205,7 +269,7 @@ type recordingSessionGateway struct {
 	selectErr error
 }

-func (g *recordingSessionGateway) CheckRateLimit(uint, string, int, time.Duration) error {
+func (g *recordingSessionGateway) ReserveRateLimit(uint, string, int, time.Duration) error {
 	g.steps = append(g.steps, "rate")
 	return g.rateErr
 }
@@ -215,7 +279,7 @@ func (g *recordingSessionGateway) SelectKey(uint) (SelectedKey, error) {
 	return g.selected, g.selectErr
 }

-func (g *recordingSessionGateway) RecordCall(uint, string, string, string, string, string) error {
+func (g *recordingSessionGateway) RecordCall(*gorm.DB, uint, string, string, string, string, string) error {
 	g.steps = append(g.steps, "record")
 	return nil
 }
--- a/backend/internal/logic/ai/rate_limit_postgres_test.go
+++ b/backend/internal/logic/ai/rate_limit_postgres_test.go
@@ -0,0 +1,72 @@
+package ai
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
+	"senlinai-agent/backend/internal/models"
+)
+
+func TestPostgresReserveRateLimitIsAtomicAcrossConcurrentConnections(t *testing.T) {
+	dsn := os.Getenv("DATABASE_URL")
+	if dsn == "" {
+		t.Skip("DATABASE_URL is not configured; skipping PostgreSQL AI rate reservation test")
+	}
+	database, err := gorm.Open(postgres.Open(dsn), &gorm.Config{TranslateError: true, Logger: logger.Default.LogMode(logger.Silent)})
+	require.NoError(t, err)
+	require.NoError(t, models.AutoMigrate(database))
+	models.DBService = database
+	suffix := fmt.Sprint(time.Now().UnixNano())
+	user := createAIRateTestUser(t, database, "postgres-rate-"+suffix+"@example.com")
+	action := "postgres_concurrent_" + suffix
+	t.Cleanup(func() {
+		database.Where("user_id = ?", user.ID).Delete(&models.SenlinAgentAIRateBucket{})
+		database.Delete(&user)
+	})
+	gateway := NewGatewayWithSecret("system-key", "test-encryption-secret")
+	const (
+		limit    = 9
+		attempts = 48
+	)
+
+	start := make(chan struct{})
+	results := make(chan error, attempts)
+	var wait sync.WaitGroup
+	for range attempts {
+		wait.Add(1)
+		go func() {
+			defer wait.Done()
+			<-start
+			results <- gateway.ReserveRateLimit(user.ID, action, limit, time.Hour)
+		}()
+	}
+	close(start)
+	wait.Wait()
+	close(results)
+
+	allowed := 0
+	limited := 0
+	for err := range results {
+		switch {
+		case err == nil:
+			allowed++
+		case errors.Is(err, ErrAIRateLimited):
+			limited++
+		default:
+			require.NoError(t, err)
+		}
+	}
+	require.Equal(t, limit, allowed)
+	require.Equal(t, attempts-limit, limited)
+	var bucket models.SenlinAgentAIRateBucket
+	require.NoError(t, database.Where("user_id = ? AND action = ?", user.ID, action).First(&bucket).Error)
+	require.Equal(t, limit, bucket.Count)
+}
--- a/backend/internal/logic/ai/rate_limit_test.go
+++ b/backend/internal/logic/ai/rate_limit_test.go
@@ -0,0 +1,101 @@
+package ai
+
+import (
+	"errors"
+	"fmt"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/glebarez/sqlite"
+	"github.com/stretchr/testify/require"
+	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
+	"senlinai-agent/backend/internal/models"
+)
+
+func TestReserveRateLimitIsAtomicUnderConcurrentSQLiteRequests(t *testing.T) {
+	database := newConcurrentAIRateTestDB(t)
+	user := createAIRateTestUser(t, database, "sqlite-rate@example.com")
+	gateway := NewGatewayWithSecret("system-key", "test-encryption-secret")
+	const (
+		limit    = 7
+		attempts = 40
+	)
+
+	start := make(chan struct{})
+	results := make(chan error, attempts)
+	var wait sync.WaitGroup
+	for range attempts {
+		wait.Add(1)
+		go func() {
+			defer wait.Done()
+			<-start
+			results <- gateway.ReserveRateLimit(user.ID, "concurrent_session_create", limit, time.Hour)
+		}()
+	}
+	close(start)
+	wait.Wait()
+	close(results)
+
+	allowed := 0
+	limited := 0
+	for err := range results {
+		switch {
+		case err == nil:
+			allowed++
+		case errors.Is(err, ErrAIRateLimited):
+			limited++
+		default:
+			require.NoError(t, err)
+		}
+	}
+	require.Equal(t, limit, allowed)
+	require.Equal(t, attempts-limit, limited)
+	var buckets []models.SenlinAgentAIRateBucket
+	require.NoError(t, database.Where("user_id = ? AND action = ?", user.ID, "concurrent_session_create").Find(&buckets).Error)
+	require.Len(t, buckets, 1)
+	require.Equal(t, limit, buckets[0].Count)
+}
+
+func TestReserveRateLimitUsesFixedWindowsAndCountsFailedAttempts(t *testing.T) {
+	database := newConcurrentAIRateTestDB(t)
+	user := createAIRateTestUser(t, database, "window-rate@example.com")
+	gateway := NewGatewayWithSecret("system-key", "test-encryption-secret")
+	current := time.Date(2026, 7, 21, 10, 15, 0, 0, time.UTC)
+	gateway.now = func() time.Time { return current }
+
+	require.NoError(t, gateway.ReserveRateLimit(user.ID, "windowed_session_create", 2, time.Hour))
+	require.NoError(t, gateway.ReserveRateLimit(user.ID, "windowed_session_create", 2, time.Hour))
+	require.ErrorIs(t, gateway.ReserveRateLimit(user.ID, "windowed_session_create", 2, time.Hour), ErrAIRateLimited)
+
+	current = current.Add(time.Hour)
+	require.NoError(t, gateway.ReserveRateLimit(user.ID, "windowed_session_create", 2, time.Hour))
+	var buckets []models.SenlinAgentAIRateBucket
+	require.NoError(t, database.Where("user_id = ? AND action = ?", user.ID, "windowed_session_create").Order("window_start asc").Find(&buckets).Error)
+	require.Len(t, buckets, 2)
+	require.Equal(t, []int{2, 1}, []int{buckets[0].Count, buckets[1].Count})
+}
+
+func newConcurrentAIRateTestDB(t *testing.T) *gorm.DB {
+	t.Helper()
+	databasePath := filepath.ToSlash(filepath.Join(t.TempDir(), "ai-rate.db"))
+	dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(10000)&_pragma=journal_mode(WAL)", databasePath)
+	database, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+	require.NoError(t, err)
+	sqlDatabase, err := database.DB()
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, sqlDatabase.Close()) })
+	sqlDatabase.SetMaxOpenConns(20)
+	require.NoError(t, models.AutoMigrate(database))
+	models.DBService = database
+	return database
+}
+
+func createAIRateTestUser(t *testing.T, database *gorm.DB, email string) models.SenlinAgentUser {
+	t.Helper()
+	user := models.SenlinAgentUser{Email: email, DisplayName: email, PasswordHash: "hash"}
+	require.NoError(t, database.Create(&user).Error)
+	return user
+}
--- a/backend/internal/logic/ai/sessions.go
+++ b/backend/internal/logic/ai/sessions.go
@@ -6,6 +6,7 @@ import (
 	"strings"
 	"time"

+	"gorm.io/gorm"
 	"senlinai-agent/backend/internal/logic/projects"
 	"senlinai-agent/backend/internal/models"
 )
@@ -21,9 +22,9 @@ var (
 )

 type sessionGateway interface {
-	CheckRateLimit(userID uint, action string, limit int, window time.Duration) error
+	ReserveRateLimit(userID uint, action string, limit int, window time.Duration) error
 	SelectKey(userID uint) (SelectedKey, error)
-	RecordCall(userID uint, provider string, usedKeyType string, action string, status string, errText string) error
+	RecordCall(database *gorm.DB, userID uint, provider string, usedKeyType string, action string, status string, errText string) error
 }

 // SessionService 只管理项目内普通 AI 会话；它不会把上下文自动转换为任务、笔记或资料。
@@ -66,14 +67,14 @@ func (s *SessionService) Create(userID uint, projectIdentity, title, context str
 		return nil, errors.New("ai gateway is required")
 	}

-	if err := s.gateway.CheckRateLimit(userID, aiSessionCreateAction, aiSessionCreateLimit, time.Hour); err != nil {
+	if err := s.gateway.ReserveRateLimit(userID, aiSessionCreateAction, aiSessionCreateLimit, time.Hour); err != nil {
 		if errors.Is(err, ErrAIRateLimited) {
-			if auditErr := s.gateway.RecordCall(userID, "none", "none", aiSessionCreateAction, "failed", "ai_rate_limited"); auditErr != nil {
+			if auditErr := s.gateway.RecordCall(models.DBService, userID, "none", "none", aiSessionCreateAction, "failed", "ai_rate_limited"); auditErr != nil {
 				return nil, fmt.Errorf("record ai rate limit failure: %w", auditErr)
 			}
 			return nil, ErrAIRateLimited
 		}
-		if auditErr := s.gateway.RecordCall(userID, "none", "none", aiSessionCreateAction, "failed", "rate_limit_check_failed"); auditErr != nil {
+		if auditErr := s.gateway.RecordCall(models.DBService, userID, "none", "none", aiSessionCreateAction, "failed", "rate_limit_reservation_failed"); auditErr != nil {
 			return nil, fmt.Errorf("check rate limit: %v; record failure: %w", err, auditErr)
 		}
 		return nil, err
@@ -85,7 +86,8 @@ func (s *SessionService) Create(userID uint, projectIdentity, title, context str
 		if errors.Is(err, ErrAIKeyMissing) {
 			code = "ai_key_missing"
 		}
-		if auditErr := s.gateway.RecordCall(userID, "none", "none", aiSessionCreateAction, "failed", code); auditErr != nil {
+		provider, keyType := selectedAuditMetadata(selected)
+		if auditErr := s.gateway.RecordCall(models.DBService, userID, provider, keyType, aiSessionCreateAction, "failed", code); auditErr != nil {
 			return nil, fmt.Errorf("select ai key: %v; record failure: %w", err, auditErr)
 		}
 		return nil, err
@@ -98,19 +100,40 @@ func (s *SessionService) Create(userID uint, projectIdentity, title, context str
 		Context:   context,
 		Status:    defaultSessionStatus,
 	}
-	if err := models.DBService.Create(&session).Error; err != nil {
-		if auditErr := s.gateway.RecordCall(userID, selected.Provider, selected.KeyType, aiSessionCreateAction, "failed", "session_create_failed"); auditErr != nil {
-			return nil, fmt.Errorf("create ai session: %v; record failure: %w", err, auditErr)
+	failureCode := "session_create_failed"
+	err = models.DBService.Transaction(func(tx *gorm.DB) error {
+		if err := tx.Create(&session).Error; err != nil {
+			return err
+		}
+		// ready 只表示会话入口已建立，不表示 provider 已回复或任何业务对象已创建。
+		failureCode = "audit_write_failed"
+		if err := s.gateway.RecordCall(tx, userID, selected.Provider, selected.KeyType, aiSessionCreateAction, defaultSessionStatus, ""); err != nil {
+			return err
+		}
+		failureCode = "session_transaction_failed"
+		return nil
+	})
+	if err != nil {
+		if auditErr := s.gateway.RecordCall(models.DBService, userID, selected.Provider, selected.KeyType, aiSessionCreateAction, "failed", failureCode); auditErr != nil {
+			return nil, fmt.Errorf("create ai session transaction: %v; record failure: %w", err, auditErr)
 		}
 		return nil, err
 	}
-	// ready 只表示会话入口已建立，不表示 provider 已回复或任何业务对象已创建。
-	if err := s.gateway.RecordCall(userID, selected.Provider, selected.KeyType, aiSessionCreateAction, defaultSessionStatus, ""); err != nil {
-		return nil, fmt.Errorf("record ai session creation: %w", err)
-	}
 	return &session, nil
 }

+func selectedAuditMetadata(selected SelectedKey) (string, string) {
+	provider := strings.TrimSpace(selected.Provider)
+	keyType := strings.TrimSpace(selected.KeyType)
+	if provider == "" {
+		provider = "none"
+	}
+	if keyType == "" {
+		keyType = "none"
+	}
+	return provider, keyType
+}
+
 func aiSessionStatus(session models.SenlinAgentAISession) string {
 	if status := strings.TrimSpace(session.Status); status != "" {
 		return status
--- a/backend/internal/models/ai_rate_bucket.go
+++ b/backend/internal/models/ai_rate_bucket.go
@@ -0,0 +1,19 @@
+package models
+
+import "time"
+
+// SenlinAgentAIRateBucket 保存用户在固定窗口内已占用的 AI 请求配额。
+// 复合唯一键让单条 UPSERT 在数据库层完成跨进程并发仲裁。
+type SenlinAgentAIRateBucket struct {
+	ID          uint      `gorm:"primaryKey"`
+	UserID      uint      `gorm:"not null;uniqueIndex:uidx_senlin_agent_ai_rate_bucket,priority:1"`
+	Action      string    `gorm:"size:100;not null;uniqueIndex:uidx_senlin_agent_ai_rate_bucket,priority:2"`
+	WindowStart time.Time `gorm:"not null;uniqueIndex:uidx_senlin_agent_ai_rate_bucket,priority:3"`
+	Count       int       `gorm:"not null"`
+	CreatedAt   time.Time
+	UpdatedAt   time.Time
+}
+
+func (SenlinAgentAIRateBucket) TableName() string {
+	return "senlin_agent_ai_rate_buckets"
+}
--- a/backend/internal/models/new.go
+++ b/backend/internal/models/new.go
@@ -38,6 +38,7 @@ func AutoMigrate(database *gorm.DB) error {
 		&SenlinAgentProjectEvent{},
 		&SenlinAgentAIKey{},
 		&SenlinAgentAICallLog{},
+		&SenlinAgentAIRateBucket{},
 		&SenlinAgentTaskShare{},
 	)
 }