fix: fix the wrong cache for second call of agent tools

[xinquiry]

变更内容

• 新功能
• 修复 Bug
• 增强重构
• 其他（请描述）

简要描述本次 PR 的主要变更内容。

相关 Issue

请关联相关 Issue（如有）：#编号

检查清单

默认已勾选，如不满足，请检查。

• 已在本地测试通过
• 已补充/更新相关文档
• 已添加测试用例
• 代码风格已经过 pre-commit 钩子检查

其他说明

如有特殊说明或注意事项，请补充。

Summary by Sourcery

增强：

• 优化 Celery worker 数据库引擎缓存机制，将引擎缓存的键从仅按进程区分改为同时按进程和事件循环区分。

Original summary in English

Summary by Sourcery

Enhancements:

• Refine Celery worker database engine caching to key engines by process and event loop rather than process only.

- 确保 Celery worker/tool 的数据库引擎在每个进程和事件循环中分别进行缓存，避免在不同事件循环之间复用连接，从而防止导致不正确的行为。

Original summary in English

Summary by Sourcery

增强：

• 优化 Celery worker 数据库引擎缓存机制，将引擎缓存的键从仅按进程区分改为同时按进程和事件循环区分。

Original summary in English

Summary by Sourcery

Enhancements:

• Refine Celery worker database engine caching to key engines by process and event loop rather than process only.

[sourcery-ai]

审阅者指南（在小型 PR 上折叠）

审阅者指南

调整 Celery worker/工具的数据库会话处理方式，使引擎按（进程、事件循环）进行缓存，而不是仅按进程缓存，从而防止在不同事件循环之间重用连接，并修复后续工具调用中的错误行为。

Celery worker 任务数据库会话创建与缓存查找的时序图

sequenceDiagram
    actor CeleryWorker
    participant EventLoop
    participant get_task_db_session
    participant _worker_engines
    participant Database

    CeleryWorker->>EventLoop: Run async task
    EventLoop->>get_task_db_session: async with get_task_db_session()
    get_task_db_session->>get_task_db_session: pid = os.getpid()
    get_task_db_session->>EventLoop: get_running_loop()
    EventLoop-->>get_task_db_session: loop instance
    get_task_db_session->>get_task_db_session: loop_id = id(loop)
    get_task_db_session->>get_task_db_session: cache_key = (pid, loop_id)

    alt cache miss for cache_key
        get_task_db_session->>_worker_engines: find old_keys for pid and loop_id != current
        _worker_engines-->>get_task_db_session: old_keys list
        get_task_db_session->>_worker_engines: delete old_keys
        get_task_db_session->>Database: create_async_engine(ASYNC_DATABASE_URL)
        Database-->>get_task_db_session: engine
        get_task_db_session->>_worker_engines: store async_sessionmaker with cache_key
    else cache hit for cache_key
        get_task_db_session->>_worker_engines: reuse async_sessionmaker
    end

    get_task_db_session->>Database: open AsyncSession via async_sessionmaker
    Database-->>get_task_db_session: AsyncSession
    get_task_db_session-->>EventLoop: yield AsyncSession
    EventLoop-->>CeleryWorker: session usable in task

Loading

数据库连接模块缓存结构的类图

classDiagram
    class ConnectionModule {
        ASYNC_DATABASE_URL: str
        _worker_engines: dict[tuple[int,int], async_sessionmaker]
        +get_task_db_session() AsyncContextManager[AsyncSession]
        +get_session() AsyncGenerator[AsyncSession, None]
    }

    class AsyncEngine {
        +dispose() None
    }

    class AsyncSessionMaker {
        +__call__() AsyncSession
    }

    class AsyncSession {
        +execute(statement) Any
        +commit() None
        +close() None
    }

    ConnectionModule "*" o-- AsyncEngine : creates
    ConnectionModule "*" o-- AsyncSessionMaker : caches per(pid,loop)
    AsyncSessionMaker "*" --> AsyncSession : produces

Loading

文件级变更

变更	详情	文件
将 worker 数据库引擎缓存键从仅使用进程 ID 改为使用进程 ID 和事件循环的组合，以避免跨事件循环重用连接。	将 worker 引擎缓存结构从以进程 ID 为键的字典更新为以（进程 ID，事件循环 ID）为键的字典。 通过 asyncio.get_running_loop() 计算当前事件循环 ID，并与进程 ID 组合为用于引擎重用的缓存键。 在创建新引擎之前，清理属于当前进程但绑定到不同事件循环的已缓存引擎，确保只保留当前事件循环对应的缓存。 调整异步会话获取逻辑，在检索 async_sessionmaker 时使用新的（进程、事件循环）缓存键。	service/app/infra/database/connection.py

---

技巧与命令

与 Sourcery 交互

• 触发新审阅： 在 pull request 上评论 @sourcery-ai review。
• 继续讨论： 直接回复 Sourcery 的审阅评论。
• 从审阅评论生成 GitHub issue： 通过回复某条审阅评论，要求 Sourcery 从该评论创建一个 issue。你也可以在审阅评论下回复 @sourcery-ai issue，从该评论创建一个 issue。
• 生成 pull request 标题： 在 pull request 标题的任意位置写上 @sourcery-ai，即可随时生成标题。你也可以在 pull request 上评论 @sourcery-ai title 来（重新）生成标题。
• 生成 pull request 摘要： 在 pull request 描述正文任意位置写上 @sourcery-ai summary，即可在你希望的位置随时生成 PR 摘要。你也可以在 pull request 上评论 @sourcery-ai summary 来（重新）生成摘要。
• 生成审阅者指南： 在 pull request 上评论 @sourcery-ai guide，即可随时（重新）生成审阅者指南。
• 一次性解决所有 Sourcery 评论： 在 pull request 上评论 @sourcery-ai resolve，即可将所有 Sourcery 评论标记为已解决。如果你已经处理完所有评论且不想再看到它们，这会很有用。
• 一次性忽略所有 Sourcery 审阅： 在 pull request 上评论 @sourcery-ai dismiss，即可忽略所有现有的 Sourcery 审阅。如果你想从一次全新的审阅开始，这尤其有用——别忘了再次评论 @sourcery-ai review 来触发新的审阅！

自定义你的体验

访问你的 控制面板 以：

• 启用或禁用审阅功能，例如 Sourcery 自动生成的 pull request 摘要、审阅者指南等。
• 更改审阅语言。
• 添加、移除或编辑自定义审阅指令。
• 调整其他审阅设置。

获取帮助

• 联系支持团队 以获取帮助或提供反馈。
• 访问我们的文档，获取详细指南和信息。
• 通过关注 X/Twitter、LinkedIn 或 GitHub 与 Sourcery 团队保持联系。

Original review guide in English

Reviewer's guide (collapsed on small PRs)

Reviewer's Guide

Adjusts Celery worker/tool database session handling so engines are cached per (process, event loop) instead of per process, preventing reuse of connections across different event loops and fixing incorrect behavior on subsequent tool calls.

Sequence diagram for Celery worker task DB session creation and cache lookup

sequenceDiagram
    actor CeleryWorker
    participant EventLoop
    participant get_task_db_session
    participant _worker_engines
    participant Database

    CeleryWorker->>EventLoop: Run async task
    EventLoop->>get_task_db_session: async with get_task_db_session()
    get_task_db_session->>get_task_db_session: pid = os.getpid()
    get_task_db_session->>EventLoop: get_running_loop()
    EventLoop-->>get_task_db_session: loop instance
    get_task_db_session->>get_task_db_session: loop_id = id(loop)
    get_task_db_session->>get_task_db_session: cache_key = (pid, loop_id)

    alt cache miss for cache_key
        get_task_db_session->>_worker_engines: find old_keys for pid and loop_id != current
        _worker_engines-->>get_task_db_session: old_keys list
        get_task_db_session->>_worker_engines: delete old_keys
        get_task_db_session->>Database: create_async_engine(ASYNC_DATABASE_URL)
        Database-->>get_task_db_session: engine
        get_task_db_session->>_worker_engines: store async_sessionmaker with cache_key
    else cache hit for cache_key
        get_task_db_session->>_worker_engines: reuse async_sessionmaker
    end

    get_task_db_session->>Database: open AsyncSession via async_sessionmaker
    Database-->>get_task_db_session: AsyncSession
    get_task_db_session-->>EventLoop: yield AsyncSession
    EventLoop-->>CeleryWorker: session usable in task

Loading

Class diagram for database connection module cache structures

classDiagram
    class ConnectionModule {
        ASYNC_DATABASE_URL: str
        _worker_engines: dict[tuple[int,int], async_sessionmaker]
        +get_task_db_session() AsyncContextManager[AsyncSession]
        +get_session() AsyncGenerator[AsyncSession, None]
    }

    class AsyncEngine {
        +dispose() None
    }

    class AsyncSessionMaker {
        +__call__() AsyncSession
    }

    class AsyncSession {
        +execute(statement) Any
        +commit() None
        +close() None
    }

    ConnectionModule "*" o-- AsyncEngine : creates
    ConnectionModule "*" o-- AsyncSessionMaker : caches per(pid,loop)
    AsyncSessionMaker "*" --> AsyncSession : produces

Loading

File-Level Changes

Change	Details	Files
Change worker database engine cache to be keyed by both process ID and event loop to avoid cross-loop connection reuse.	Update worker engine cache structure from a dict keyed by process ID to a dict keyed by (process ID, event loop ID). Compute the current event loop ID via asyncio.get_running_loop() and combine with process ID into a cache key for engine reuse. Purge cached engines belonging to the current process but tied to different event loops before creating a new engine, ensuring only per-loop cache remains. Adjust the async session acquisition to use the new (process, loop) cache key when retrieving the async_sessionmaker.	service/app/infra/database/connection.py

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sourcery-ai]

Hey - 我在这里给出了一些总体反馈：

• 在清理旧事件循环所对应的 engine 时，你目前只是从 _worker_engines 中删除条目，却没有释放底层的 engine，这可能会导致连接泄漏；建议在移除之前调用 await task_engine.dispose()（或以其他方式关闭）。
• 使用 id(asyncio.get_running_loop()) 作为循环标识在实践中是可行的，但依赖于 CPython 的对象标识语义；如果你需要在时间维度上更强的保证，可以考虑使用单调递增的循环 token 或显式的循环名称来代替 id()。
• 对全局 _worker_engines 字典的访问目前是无同步保护的；如果 Celery 未来被配置为在单个进程中使用线程或多个事件循环，你可能需要用锁来保护对该字典的修改，以避免多个任务并发初始化或清理 engine 时产生竞争条件。

给 AI 代理的提示

Please address the comments from this code review:

## Overall Comments
- 在清理旧事件循环所对应的 engine 时，你目前只是从 `_worker_engines` 中删除条目，却没有释放底层的 engine，这可能会导致连接泄漏；建议在移除之前调用 `await task_engine.dispose()`（或以其他方式关闭）。
- 使用 `id(asyncio.get_running_loop())` 作为循环标识在实践中是可行的，但依赖于 CPython 的对象标识语义；如果你需要在时间维度上更强的保证，可以考虑使用单调递增的循环 token 或显式的循环名称来代替 `id()`。
- 对全局 `_worker_engines` 字典的访问目前是无同步保护的；如果 Celery 未来被配置为在单个进程中使用线程或多个事件循环，你可能需要用锁来保护对该字典的修改，以避免多个任务并发初始化或清理 engine 时产生竞争条件。

---

Sourcery 对开源项目免费——如果你觉得我们的评审有帮助，欢迎分享 ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据你的反馈改进后续的评审。}

Original comment in English

Hey - I've left some high level feedback:

• When cleaning up engines for old event loops you currently just delete the entry from _worker_engines without disposing the underlying engine, which can leak connections; consider calling await task_engine.dispose() (or otherwise closing) before removing them.
• Using id(asyncio.get_running_loop()) as the loop identifier works in practice but relies on CPython object identity semantics; if you need stronger guarantees over time, consider using a monotonically increasing loop token or explicit loop name instead of id().
• Access to the global _worker_engines dict is unsynchronized; if Celery is ever configured to use threads or multiple event loops in a single process, you may want to guard modifications with a lock to avoid race conditions when multiple tasks initialize or clean up engines concurrently.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- When cleaning up engines for old event loops you currently just delete the entry from `_worker_engines` without disposing the underlying engine, which can leak connections; consider calling `await task_engine.dispose()` (or otherwise closing) before removing them.
- Using `id(asyncio.get_running_loop())` as the loop identifier works in practice but relies on CPython object identity semantics; if you need stronger guarantees over time, consider using a monotonically increasing loop token or explicit loop name instead of `id()`.
- Access to the global `_worker_engines` dict is unsynchronized; if Celery is ever configured to use threads or multiple event loops in a single process, you may want to guard modifications with a lock to avoid race conditions when multiple tasks initialize or clean up engines concurrently.

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}