Repository navigation
Feature/advanced memory service - #334
CongkeChen wants to merge 6 commits into
Conversation
Please enter the commit message for your changes. Lines starting
| from trpc_agent_sdk.context import AgentContext | ||
| from trpc_agent_sdk.sessions import Session | ||
|
|
||
| if TYPE_CHECKING: |
| and calls ``save_memory``. Automatically storing every Session here | ||
| would mix transient conversation history with long-term memory. | ||
| """ | ||
| return None |
There was a problem hiding this comment.
因为本来这个方法不需要store_session逻辑，长期记忆只靠模型主动通过工具进行存储
|
|
||
| Advanced long-term memory is intentionally accessed through its | ||
| ``save_memory``, ``read_memory``, and ``list_memory_index`` tools. | ||
| """ |
There was a problem hiding this comment.
这两处只是为了满足标准 MemoryServiceABC 生命周期接口
| self._context_messages = context_messages | ||
|
|
||
| @property | ||
| def session_config(self): |
| session_service = memory_service.bind(agent, session_service) | ||
| elif isinstance(session_service, AdvancedMemorySessionService): | ||
| session_service = session_service.bind(agent) | ||
| if memory_service is not None: |
There was a problem hiding this comment.
这里最好不要修改，这个 runner 是运行流程不应该和具体的类绑定起来
AI Code Review审查结论不通过 本次审查覆盖 base_commit f05797d 至 head_commit 3c0f340 共 109 个变更文件（+6118/-6344），核心是 advanced memory service 独立化与 session compaction 抽象化（sessions/compact/{default,advanced}、CompactSummarizerABC/ManagerABC、SessionMemoryExtractor、update_session_state 三后端增量实现、compact_events）。计划符合性：advanced memory（save/read/index 工具、长时记忆索引注入、跨会话持久化）与 advanced compaction（BEFORE_MODEL 自动压缩、会话记忆提取、circuit-breaker）均以可组装组件加示例（examples/memory_service_with_advanced_memory、examples/session_summarizer_with_advanced）提供；但压缩链路在 SDK 内无默认安装路径，需用户显式手工接线。主要风险：token 模式 usage-baseline 的指纹记录与读取断链（写 session.state、读 event metadata，保护永不生效），导致上下文预算基于过期或误配 usage；default summarizer 的 conversation_count 复位语义从探测即复位变为摘要成功才复位，LLM 摘要失败时每轮重复触发；usage 统计把 output tokens 计入 context 基线造成高估；会话锁与状态表只增不减造成进程内缓慢增长。测试充分性：tests/sessions/compact/（7 个端到端用例）覆盖禁用回路、manager 装配、end-of-turn 压缩归档、session memory 落 state、tool-result 预算、协调锁；test_token_budget.py 覆盖 usage-baseline 匹配与指纹不匹配回退；但均未覆盖指纹从请求转移到响应事件的完整链路（该链路实际不存在）、conversation_count 失败重试语义、锁表生命周期。门禁结论：存在 1 个 SEVERE（token 模式基线决策错误）与 1 个 MODERATE（conversation_count 重试风暴），其余为 LOW；建议修复后合入。 发现的问题严重
问题： TokenContextTracker.record_request_context 只把请求指纹写入 session.state["advanced_memory_pending_request_context_fingerprint"]，整个仓库（已全量搜索 trpc_agent_sdk/ 与 tests/）没有任何代码把该值转移到响应事件的 custom_metadata["advanced_memory_request_context_fingerprint"]；而唯一读取点 _latest_usage_baseline（本文件第 178 行）只在这个 event metadata 上做 recorded_fingerprint != static_fingerprint → continue 的防错配保护。写入点与读取点之间是死链，指纹保护因此永远不会生效。 触发条件： 启用 token 模式（显式配置 model_context_window_tokens 或提供 context_window_resolver）后，AdvancedAutoCompactSummarizer.create_session_summary_by_request（_auto_compact.py:581）在每次非 blocked 压缩后调用 record_request_context，但下一轮 budget() 经 _latest_usage_baseline 读到的 recorded_fingerprint 恒为 None，保护分支永不触发；baseline 退化为仅靠 request 内容 fingerprint 匹配历史事件 content。当会话中不同轮次出现相同/相似内容（如重复指令、批处理、重放），会匹配到早前不同请求的 usage_metadata（total_token_count 含 output）作为当前上下文大小基线。 实际影响： token 模式下的自动压缩/阻断与阈值决策（trigger_warning/auto_compact/blocking 对比）基于错误的使用量基线；可能过早触发压缩破坏上下文，或在接近硬限制时使用低估基线导致请求越界失败；record_request_context 的调用方（每轮一次）也会把无消费者的状态键写入 session.state 造成状态膨胀。 修正方向： 在响应事件落库处（如 append_event 或过滤器 _after 阶段）把 state 中暂存的 advanced_memory_pending_request_context_fingerprint 写入该次调用对应事件的 custom_metadata，或直接让 record_request_context 接收将要持久化的响应事件写入 metadata；同时为 _latest_usage_baseline 增加事件 timestamp 大于当前请求起始时间的校验，避免误配基线。补充端到端测试覆盖指纹从请求转移到事件再到下一轮 baseline 的完整链路。 |
| @classmethod | ||
| def record_request_context( | ||
| self, | ||
| request: "LlmRequest", | ||
| ctx: "InvocationContext | None", | ||
| cls, | ||
| request: LlmRequest, | ||
| ctx: InvocationContext, | ||
| ) -> None: | ||
| """Stage the final request fingerprint for persistence on the response Event.""" | ||
| session = getattr(ctx, "session", None) if ctx is not None else None | ||
| state = getattr(session, "state", None) | ||
| session = ctx.session | ||
| state = session.state | ||
| if isinstance(state, dict): | ||
| state["advanced_memory_pending_request_context_fingerprint"] = _request_static_fingerprint(request) |
There was a problem hiding this comment.
问题： TokenContextTracker.record_request_context 只把请求指纹写入 session.state["advanced_memory_pending_request_context_fingerprint"]，整个仓库（已全量搜索 trpc_agent_sdk/ 与 tests/）没有任何代码把该值转移到响应事件的 custom_metadata["advanced_memory_request_context_fingerprint"]；而唯一读取点 _latest_usage_baseline（本文件第 178 行）只在这个 event metadata 上做 recorded_fingerprint != static_fingerprint → continue 的防错配保护。写入点与读取点之间是死链，指纹保护因此永远不会生效。
触发条件： 启用 token 模式（显式配置 model_context_window_tokens 或提供 context_window_resolver）后，AdvancedAutoCompactSummarizer.create_session_summary_by_request（_auto_compact.py:581）在每次非 blocked 压缩后调用 record_request_context，但下一轮 budget() 经 _latest_usage_baseline 读到的 recorded_fingerprint 恒为 None，保护分支永不触发；baseline 退化为仅靠 request 内容 fingerprint 匹配历史事件 content。当会话中不同轮次出现相同/相似内容（如重复指令、批处理、重放），会匹配到早前不同请求的 usage_metadata（total_token_count 含 output）作为当前上下文大小基线。
实际影响： token 模式下的自动压缩/阻断与阈值决策（trigger_warning/auto_compact/blocking 对比）基于错误的使用量基线；可能过早触发压缩破坏上下文，或在接近硬限制时使用低估基线导致请求越界失败；record_request_context 的调用方（每轮一次）也会把无消费者的状态键写入 session.state 造成状态膨胀。
修正方向： 在响应事件落库处（如 append_event 或过滤器 _after 阶段）把 state 中暂存的 advanced_memory_pending_request_context_fingerprint 写入该次调用对应事件的 custom_metadata，或直接让 record_request_context 接收将要持久化的响应事件写入 metadata；同时为 _latest_usage_baseline 增加事件 timestamp 大于当前请求起始时间的校验，避免误配基线。补充端到端测试覆盖指纹从请求转移到事件再到下一轮 baseline 的完整链路。
a297dc1 to
f26d550
Compare
f26d550 to
debbecd
Compare
AI Code Review审查结论不通过 审查范围：f05797d9..0aae51a（Feature/advanced memory service，计划 "Feature/advanced memory service"），111 个文件、+6191/-6569。计划符合性：整体符合——advanced memory 被拆分为 sessions/compact/advanced（自动压缩、记忆提取、协调器）与 default 两条链路，抽象层（_compact.py、_base_session_service、filter 链）一致性较好，但删除 AdvancedMemoryService/AdvancedMemorySessionService/create_advanced_memory_tools 等公共 API 时未提供任何迁移垫片。主要风险：(1) SessionOperationCoordinator.guard 存在 asyncio.wait_for 超时取消与底层线程锁获取成功的竞态，超时后锁被永久持有（验证模拟 977/977 次永久持锁），配合连续失败计数会硬阻断该 session 的后续记忆提取与自动压缩；(2) Redis 后端 update_session_state 采用 GET→merge→SET 非原子读写，多实例并发覆盖为脏写；(3) _filters.py 对 end_invocation 的死写入（全 SDK 无读取方），实际阻断仅由 is_continue=False 驱动，后续维护易误读；(4) _persist_session_compaction 在边界事件签名无法映射时抛 ValueError 并计入连续失败，可达硬阻断；(5) 双锁嵌套（asyncio.Lock 内再取协调线程锁）顺序不固定，且锁内存在长 LLM 调用，慢调用会耗尽其他 session 的 wait_timeout_seconds；(6) 测试大幅缩减（tests +638/-2488），advanced 压缩/提取/协调的核心路径无回归测试，新增 7 个测试未覆盖 guard 超时、并发、失败恢复。门禁结论：FAILED——存在可导致运行时硬阻断的高置信缺陷（guard 竞态永久持锁），且此竞态根因位于 _coordination.py（本范围内为纯 R100 重命名），故按其调用点锚定到新增文件 _compaction_memory_extractor.py 的 extract_if_needed（第 700-704 行）。 发现的问题严重
问题： 触发条件： 任一 session 的记忆提取调用在 实际影响： 该 session 的 修正方向： 在 中等
问题： 触发条件： 多实例（或多协程）对同一 实际影响： 后写覆盖先写，丢失其中一方对 修正方向： 使用 Redis 中等
问题： 触发条件： 任何启用该过滤器的对话轮次， 实际影响： 该字段无法起到“结束本次调用”的作用，属于误导性 API；后续维护者可能误用该字段实现提前终止而失效，导致无效的模型调用与 token 消耗。 修正方向： 删除 中等
问题： 触发条件： 压缩后的内容签名映射不到任何已存在的 session 事件（如事件记录被裁剪、压缩边界落在无签名记录的事件上，或并发压缩后事件被改写）。 实际影响： 一次本可恢复的映射失败使 修正方向： 将签名映射失败与外部故障分类处理（不进入连续失败计数，或计数达到阈值前自动重置）；或改为“跳过本次持久化、保留旧 checkpoint”的可降级路径，避免把确定性数据映射错误升级为硬阻断。 中等
问题： 本次变更直接删除公共 API： 触发条件： 任何依赖旧 advanced memory 公共接口的现有用户/下游代码在升级后 import 或调用即失败（ 实际影响： 破坏性兼容变更：存量集成方升级即报错，无平滑迁移路径；与计划中途段“重命名/解耦但不破坏公共契约”的意图不符。 修正方向： 在 中等
问题： 触发条件： 多个 session 并发触发压缩，或同一 session 的记忆提取与压缩并发进入；当一个请求在锁内执行慢 LLM 调用时，其他请求在入口处等锁超时。 实际影响： 锁内长调用使 修正方向： 修正方向： 修正方向：** 修正方向：修正方向：** 修正方向：** 将 LLM 调用移出锁区间（锁内仅做状态判定与 checkpoint 判定），并统一锁获取顺序（固定先 asyncio.Lock 后协调器锁），或直接复用同一把异步锁。 较低
问题： 本次变更删除原 advanced memory 测试约 2488 行，仅保留 7 个新增的 268 行（7 个测试文件（本文件共 268 行），且新测试只覆盖极简 happy path：默认 runtime 无外部存储、disabled 短路、end-of-turn 压缩、session memory 写回、tool budget 保留 event id 等； 触发条件： 后续对协调器、Redis 状态合并或失败计数逻辑的任何改动。 实际影响： 上述缺陷无法被 本次变更：核心风险路径缺乏回归保护，测试规模与实现复杂度（+6191 行实现）严重不匹配）不匹配，本审查中发现的 SEVERE 竞态因此未被任何测试捕获。 修正方向： 为 |
| async with self._runtime.coordination.guard( | ||
| session_key, | ||
| timeout=config.wait_timeout_seconds, | ||
| ) as acquired: | ||
| if not acquired: |
There was a problem hiding this comment.
问题： extract_if_needed 通过 self._runtime.coordination.guard(session_key, timeout=config.wait_timeout_seconds) 获取协调锁，而 SessionOperationCoordinator.guard 内部为 acquired = await asyncio.wait_for(lock.acquire(), timeout=timeout)；asyncio.wait_for 超时取消协程与底层 CrossLoopLock（线程锁）acquire() 已成功之间是竞态：超时路径抛出 TimeoutError 后把 acquired 置为 False，但线程锁此刻可能已被该协程成功获取（实测模拟 977/977 次在超时后锁被该会话永久持有），finally 仅在 acquired=True 时 release()，锁从此无人释放。
触发条件： 任一 session 的记忆提取调用在 wait_timeout_seconds（默认 15 秒，_config.py 中 SessionMemoryExtractorConfig.wait_timeout_seconds）内未完成（如锁被上一次遗留持有、或锁内 LLM 长调用排队），wait_for 超时且底层 acquire() 恰好在该取消窗口内成功。
实际影响： 该 session 的 CrossLoopLock 被永久持有，后续所有 extract_if_needed 调用持续返回 coordination-timeout；在 _auto_compact.py 的 _apply_scoped 中由此累加 consecutive_failures 直至 max_failures 触发硬阻断，该 session 的自动压缩与记忆提取永久失效，且进程存活期间锁无法通过正常路径恢复。
修正方向： 在 guard 中避免 “取消与获取成功” 竞态：改用带超时的异步锁原语（如 asyncio.Lock + asyncio.wait_for 的获取结果与取消保护），或将 acquire 移入 await asyncio.shield(lock.acquire()) 并在超时后仍校验实际获取状态、由获取方负责 release()；同时 CrossLoopLock 的线程获取结果应与协程结果合并，确保 finally 中的 release() 与真实持有状态一致。
| async def update_session_state( | ||
| self, | ||
| session: Session, | ||
| state_delta: dict[str, Any], | ||
| ) -> None: | ||
| """Persist session-scoped state without replacing the caller's Event window.""" | ||
| if not state_delta: | ||
| return | ||
| session.state.update(state_delta) | ||
|
|
||
| async with self._redis_storage.create_db_session() as redis_session: | ||
| key = session_key(session.app_name, session.user_id, session.id) | ||
| storage_session = await self._get_session(redis_session, key) | ||
| if not storage_session: | ||
| logger.warning( | ||
| "Session %s not found in Redis while updating state", | ||
| session.id, | ||
| ) | ||
| return | ||
| storage_session.state.update(state_delta) | ||
| await self._set_session(redis_session, storage_session) |
There was a problem hiding this comment.
问题： update_session_state 采用 GET（读取现有 SessionStateWithTTL）→ 内存 merge → SET 写回的非原子读改写流程，无任何锁或 CAS/LUA 脚本保护。
触发条件： 多实例（或多协程）对同一 memorized_id 并发执行压缩写入（本变更同时引入了跨进程并行的 AdvancedMemoryRuntime/session_compaction 幂等写入路径），两个进程同时读到同一旧状态后在各自 merge 基础上覆写。
实际影响： 后写覆盖先写，丢失其中一方对 events/session_state 的更新（脏写），压缩后的 session 状态在多实例部署下不可靠；与 InMemory 后端 stored.session.state.update 的原子语义也不一致。
修正方向： 使用 Redis WATCH/事务或 SET（带 NX/版本号条件）实现原子读改写，或改用 Lua 脚本在服务端合并 SessionStateWithTTL 后原子写回。
| invocation_ctx: InvocationContext = get_invocation_ctx() | ||
| summarizer_manager = self.get_summarizer_manager(invocation_ctx) | ||
| result = await summarizer_manager.create_session_summary_before_model( | ||
| req, | ||
| invocation_ctx, | ||
| ) | ||
| if not result: | ||
| return None | ||
| invocation_ctx.end_invocation = True | ||
| rsp.rsp = result | ||
| rsp.is_continue = False |
There was a problem hiding this comment.
问题： _before 中在模型调用前写入 invocation_ctx.end_invocation = True，但整个 SDK（含 run_stream/_handle_co 及 end_invocation 的所有可能出现位置）不存在任何读取该字段的代码，属于死写入；实际的响应阻断仅由 rsp.is_continue = False 驱动（run_stream 据此跳过 handle 并直接产出 event.rsp）。
触发条件： 任何启用该过滤器的对话轮次，end_invocation 被无意义置位，而需要提前结束的轮次若只依赖该字段则会继续调用模型。
实际影响： 该字段无法起到“结束本次调用”的作用，属于误导性 API；后续维护者可能误用该字段实现提前终止而失效，导致无效的模型调用与 token 消耗。
修正方向： 删除 end_invocation 赋值（及接口中该字段），阻断逻辑统一收敛到 is_continue=False 语义；若确需“结束调用”语义，应将其接入 run_stream/_handle_co 的真实分支判断。
| async def _persist_session_compaction( | ||
| self, | ||
| ctx: InvocationContext, | ||
| record: AdvancedAutoCompactRecord, | ||
| ) -> None: | ||
| """Persist the compacted active window through the original SessionService.""" | ||
| compact_events = ctx.session.compact_events | ||
| if not callable(compact_events): | ||
| # AdvancedAutoCompactSummarizer remains usable as a request-only primitive in unit | ||
| # tests and custom integrations. The standard Manager supplies | ||
| # the framework Session and persists the compacted window. | ||
| return | ||
|
|
||
| boundary_event_id = record.boundary_event_id or self._resolve_boundary_event_id( | ||
| ctx, | ||
| record.boundary_signature, | ||
| record.boundary_occurrence, | ||
| ) | ||
| if boundary_event_id is None: | ||
| raise ValueError("Cannot map the AdvancedAutoCompactSummarizer boundary to an active Session Event") | ||
|
|
||
| compaction_id = record.compaction_id or f"advanced_autocompact:{uuid.uuid4().hex}" | ||
| summary_event = Event( | ||
| invocation_id="summary", | ||
| author="system", | ||
| content=self._summary_content(record.summary), | ||
| custom_metadata={ | ||
| "session_compaction_source": record.source, | ||
| "session_compaction_boundary_signature": record.boundary_signature, | ||
| "session_compaction_boundary_occurrence": record.boundary_occurrence, | ||
| }, | ||
| ) | ||
| active_before = list(ctx.session.events) | ||
| historical_before = list(ctx.session.historical_events) | ||
| last_update_before = ctx.session.last_update_time | ||
| try: | ||
| changed = compact_events( | ||
| summary_event, | ||
| boundary_event_id, | ||
| compaction_id=compaction_id, | ||
| ) | ||
| if changed: | ||
| await ctx.session_service.update_session(ctx.session) | ||
| except Exception: | ||
| ctx.session.events = active_before | ||
| ctx.session.historical_events = historical_before | ||
| ctx.session.last_update_time = last_update_before | ||
| raise |
There was a problem hiding this comment.
问题： _persist_session_compaction 在 boundary_event_id is None 时按内容签名在既有事件中查找边界事件，找不到会抛 ValueError；该异常沿 _apply_scoped（第 619 行起）的捕获路径进入 consecutive_failures += 1（第 776 行），与运行时/网络类故障同等对待。
触发条件： 压缩后的内容签名映射不到任何已存在的 session 事件（如事件记录被裁剪、压缩边界落在无签名记录的事件上，或并发压缩后事件被改写）。
实际影响： 一次本可恢复的映射失败使 consecutive_failures 计数上升，连续多次即触发 max_failures 硬阻断，该 session 记忆提取与自动压缩被永久停用，且该失败模式在正常情况下无法自愈。
修正方向： 将签名映射失败与外部故障分类处理（不进入连续失败计数，或计数达到阈值前自动重置）；或改为“跳过本次持久化、保留旧 checkpoint”的可降级路径，避免把确定性数据映射错误升级为硬阻断。
| # the tools package free of optional file/web tool dependencies) but exposed | ||
| # here for discoverability. Not in ``__all__`` so ``import *`` stays lazy. | ||
| _LAZY_REEXPORTS = { | ||
| "AdvancedMemoryTools": ( | ||
| "trpc_agent_sdk.tools._advanced_memory_tool", | ||
| "AdvancedMemoryTools", | ||
| "AdvancedMemory": ( | ||
| "trpc_agent_sdk.tools.advanced_memory", | ||
| "AdvancedMemory", | ||
| ), | ||
| "create_advanced_memory_tools": ( | ||
| "trpc_agent_sdk.tools._advanced_memory_tool", | ||
| "create_advanced_memory_tools", | ||
| "AdvancedMemoryToolSet": ( | ||
| "trpc_agent_sdk.tools.advanced_memory", | ||
| "AdvancedMemoryToolSet", | ||
| ), | ||
| "create_advanced_memory_toolset": ( | ||
| "trpc_agent_sdk.tools.advanced_memory", | ||
| "create_advanced_memory_toolset", | ||
| ), |
There was a problem hiding this comment.
问题： 本次变更直接删除公共 API：AdvancedMemoryService、AdvancedMemorySessionService、AdvancedMemoryTools/create_advanced_memory_tools、trpc_agent_sdk.advanced_memory 包（AdvancedMemoryConfig）及 Runner 绑定分支，且未提供任何 deprecation 别名或迁移垫片，_LAZY_REEXPORTS 中旧名称同步移除。
触发条件： 任何依赖旧 advanced memory 公共接口的现有用户/下游代码在升级后 import 或调用即失败（ImportError/AttributeError），运行期无任何提示。
实际影响： 破坏性兼容变更：存量集成方升级即报错，无平滑迁移路径；与计划中途段“重命名/解耦但不破坏公共契约”的意图不符。
修正方向： 在 tools/__init__.py 或保留的包路径中提供别名转发到新实现（如 create_advanced_memory_tools → 新 toolset 工厂），并标注 DeprecationWarning 至少一个发布周期后再移除。
| async with self._session_lock(session_id): | ||
| request.contents = [content.model_copy(deep=True) for content in request.contents] | ||
| state = await self._load_state(session_id) | ||
| reapplied = False | ||
| if state.latest_compaction is not None: | ||
| reapplied = self._apply_record(request, state.latest_compaction) | ||
|
|
||
| request_chars_before = estimate_request_chars(request) | ||
| token_budget_before = tracker.budget(request, ctx) | ||
| token_mode = token_budget_before.token_mode_enabled | ||
| request_tokens_before = token_budget_before.estimate.tokens | ||
| comparison_tokens_before = (tracker.estimate_request_tokens(request) if token_mode else None) | ||
| blocking_reached = (request_tokens_before >= token_budget_before.blocking_threshold_tokens | ||
| if token_mode else request_chars_before >= self._auto_compact_config.blocking_chars) | ||
| if state.consecutive_failures >= auto_compact_config.max_failures and blocking_reached: | ||
| return AdvancedAutoCompactResult( | ||
| compacted=False, | ||
| reapplied=reapplied, | ||
| blocked=True, | ||
| source=None, | ||
| request_chars_before=request_chars_before, | ||
| request_chars_after=request_chars_before, | ||
| consecutive_failures=state.consecutive_failures, | ||
| ) | ||
| if state.consecutive_failures >= auto_compact_config.max_failures: | ||
| return AdvancedAutoCompactResult( | ||
| compacted=False, | ||
| reapplied=reapplied, | ||
| blocked=False, | ||
| source=None, | ||
| request_chars_before=request_chars_before, | ||
| request_chars_after=request_chars_before, | ||
| consecutive_failures=state.consecutive_failures, | ||
| ) | ||
| auto_compact_reached = (request_tokens_before >= token_budget_before.auto_compact_threshold_tokens | ||
| if token_mode else request_chars_before >= auto_compact_config.trigger_chars) | ||
| if not force and not auto_compact_reached: | ||
| return AdvancedAutoCompactResult( | ||
| compacted=False, | ||
| reapplied=reapplied, | ||
| blocked=False, | ||
| source=state.latest_compaction.source if reapplied and state.latest_compaction else None, | ||
| request_chars_before=request_chars_before, | ||
| request_chars_after=request_chars_before, | ||
| consecutive_failures=state.consecutive_failures, | ||
| ) | ||
|
|
||
| original_contents = [content.model_copy(deep=True) for content in request.contents] | ||
| try: | ||
| compact_record: AdvancedAutoCompactRecord | None = None | ||
| if self._session_memory_extractor is not None: | ||
| await self._session_memory_extractor.extract_if_needed( | ||
| ctx, | ||
| force=True, | ||
| ) |
There was a problem hiding this comment.
问题： _apply_scoped 在 asyncio.Lock（第 643 行）保护下先调用 extract_if_needed(ctx, force=True)（第 694-697 行），后者内部再获取协调器线程锁并按 wait_timeout_seconds 等待——两层锁嵌套顺序在每个入口不固定（extract_if_needed 也可不经 _apply_scoped 单独调用），且锁保护区间跨越长耗时的 LLM 摘要调用。
触发条件： 多个 session 并发触发压缩，或同一 session 的记忆提取与压缩并发进入；当一个请求在锁内执行慢 LLM 调用时，其他请求在入口处等锁超时。
实际影响： 锁内长调用使 wait_timeout_seconds 形同虚设——排队请求大量返回 coordination-timeout 并累加失败计数（叠加到硬阻断），高并发下表现为压缩/提取大面积失败；双锁嵌套顺序不一致还使死锁风险面扩大。
修正方向： 修正方向： 修正方向：** 修正方向：修正方向：** 修正方向：** 将 LLM 调用移出锁区间（锁内仅做状态判定与 checkpoint 判定），并统一锁获取顺序（固定先 asyncio.Lock 后协调器锁），或直接复用同一把异步锁。
| def test_compact_runtime_has_no_external_storage() -> None: | ||
| runtime = AdvancedAutoCompactSummarizerRuntime(AdvancedAutoCompactSummarizerConfig()) | ||
|
|
||
| assert not hasattr(runtime, "transcripts") | ||
| assert not hasattr(runtime, "tool_results") | ||
| assert not hasattr(runtime, "paths") | ||
|
|
||
|
|
||
| @pytest.mark.asyncio | ||
| async def test_disabled_auto_compact_returns_without_recursion() -> None: | ||
| config = AdvancedAutoCompactSummarizerConfig( | ||
| auto_compact=AutoCompactSummarizerConfig(enabled=False), | ||
| ) | ||
| summarizer = AdvancedAutoCompactSummarizer(config) | ||
| session = SimpleNamespace(id="session", app_name="app", user_id="user") | ||
| ctx = SimpleNamespace(session=session, session_id=session.id) | ||
|
|
||
| result = await summarizer.apply(LlmRequest(), ctx=ctx) | ||
|
|
||
| assert result.compacted is False | ||
| assert result.blocked is False | ||
|
|
||
|
|
||
| @pytest.mark.asyncio | ||
| async def test_session_service_accepts_a_configured_compact_manager() -> None: | ||
| summarizer = AdvancedAutoCompactSummarizer(AdvancedAutoCompactSummarizerConfig()) | ||
| manager = AdvancedAutoCompactSummarizerManager(summarizer) | ||
| service = InMemorySessionService( | ||
| session_config=SessionServiceConfig(store_historical_events=True), | ||
| summarizer_manager=manager, | ||
| ) | ||
|
|
||
| assert service.summarizer_manager is manager | ||
| await service.close() | ||
|
|
||
|
|
||
| @pytest.mark.asyncio | ||
| async def test_advanced_summarizer_implements_compact_abc_and_timing() -> None: | ||
| summarizer = AdvancedAutoCompactSummarizer(AdvancedAutoCompactSummarizerConfig( | ||
| session_memory=SessionMemoryExtractorConfig(enabled=False), | ||
| )) | ||
| before_model_manager = AdvancedAutoCompactSummarizerManager(summarizer) | ||
| after_turn_manager = AdvancedAutoCompactSummarizerManager( | ||
| summarizer, | ||
| compact_trigger=CompactTrigger.AFTER_TURN, | ||
| ) | ||
|
|
||
| assert isinstance(summarizer, CompactSummarizerABC) | ||
| assert before_model_manager.compact_trigger == CompactTrigger.BEFORE_MODEL | ||
| assert after_turn_manager.compact_trigger == CompactTrigger.AFTER_TURN | ||
|
|
||
| session = SimpleNamespace(events=[]) | ||
| ctx = SimpleNamespace(session=session) | ||
| summarizer.should_summarize = AsyncMock(return_value=True) | ||
| summarizer.create_session_summary = AsyncMock(return_value="summary") | ||
| await before_model_manager.create_session_summary(session, ctx=ctx) | ||
| summarizer.should_summarize.assert_not_awaited() | ||
|
|
||
| await after_turn_manager.create_session_summary(session, ctx=ctx) | ||
| summarizer.should_summarize.assert_awaited_once_with(session) | ||
| summarizer.create_session_summary.assert_awaited_once_with( | ||
| session, | ||
| ctx=ctx, | ||
| store_historical_events=True, | ||
| ) | ||
|
|
||
|
|
||
| @pytest.mark.asyncio | ||
| async def test_advanced_end_of_turn_compaction_moves_old_events_to_history() -> None: | ||
| service = InMemorySessionService( | ||
| session_config=SessionServiceConfig(store_historical_events=True), | ||
| ) | ||
| session = await service.create_session( | ||
| app_name="app", | ||
| user_id="user", | ||
| session_id="session", | ||
| ) | ||
| for index in range(3): | ||
| await service.append_event( | ||
| session, | ||
| _event( | ||
| f"event-{index}", | ||
| Content(role="user", parts=[Part.from_text(text=f"{index}:" + "x" * 2_000)]), | ||
| ), | ||
| ) | ||
| original_ids = [event.id for event in session.events] | ||
| config = AdvancedAutoCompactSummarizerConfig( | ||
| session_memory=SessionMemoryExtractorConfig(enabled=False), | ||
| auto_compact=AutoCompactSummarizerConfig( | ||
| trigger_chars=500, | ||
| target_chars=250, | ||
| blocking_chars=10_000, | ||
| keep_recent_contents=1, | ||
| ), | ||
| ) | ||
| summarizer = AdvancedAutoCompactSummarizer(config, model=_SummaryModel()) | ||
| ctx = SimpleNamespace( | ||
| session=session, | ||
| session_id=session.id, | ||
| session_service=service, | ||
| agent_context=new_agent_context(), | ||
| agent=SimpleNamespace(model=_SummaryModel()), | ||
| ) | ||
|
|
||
| assert await summarizer.should_summarize(session) is True | ||
| summary = await summarizer.create_session_summary( | ||
| session, | ||
| ctx=ctx, | ||
| store_historical_events=True, | ||
| ) | ||
|
|
||
| assert summary is not None and "Compact summary" in summary | ||
| assert len(session.events) == 2 | ||
| assert session.events[0].is_summary_event() | ||
| assert session.events[1].id == original_ids[-1] | ||
| assert [event.id for event in session.historical_events] == original_ids[:-1] | ||
| await service.close() | ||
|
|
||
|
|
||
| @pytest.mark.asyncio | ||
| async def test_session_memory_is_written_to_session_state() -> None: | ||
| service = InMemorySessionService(session_config=SessionServiceConfig(store_historical_events=True), ) | ||
| session = await service.create_session( | ||
| app_name="app", | ||
| user_id="user", | ||
| session_id="session", | ||
| ) | ||
| await service.append_event( | ||
| session, | ||
| _event("event-1", Content(parts=[Part.from_text(text="hello")])), | ||
| ) | ||
| config = AdvancedAutoCompactSummarizerConfig( | ||
| session_memory=SessionMemoryExtractorConfig( | ||
| initial_chars=1, | ||
| update_chars=1, | ||
| ), | ||
| ) | ||
| extractor = SessionMemoryExtractor( | ||
| AdvancedAutoCompactSummarizerRuntime(config), | ||
| model=_MemoryModel(), | ||
| ) | ||
|
|
||
| result = await extractor.extract_if_needed( | ||
| SimpleNamespace( | ||
| session=session, | ||
| session_service=service, | ||
| agent=SimpleNamespace(model=_MemoryModel(), generate_content_config=None), | ||
| override_messages=None, | ||
| ), | ||
| force=True, | ||
| ) | ||
|
|
||
| assert result.extracted is True | ||
| assert "_trpc_agent:summary" in session.state | ||
| stored = await service.get_session( | ||
| app_name="app", | ||
| user_id="user", | ||
| session_id="session", | ||
| ) | ||
| assert stored is not None | ||
| assert "_trpc_agent:summary" in stored.state | ||
| await service.close() | ||
|
|
||
|
|
||
| @pytest.mark.asyncio | ||
| async def test_tool_result_budget_keeps_the_session_event_id() -> None: | ||
| service = InMemorySessionService() | ||
| session = await service.create_session( | ||
| app_name="app", | ||
| user_id="user", | ||
| session_id="session", | ||
| ) | ||
| content = Content(parts=[ | ||
| Part(function_response=FunctionResponse( | ||
| id="tool-call-1", | ||
| name="demo", | ||
| response={"output": "x" * 500}, | ||
| )) | ||
| ]) | ||
| await service.append_event(session, _event("event-tool", content)) | ||
| request = LlmRequest(model="test", contents=[content.model_copy(deep=True)]) | ||
| config = AdvancedAutoCompactSummarizerConfig( | ||
| tool_result_budget=ToolResultBudgetConfig( | ||
| max_chars=100, | ||
| preview_chars=20, | ||
| ), | ||
| ) | ||
| budget = ToolResultBudget( | ||
| AdvancedAutoCompactSummarizerRuntime(config), | ||
| ) | ||
|
|
||
| await budget.apply( | ||
| request, | ||
| ctx=SimpleNamespace(session=session), | ||
| ) | ||
|
|
||
| replacement = request.contents[0].parts[0].function_response.response | ||
| assert replacement["session_event_id"] == "event-tool" | ||
| assert "path" not in replacement | ||
| await service.close() |
There was a problem hiding this comment.
问题： 本次变更删除原 advanced memory 测试约 2488 行，仅保留 7 个新增的 268 行（7 个测试文件（本文件共 268 行），且新测试只覆盖极简 happy path：默认 runtime 无外部存储、disabled 短路、end-of-turn 压缩、session memory 写回、tool budget 保留 event id 等；_coordination.py 的 guard 超时/竞态、update_session_state 并发、consecutive_failures 硬阻断恢复、_persist_session_compaction 边界映射失败等关键路径均无测试。
触发条件： 后续对协调器、Redis 状态合并或失败计数逻辑的任何改动。
实际影响： 上述缺陷无法被 本次变更：核心风险路径缺乏回归保护，测试规模与实现复杂度（+6191 行实现）严重不匹配）不匹配，本审查中发现的 SEVERE 竞态因此未被任何测试捕获。
修正方向： 为 guard（超时取消与线程锁竞态）、Redis update_session_state 并发、consecutive_failures 达到阈值后的阻断恢复与解除阻断、以及 `_persist 恢复路径增加针对性测试，并恢复对原 advanced memory 行为的存量回归用例。
AI Code Review审查结论不通过 审查范围：base_commit f05797d（v1.1.20）到 head_commit debbecd 共 111 个文件变更（+6187/-6574），核心是 advanced memory 服务化重构：sessions 包拆出 compact/default 与 compact/advanced 两个压缩管线，删除旧 advanced_memory 与 AdvancedMemorySessionService，新增 tools/advanced_memory 独立组件（local/redis/sql 三后端）、Session.compact_events 幂等压缩、session_service.update_session_state 增量状态写入、SQL get_for_update 行锁。计划 Feature/advanced memory service 的模块结构与模块拆分已基本落地。计划符合性方面，模块拆分边界、存储后端、记忆与上下文解耦的核心路径与计划一致。主要风险：(1) 严重——AdvancedAutoCompactSummarizer 默认构造（config=None）直接 AttributeError，组件默认配置不可用；(2) DefaultSessionSummarizerManager 未达阈值时每轮全量 update_session 写放大，与旧行为不一致；(3) before-model handler 类型耦合，非 Advanced 管线绑定时报错路径每轮抛错；(4) 提取输入截断导致长期记忆遗漏早期事实；(5) compact_events 幂等性仅扫描当前 events 窗口，窗口裁剪后可重复压缩。测试充分性：新增 tests/sessions/compact/test_session_compact.py 覆盖核心压缩、session memory 状态写入、tool result budget 事件引用等用例，会话服务三后端新增了压缩持久化与状态增量写入测试，覆盖较充分；但未覆盖 AdvancedAutoCompactSummarizer 默认构造路径与 handler 类型不匹配路径。门禁结论：存在 1 个 SEVERE（默认构造崩溃）与多个 MODERATE 高置信问题，建议修复后再合入。 发现的问题严重
问题： 触发条件： 任何以 实际影响： 构造直接崩溃 修正方向： 保存解析后的 config： 中等
问题： 触发条件： 任一使用 实际影响： 不需要压缩的会话每轮全量落库——SQL 后端 修正方向： 仅在真正发生压缩时才更新存储：把 中等
问题： 触发条件： 用户按框架支持的方式把 实际影响： 每次模型调用在 before-model 过滤器阶段抛异常，LLM 请求直接失败/被转换为错误事件——类型不匹配本应在绑定/构造阶段校验，而非热路径每轮抛错。 修正方向： 在 中等
问题： 触发条件： 会话上下文超过 实际影响： 会话记忆文档缺失关键历史事实（早期目标/决定），跨会话恢复时模型无法还原被截断内容（历史事件在后续压缩中可能被移除），长期记忆正确性受损。 修正方向： 截断 中等
问题： 触发条件： 压缩成功持久化后，events 窗口因 实际影响： 同一边界内容被二次压缩，历史事件重复归档、事件窗口意外缩小，模型可见上下文丢失，属于数据完整性风险。 修正方向： 幂等性检查除扫描 较低
问题： 触发条件： 多个会话并发触发 before-model 管线，第二个会话在等待锁时超过 实际影响： 记忆提取被静默跳过（ 修正方向： 对 |
| self._runtime = AdvancedAutoCompactSummarizerRuntime(config=config or AdvancedAutoCompactSummarizerConfig()) | ||
| self._auto_compact_config = config.auto_compact | ||
| self._session_memory_extractor = self._create_extractor(session_memory_extractor) |
There was a problem hiding this comment.
问题： AdvancedAutoCompactSummarizer.__init__（trpc_agent_sdk/sessions/compact/advanced/_auto_compact.py:127-129）在 config is None 时构造了默认 config 并赋给 self._runtime，但下一行仍执行 config.auto_compact——config 为 None 时直接 AttributeError。
触发条件： 任何以 AdvancedAutoCompactSummarizer() 无参方式构造的调用（API 设计允许默认构造，示例代码也以 config 构造为主）。
实际影响： 构造直接崩溃 AttributeError: 'NoneType' object has no attribute 'auto_compact'，该组件在默认配置下完全不可用。
修正方向： 保存解析后的 config：resolved = config or AdvancedAutoCompactSummarizerConfig()，随后统一使用 resolved（self._runtime = AdvancedAutoCompactSummarizerRuntime(config=resolved)、self._auto_compact_config = resolved.auto_compact）。
| @@ -122,25 +107,28 @@ async def create_session_summary(self, | |||
| self._summarizer_cache[app_name] = {} | |||
| if user_id not in self._summarizer_cache[app_name]: | |||
| self._summarizer_cache[app_name][user_id] = {} | |||
| self._summarizer_cache[app_name][user_id][session.id] = SessionSummary( | |||
| self._summarizer_cache[app_name][user_id][session.id] = DefaultSessionSummary( | |||
| session_id=session.id, | |||
| summary_text=summary_text, | |||
| original_event_count=original_event_count, | |||
| compressed_event_count=len(session.events), | |||
| summary_timestamp=time.time(), | |||
| model_name=getattr(self._summarizer.model, "name", ""), | |||
| ) | |||
| session.conversation_count = 0 | |||
| # Update the stored session | |||
| if self._base_service: | |||
| await self._base_service.update_session(session) | |||
There was a problem hiding this comment.
问题： DefaultSessionSummarizerManager.create_session_summary（trpc_agent_sdk/sessions/compact/default/_summarizer_manager.py:86-121）在 force=False 且 should_summarize_session 返回 False（不需要压缩）时仍执行 if self._base_service: await self._base_service.update_session(session)。旧实现 SummarizerSessionManager（f05797d 的 _summarizer_manager.py）把 update_session 放在 is_should_summarize 分支内；新实现分支外的无条件更新是本 PR 引入的。
触发条件： 任一使用 DefaultSessionSummarizerManager 的会话，只要某轮 should_summarize 返回 False（set_summarizer_token_threshold 因缺少 usage_metadata 事件返回 False、事件数不足、conversation_count 未达标、auto_summarize=False），post-turn 钩子每轮都会执行 update_session。
实际影响： 不需要压缩的会话每轮全量落库——SQL 后端 update_session 先 delete 再 insert 该会话全部 events，Redis 后端重写整个 session JSON。每轮一次的无谓数据库写放大，与旧行为不一致，高频短会话场景下开销显著。
修正方向： 仅在真正发生压缩时才更新存储：把 update_session 移入 if summary_text: 分支内，未压缩时直接 return。
| @@ -607,19 +581,60 @@ def missing_context(end: int) -> list[str]: | |||
|
|
|||
| return [], None | |||
There was a problem hiding this comment.
问题： SessionMemoryExtractor._build_extraction_input（_compaction_memory_extractor.py:500-582）在上下文超预算时把 _context_suffix 后的旧消息直接截掉，且 safe_ends 只保证不把带 tool-call 的事件作为 last_event_id，没有保证被截断的 tool-call/tool-response 配对 turn 完整进入提取输入——被丢弃的旧消息（如用户目标、已确认决定）随后被 _persist_checkpoint 前移 last_event_id，永不再参与提取。
触发条件： 会话上下文超过 prompt_max_chars/上下文窗口预算触发会话记忆更新时，fraction 降级路径（0.8/0.6/0.4）丢弃早期消息。
实际影响： 会话记忆文档缺失关键历史事实（早期目标/决定），跨会话恢复时模型无法还原被截断内容（历史事件在后续压缩中可能被移除），长期记忆正确性受损。
修正方向： 截断 _context_suffix 后，优先保留与 pending tool-call/tool-response 配对完整的 turn（把相关消息一起保留），或在 missing_context 中把 tool-call 所在的完整 turn 一并补回，避免半截 turn 进入提取输入。
|
|
||
| def compact_events( | ||
| self, | ||
| summary_event: Event, | ||
| boundary_event_id: str, | ||
| *, | ||
| compaction_id: str, | ||
| ) -> bool: | ||
| """Replace the active prefix through ``boundary_event_id`` with a summary. | ||
|
|
||
| The replaced active Events remain recoverable in ``historical_events``. | ||
| ``compaction_id`` makes retries idempotent when a persistence operation | ||
| succeeds but its caller does not observe the result. | ||
| """ | ||
| for event in self.events: | ||
| metadata = event.custom_metadata or {} | ||
| if metadata.get("session_compaction_id") == compaction_id: | ||
| return False | ||
|
|
||
| boundary_index = next( | ||
| (index for index, event in enumerate(self.events) if event.id == boundary_event_id), | ||
| None, | ||
| ) | ||
| if boundary_index is None: | ||
| raise ValueError(f"Session compaction boundary Event {boundary_event_id!r} " | ||
| "is not in the active event window") | ||
|
|
||
| replaced = self.events[:boundary_index + 1] | ||
| if not replaced: | ||
| return False | ||
|
|
||
| metadata = dict(summary_event.custom_metadata or {}) | ||
| metadata.update({ | ||
| "session_compaction_id": compaction_id, | ||
| "session_compaction_boundary_event_id": boundary_event_id, | ||
| }) | ||
| summary_event.custom_metadata = metadata | ||
| summary_event.set_summary_event(True) | ||
| # SQL backends restore active Events in timestamp order. Give the | ||
| # replacement summary the prefix's timestamp so it remains the anchor | ||
| # before every retained Event after persistence. | ||
| summary_event.timestamp = replaced[0].timestamp | ||
|
|
||
| historical_ids = {event.id for event in self.historical_events} | ||
| self.historical_events.extend(event for event in replaced if event.id not in historical_ids) | ||
| self.events = [summary_event, *self.events[boundary_index + 1:]] | ||
| self.last_update_time = max(self.last_update_time, summary_event.timestamp) | ||
| return True |
There was a problem hiding this comment.
问题： Session.compact_events（trpc_agent_sdk/sessions/_session.py:139-186）的幂等性检查只扫描当前 self.events 中带 session_compaction_id 的 summary Event。当一次压缩已成功持久化、但随后事件窗口被裁剪或重新加载（TTL 过滤、num_recent_events 截断、get_session 重新加载）导致该 summary Event 离开 self.events 时，同一 boundary_signature 在下一次自动压缩中会被再次压缩。
触发条件： 压缩成功持久化后，events 窗口因 apply_event_filtering（_session.py:_apply_event_filtering_and_get_filtered_events）或 Service 重新加载发生裁剪，同一边界在下一次自动压缩中再次命中。
实际影响： 同一边界内容被二次压缩，历史事件重复归档、事件窗口意外缩小，模型可见上下文丢失，属于数据完整性风险。
修正方向： 幂等性检查除扫描 self.events 外，同时扫描 self.historical_events 中的 session_compaction_id，或把已处理的 compaction_id 记录到 session state 后再检查。
| processor = self._scoped_processors.get(runtime.scope) | ||
| if processor is None: | ||
| processor = copy.copy(self) | ||
| processor._runtime = runtime | ||
| processor._states = {} | ||
| processor._session_locks = {} | ||
| self._scoped_processors[runtime.scope] = processor | ||
| return await processor.apply(request, ctx=ctx, force=force) |
There was a problem hiding this comment.
问题： AdvancedAutoCompactSummarizer 的 _scoped_processors 缓存（_auto_compact.py:610-617）与各个 handler 的共享处理器在 _apply_scoped 的锁之上依赖 asyncio.Lock 按 session 串行化；SessionOperationCoordinator.guard 超时（wait_timeout_seconds=15s）后 acquired=False 直接 return，不保留后续重试。
触发条件： 多个会话并发触发 before-model 管线，第二个会话在等待锁时超过 wait_timeout_seconds。
实际影响： 记忆提取被静默跳过（coordination-timeout），跨会话长期记忆更新延迟；不构成数据损坏。
修正方向： 对 coordination-timeout 分支在后续 turn 恢复重试（当前直接 return 不排队），或完善 CrossLoopLock.acquire 的中途取消处理（asyncio.wait_for 取消后锁状态一致）。
AI Code Review审查结论不通过 审查范围：base f05797d..head 9ff3397，共 107 个变更文件（+6008/-6410），核心是将 advanced memory 机制重构为 trpc_agent_sdk/sessions/compact/advanced/ 筛选器链（AutoCompact、TokenBudget、HistorySnip、MicroCompact、ToolResultBudget、CompactionMemoryExtractor、RedisStores、Coordination）与 trpc_agent_sdk/memory/advanced_memory/ 运行时，并删除旧版实现及对应测试。计划符合性：总体目标（advanced memory 服务化、记忆与上下文解耦、可配置、并发协调、筛选器链与 summarizer 迁移）基本达成，机制层次与抽象完整，新增测试覆盖显式配置下的内存态 happy path（test_session_compact、test_token_budget、test_coordination）并附示例。主要风险：① AutoCompact 断路器达到 max_failures 后进程内永久短路且无恢复路径，阻塞性失败会让超大请求被直接放行；② 断路器计数、各筛选器状态、pending fingerprint 全部仅存内存，旧版 transcript/持久化已删除，重启后同一 tool_call id 不同内容会触发 content-reuse ValueError 使模型调用错误化；③ usage fingerprint 写入方保留而消费方（pending 写入 event custom_metadata）已删除，usage baseline 跨模型/配置复用失效并污染 session.state；④ 筛选器异常路径将 blocked 阻断响应与 error 一并吞掉并向上产出 None 事件，导致本应被阻止的超大请求继续发送或请求直接失败；⑤ Redis read_index 锁外 prune 与锁内 write_index 并发丢失更新、SQL 端 update_session_state 不滚动 update_time 导致 TTL 语义与内存/Redis 后端不一致；⑥ checkpoint 事件被 compact_events 移入 historical_events 后按时间戳恢复失败，提取永久停滞在 no-new-events；⑦ 会话锁内执行秒级 LLM 压缩调用阻塞同会话并发请求。测试充分性：None-config、重启、并发、断路器重置、Redis/SQL 后端及工具注入路径均无覆盖。门禁结论：FAILED（存在 SEVERE 评论，阻止合入）。 发现的问题严重
问题： 触发条件： 调用方以 实际影响： 对象构造即崩溃，advanced memory 筛选器初始化失败； 修正方向： 解引用前先归一化，例如 严重
问题： 断路器早退分支在 触发条件： 累计 N 次压缩失败（如 LLM 不可用、压缩未降体积）达到 实际影响： 早退为 blocked 的请求携带错误摘要直接放行超大上下文（信息截断语义被破坏），早退为非 blocked 的请求则永久跳过压缩；断路器没有冷却时间或半开探测，只能靠重启进程恢复（计数仅存内存）。 修正方向： 在早退分支发现失败计数已达上限但本次请求按当前指标已无需压缩（非 blocking）时允许重置计数，或在成功（未失败）路径的 try 外也执行计数清零 / 引入基于时间窗的半开重试机制。 中等
问题： 触发条件： 服务重启后，会话上下文中再次出现与重启前相同 id 的 tool_call（工具重放、请求重试、多 worker 共享会话等场景）。 实际影响： 异常经筛选器 修正方向： 对 tool_call id 附加时间/会话代际维度使其跨重启唯一，或对 content 不一致的场景降级为忽略旧记录而不是抛异常；同时考虑把关键状态落盘（Redis/SQL）而非仅存内存。 中等
问题： 触发条件： 任意携带上轮 usage 统计的请求继续进入 token 估算流程；session.state 中残留的 pending 指纹字段也会被持久化。 实际影响： 指纹校验永不生效，usage baseline 会跨模型、跨配置地被复用，token 预算估算失真（该压缩时未压缩、该放行时误阻断）；session.state 被无关字段污染并随会话持久化。 修正方向： 恢复 pending 指纹到 event 元数据的传递链路，或在 中等
问题： 触发条件： 任意 handler（tool_result_budget、history_snip、micro_compact、auto_compact）抛异常（如 content-reuse ValueError、LLM 压缩失败），或模型调用本身失败时。 实际影响： 本应及时返回给用户的阻断响应（“内容超限已阻止发送”）在异常路径上被丢弃，改为向上抛错误事件/异常使会话失败；本应被阻止的超大请求在部分异常场景下反而继续发送。 修正方向： 在 except 路径保留已构造的 中等
问题： 触发条件： checkpoint 落在已被压缩的事件区间内（长期运行会话的典型情况），随后再次调用 实际影响： 修正方向： 在 historical_events 中按 checkpoint 定位被替换的原始事件并以其时间戳作为恢复基准，或在提取完成前不压缩 checkpoint 所在区间，保证 checkpoint 始终可解析。 中等
问题： 触发条件： 两个及以上进程/线程同时操作同一会话记忆索引（prune 与 save_memory 交错）。 实际影响： 刚保存的记忆条目可能被并发 prune 覆盖丢失，或 prune 结果被旧快照覆盖导致索引膨胀，记忆索引与条目不一致。 修正方向： 将 prune 的读取、计算、写入整体移入 中等
问题： 触发条件： 会话仅通过 实际影响： SQL 后端下活跃会话（持续写 state 但无 event 追加）被判定过期并清理，checkpoint/记忆状态被提前逐出，与内存/Redis 后端表现不一致。 修正方向： 在 中等
问题： 触发条件： 同一 session 的多个请求并发到达且任一请求触发迫近 blocking 的压缩路径。 实际影响： 后续同会话请求在锁上串行等待秒级 LLM 压缩，请求延迟被成倍放大，高并发下形成排队拥塞，加剧 TokenBudget 超阈值。 修正方向： 将 LLM 压缩调用移出锁外（锁仅保护状态读改写），或在锁外预取 session memory 记录、锁内仅做纯计算。 较低
问题： 786 行 触发条件： 断路器触发 blocked 早退（达到 max_failures 且请求超限）时，调用方在结果中查询失败原因。 实际影响： 调用方（筛选器/上层）无法从结果感知熔断原因，只能靠日志排查；blocked 结果与正常结果的 error 字段区分度为零。 修正方向： 早退分支把当前失败计数/最近一次异常写入 较低
问题： 根 runtime 的 触发条件： 配置 Redis/SQL 后端，但工具以非 FunctionTool 注入方式（Client 直调等）执行记忆读写。 实际影响： 记忆读写静默落到本地磁盘而非配置的后端，数据分叉且无法被其他节点访问；正常 FunctionTool 路径不受影响，属于配置错配的隐性降级。 修正方向： 较低
问题： 触发条件： 记忆条目数较大的会话每次请求读取索引/主题时。 实际影响： 每次读取产生 O(成员数) 次额外 Redis 往返，读放大明显，索引越大开销越高。 修正方向： 将 TTL 刷新降频（如批量 PIPELINE 一次发送、按固定窗口计时刷新），或在写入侧统一续期。 较低
问题： 触发条件： 长期运行的服务持续为不同 session 创建压缩状态。 实际影响： 内存随会话数单调增长直至 OOM；重启后全部状态丢失又与 content-reuse 场景叠加放大错误面。 修正方向： 为进程级 dict 增加基于最后访问时间的淘汰清理，并考虑将关键状态持久化到 Redis/SQL 以同时解决重启丢失。 |
| self._model = model | ||
| self._runtime = AdvancedAutoCompactSummarizerRuntime(config=config or AdvancedAutoCompactSummarizerConfig()) | ||
| self._auto_compact_config = config.auto_compact |
There was a problem hiding this comment.
问题： __init__ 签名标注 config: Optional[...] 且下方有 config or AdvancedAutoCompactSummarizerConfig() 的兜底意图，但兜底结果未回写，紧接着的 self._auto_compact_config = config.auto_compact 直接解引用原始 config，传 None 时抛 AttributeError。
触发条件： 调用方以 AdvancedAutoCompactSummarizer() 无参或显式传 None 构造（签名与兜底分支均表明该用法被预期），例如依赖默认配置的接入路径。
实际影响： 对象构造即崩溃，advanced memory 筛选器初始化失败；config or ... 兜底分支成为不可达死代码，默认配置路径整体不可用。
修正方向： 解引用前先归一化，例如 config = config or AdvancedAutoCompactSummarizerConfig()，再访问 config.auto_compact 等字段。
| if state.consecutive_failures >= auto_compact_config.max_failures and blocking_reached: | ||
| return AdvancedAutoCompactResult( | ||
| compacted=False, | ||
| reapplied=reapplied, | ||
| blocked=True, | ||
| source=None, | ||
| request_chars_before=request_chars_before, | ||
| request_chars_after=request_chars_before, | ||
| consecutive_failures=state.consecutive_failures, | ||
| ) | ||
| if state.consecutive_failures >= auto_compact_config.max_failures: | ||
| return AdvancedAutoCompactResult( | ||
| compacted=False, | ||
| reapplied=reapplied, | ||
| blocked=False, | ||
| source=None, | ||
| request_chars_before=request_chars_before, | ||
| request_chars_after=request_chars_before, | ||
| consecutive_failures=state.consecutive_failures, | ||
| ) |
There was a problem hiding this comment.
问题： 断路器早退分支在 try 块之外（657-676 行），成功路径中唯一的重置点 state.consecutive_failures = 0（760 行）位于 try 成功分支内；一旦 consecutive_failures 达到 max_failures，后续请求无论成功与否都直接走早退返回，计数永远不会被清零，进程内永久锁死。
触发条件： 累计 N 次压缩失败（如 LLM 不可用、压缩未降体积）达到 max_failures 后，条件恢复、后续压缩全部成功时。
实际影响： 早退为 blocked 的请求携带错误摘要直接放行超大上下文（信息截断语义被破坏），早退为非 blocked 的请求则永久跳过压缩；断路器没有冷却时间或半开探测，只能靠重启进程恢复（计数仅存内存）。
修正方向： 在早退分支发现失败计数已达上限但本次请求按当前指标已无需压缩（非 blocking）时允许重置计数，或在成功（未失败）路径的 try 外也执行计数清零 / 引入基于时间窗的半开重试机制。
| async with self._session_lock(session.id): | ||
| request.contents = [content.model_copy(deep=True) for content in request.contents] | ||
| state = await self._load_state(session_id) | ||
| groups = self._collect_candidates(request) | ||
| state = await self._load_state(session.id) | ||
| groups = self._collect_candidates(request, session) | ||
| for group in groups: | ||
| for candidate in group: | ||
| known_hash = state.result_hashes.get(candidate.result_id) | ||
| current_hash = tool_result_sha256(candidate.serialized_result) | ||
| if known_hash is not None and known_hash != current_hash: | ||
| raise ValueError(f"Tool result id {candidate.result_id!r} is reused with different content") |
There was a problem hiding this comment.
问题： stable_tool_result_id 基于 tool_call id 生成，_states 仅存进程内存且旧版持久化已删除；进程重启后同一 tool_call id 关联到不同的工具结果内容时，_apply_scoped 抛出 ValueError("Tool result id ... reused with different content") 并沿 handle 向上传播（_history_snip.py 185-195、_micro_compact.py 185-192 同模式）。
触发条件： 服务重启后，会话上下文中再次出现与重启前相同 id 的 tool_call（工具重放、请求重试、多 worker 共享会话等场景）。
实际影响： 异常经筛选器 _handle_co 的 except 分支转为 FilterResult(error=..., is_continue=False) 再 yield None 事件，模型调用被错误化或直接失败；跨重启的记忆状态（含 latest_compaction）全部丢失，压缩行为与重启前不一致。
修正方向： 对 tool_call id 附加时间/会话代际维度使其跨重启唯一，或对 content 不一致的场景降级为忽略旧记录而不是抛异常；同时考虑把关键状态落盘（Redis/SQL）而非仅存内存。
| def record_request_context( | ||
| self, | ||
| request: "LlmRequest", | ||
| ctx: "InvocationContext | None", | ||
| cls, | ||
| request: LlmRequest, | ||
| ctx: InvocationContext, | ||
| ) -> None: | ||
| """Stage the final request fingerprint for persistence on the response Event.""" | ||
| session = getattr(ctx, "session", None) if ctx is not None else None | ||
| state = getattr(session, "state", None) | ||
| session = ctx.session | ||
| state = session.state | ||
| if isinstance(state, dict): | ||
| state["advanced_memory_pending_request_context_fingerprint"] = _request_static_fingerprint(request) |
There was a problem hiding this comment.
问题： record_request_context 把指纹写入 session.state["advanced_memory_pending_request_context_fingerprint"]（242-251 行），但本变更已删除旧版 _session_service 中将该 pending 指纹搬运到 event custom_metadata 的消费者逻辑；读取侧 _latest_usage_baseline（178 行）只从 event.custom_metadata 校验指纹，二者断链。
触发条件： 任意携带上轮 usage 统计的请求继续进入 token 估算流程；session.state 中残留的 pending 指纹字段也会被持久化。
实际影响： 指纹校验永不生效，usage baseline 会跨模型、跨配置地被复用，token 预算估算失真（该压缩时未压缩、该放行时误阻断）；session.state 被无关字段污染并随会话持久化。
修正方向： 恢复 pending 指纹到 event 元数据的传递链路，或在 _latest_usage_baseline 中直接读取 session.state 中的 pending 指纹并消费后清理，保证指纹写入、消费、清理形成闭环。
| invocation_ctx.end_invocation = True | ||
| rsp.rsp = result | ||
| rsp.is_continue = False | ||
| rsp.error = None | ||
| session_memory_extractor = summarizer_manager.get_session_memory_extractor() | ||
| if session_memory_extractor is not None: | ||
| await session_memory_extractor.extract_if_needed( |
There was a problem hiding this comment.
问题： _before 设置 rsp.rsp = result / rsp.is_continue = False / rsp.end_invocation = True 表达阻断语义，但该阻断信息只存在于本次 _before 的局部返回值链上；_base_filter.py 154-160 的 except 分支 yield FilterResult(error=ex, is_continue=False) 经 _run_filter.py:57 yield event.rsp 产出 None rsp，上游 call_llm_async 对 None 调用 has_content()/is_error() 抛 AttributeError。
触发条件： 任意 handler（tool_result_budget、history_snip、micro_compact、auto_compact）抛异常（如 content-reuse ValueError、LLM 压缩失败），或模型调用本身失败时。
实际影响： 本应及时返回给用户的阻断响应（“内容超限已阻止发送”）在异常路径上被丢弃，改为向上抛错误事件/异常使会话失败；本应被阻止的超大请求在部分异常场景下反而继续发送。
修正方向： 在 except 路径保留已构造的 rsp.rsp 与阻断标志（异常仅记录到 error 字段），并确保 _run_filter.py/call_llm_async 对 None rsp 做防御处理而不是直接属性访问。
| @@ -381,24 +360,14 @@ def _event_records_after_checkpoint( | |||
| ) | |||
| return recovered | |||
| logger.warning( | |||
| "Session memory checkpoint %s is missing from transcript; " | |||
| "Session memory checkpoint %s is missing from active events; " | |||
| "skipping extraction to avoid replaying the full history", | |||
| checkpoint_event_id, | |||
| ) | |||
| return [] | |||
There was a problem hiding this comment.
问题： _event_records_after_checkpoint 在 checkpoint_id 不在 active events 中时回退到按时间戳恢复（要求事件 recorded_at 严格大于 checkpoint 的），但 checkpoint 事件被 compact_events 移入 historical_events 后，active events 中仅剩 summary 事件（时间戳被改写为被替换事件的首时间戳），时间戳恢复必然失败并返回空列表。
触发条件： checkpoint 落在已被压缩的事件区间内（长期运行会话的典型情况），随后再次调用 extract_if_needed。
实际影响： extract_if_needed 永久停在 no-new-events 分支，session memory 不再推进，_latest_session_memory_record 恒为 None，auto compact 回退 legacy 摘要路径，新增记忆持续丢失。
修正方向： 在 historical_events 中按 checkpoint 定位被替换的原始事件并以其时间戳作为恢复基准，或在提取完成前不压缩 checkpoint 所在区间，保证 checkpoint 始终可解析。
| async def read_index(self) -> str: | ||
| key = f"{self._user_base}:memory:index" | ||
| value = self._text(await self._command("get", key)) or "" | ||
| await self._refresh_memory_ttl() | ||
| index = value | ||
| valid_filenames = set() | ||
| for entry in parse_memory_index(index): | ||
| topic_key = f"{self._user_base}:memory:topic:{self._topic_name(entry.filename)}" | ||
| if await self._command("exists", topic_key): | ||
| valid_filenames.add(entry.filename) | ||
| pruned_index = prune_memory_index(index, valid_filenames) | ||
| if pruned_index != index: | ||
| async with self._memory_write_lock(): | ||
| await self._command("set", key, pruned_index) | ||
| await self._refresh_memory_ttl(key) | ||
| return limit_memory_index( | ||
| pruned_index, |
There was a problem hiding this comment.
问题： read_index 的 prune 路径（147-151 行）在 _memory_write_lock 外直接 set 清理后的索引，与持有同一把锁的 write_index（164-167 行，读-改-写）并发执行时产生丢失更新。
触发条件： 两个及以上进程/线程同时操作同一会话记忆索引（prune 与 save_memory 交错）。
实际影响： 刚保存的记忆条目可能被并发 prune 覆盖丢失，或 prune 结果被旧快照覆盖导致索引膨胀，记忆索引与条目不一致。
修正方向： 将 prune 的读取、计算、写入整体移入 _memory_write_lock 临界区（或改用 Lua 脚本原子执行 prune+set）。
| async def update_session_state( | ||
| self, | ||
| session: Session, | ||
| state_delta: dict[str, Any], | ||
| ) -> None: | ||
| """Persist session-scoped state without rewriting Event rows.""" | ||
| if not state_delta: | ||
| return | ||
| session.state.update(state_delta) | ||
|
|
||
| async with self._sql_storage.create_db_session() as sql_session: | ||
| session_key = SqlKey( | ||
| key=(session.app_name, session.user_id, session.id), | ||
| storage_cls=StorageSession, | ||
| ) | ||
| storage_session: Optional[StorageSession] = await self._sql_storage.get_for_update( | ||
| sql_session, | ||
| session_key, | ||
| ) | ||
| if storage_session is None: | ||
| logger.warning( | ||
| "Session %s not found in storage while updating state", | ||
| session.id, | ||
| ) | ||
| return | ||
|
|
||
| persisted_state = dict(storage_session.state or {}) | ||
| persisted_state.update(state_delta) | ||
| storage_session.state = persisted_state # type: ignore | ||
| await self._sql_storage.commit(sql_session) | ||
| await self._sql_storage.refresh(sql_session, storage_session) | ||
| session.last_update_time = storage_session.update_timestamp_tz | ||
|
|
||
| @override | ||
| async def update_session(self, session: Session) -> None: |
There was a problem hiding this comment.
问题： update_session_state 仅写 state 列并 commit/refresh，不触碰 onupdate=func.now() 的 update_time 列（159 行），而 TTL 过期判定基于 update_time + ttl_seconds（513/776 行）；内存实现与 Redis 后端在每次状态更新时都会刷新过期时间，SQL 后端行为不一致。
触发条件： 会话仅通过 update_session_state 更新（如 checkpoint 持久化路径），期间无 append_event/get_session 等其他触达 update_time 的操作，且会话空闲时间超过 ttl_seconds。
实际影响： SQL 后端下活跃会话（持续写 state 但无 event 追加）被判定过期并清理，checkpoint/记忆状态被提前逐出，与内存/Redis 后端表现不一致。
修正方向： 在 update_session_state 中显式刷新 update_time（赋值 func.now() 或单独 update 该列），使三种后端 TTL 语义一致。
| original_contents = [content.model_copy(deep=True) for content in request.contents] | ||
| try: | ||
| compact_record: AdvancedAutoCompactRecord | None = None | ||
| if self._session_memory_extractor is not None: | ||
| await self._session_memory_extractor.extract_if_needed( | ||
| ctx, | ||
| force=True, | ||
| ) |
There was a problem hiding this comment.
问题： _apply_scoped 在 self._session_lock(session_id) 临界区内调用 extract_if_needed(force=True)，而 extract_if_needed 内部在 coordination guard 内执行 _call_llm_model(...) 的 generate_async(stream=False)（秒级）+ 重试（_compaction_memory_extractor.py 698-705 等）。
触发条件： 同一 session 的多个请求并发到达且任一请求触发迫近 blocking 的压缩路径。
实际影响： 后续同会话请求在锁上串行等待秒级 LLM 压缩，请求延迟被成倍放大，高并发下形成排队拥塞，加剧 TokenBudget 超阈值。
修正方向： 将 LLM 压缩调用移出锁外（锁仅保护状态读改写），或在锁外预取 session memory 记录、锁内仅做纯计算。
| except Exception as exc: # noqa: BLE001 | ||
| request.contents = original_contents | ||
| state.consecutive_failures += 1 | ||
| blocked = state.consecutive_failures >= auto_compact_config.max_failures and blocking_reached | ||
| return AdvancedAutoCompactResult( | ||
| compacted=False, | ||
| reapplied=reapplied, | ||
| blocked=blocked, | ||
| source=None, | ||
| request_chars_before=request_chars_before, | ||
| request_chars_after=request_chars_before, | ||
| consecutive_failures=state.consecutive_failures, | ||
| error=str(exc) if exc else None if blocked else None, | ||
| request_tokens_before=request_tokens_before if token_mode else None, | ||
| request_tokens_after=comparison_tokens_before if token_mode else None, | ||
| token_source=token_budget_before.estimate.source if token_mode else None, | ||
| ) |
There was a problem hiding this comment.
问题： 786 行 error=str(exc) if exc else None if blocked else None 为右结合三元，except Exception as exc 中 exc 恒为 truthy，表达式恒等于 str(exc)；而断路器早退分支（657-676 行）构造的 blocked 结果 error 恒为 None。
触发条件： 断路器触发 blocked 早退（达到 max_failures 且请求超限）时，调用方在结果中查询失败原因。
实际影响： 调用方（筛选器/上层）无法从结果感知熔断原因，只能靠日志排查；blocked 结果与正常结果的 error 字段区分度为零。
修正方向： 早退分支把当前失败计数/最近一次异常写入 error 字段（如 f"circuit open after {state.consecutive_failures} failures"），并简化 786 行为 error=str(exc)。
| long_term_memory=LongTermMemoryStore(resolved_config, paths), | ||
| _redis_storage=redis_storage, | ||
| _sql_storage=sql_storage, | ||
| _sql_cleanup=sql_cleanup, | ||
| _local_cleanup=local_cleanup, | ||
| ) |
There was a problem hiding this comment.
问题： 根 runtime 的 create() 恒构造本地 LongTermMemoryStore（79 行），只有 FunctionTool 注入 tool_context 的路径才会走 scoped runtime；_advanced_memory_tool.py 62-67 行在 tool_context=None 时回退到 root runtime。
触发条件： 配置 Redis/SQL 后端，但工具以非 FunctionTool 注入方式（Client 直调等）执行记忆读写。
实际影响： 记忆读写静默落到本地磁盘而非配置的后端，数据分叉且无法被其他节点访问；正常 FunctionTool 路径不受影响，属于配置错配的隐性降级。
修正方向： create() 按全局配置选择后端，或在 _runtime_for_context 的 None 分支显式按配置构造 scoped runtime 并告警。
| async def _refresh_ttl_group( | ||
| self, | ||
| registry: str, | ||
| keys: list[str], | ||
| ttl: int | None, | ||
| skip_prefixes: tuple[str, ...] = (), | ||
| ) -> None: | ||
| """Track and refresh every key in one logical memory group.""" | ||
| if ttl is None: | ||
| return | ||
| if keys: | ||
| await self._command("sadd", registry, *keys) | ||
| tracked = await self._command("smembers", registry) or [] | ||
| tracked_keys = {self._text(value) for value in tracked} | ||
| tracked_keys.update(keys) | ||
| for key in tracked_keys: | ||
| if key and not key.startswith(skip_prefixes): | ||
| await self._command("expire", key, ttl) | ||
| await self._command("expire", registry, ttl) |
There was a problem hiding this comment.
问题： _refresh_ttl_group 在每次 read_index/read_topic 时都执行 SMEMBERS 全量取成员并对每个 key 逐一 EXPIRE。
触发条件： 记忆条目数较大的会话每次请求读取索引/主题时。
实际影响： 每次读取产生 O(成员数) 次额外 Redis 往返，读放大明显，索引越大开销越高。
修正方向： 将 TTL 刷新降频（如批量 PIPELINE 一次发送、按固定窗口计时刷新），或在写入侧统一续期。
| async def _load_state(self, session_id: str) -> AdvancedAutoCompactState: | ||
| """Restore process-local compaction state.""" | ||
| state_key = self._runtime.session_key(session_id) | ||
| state = self._states.get(state_key) | ||
| if state is not None: | ||
| return state | ||
| state = AdvancedAutoCompactState(latest_compaction=None, consecutive_failures=0) | ||
| self._states[state_key] = state | ||
| return state |
There was a problem hiding this comment.
问题： _states、_session_locks（本文件）与 coordination._locks、_scoped_processors、_scoped_runtimes、_index_locks（_runtime.py/_redis_stores.py 等）等全部进程级 dict 只增不删，没有任何 pop/clear/discard 路径，且 _states 仅存内存。
触发条件： 长期运行的服务持续为不同 session 创建压缩状态。
实际影响： 内存随会话数单调增长直至 OOM；重启后全部状态丢失又与 content-reuse 场景叠加放大错误面。
修正方向： 为进程级 dict 增加基于最后访问时间的淘汰清理，并考虑将关键状态持久化到 Redis/SQL 以同时解决重启丢失。
AI Code Review审查结论通过 审查范围 计划符合性:重构按计划完成—— 主要风险:3 个真实缺陷——1) 测试: 门禁结论:通过(REVIEW_CODE_STATE_PASSED),建议修复 3 个中低严重度问题后再合入。 发现的问题中等
问题： 触发条件： 任何以 实际影响： 构造即抛 修正方向： 统一从已兜底的 runtime 取值: 较低
问题： 触发条件： 用户将 实际影响： 使用 scoped runtime 的集成在主路径上工具调用崩溃,且错误信息无法提示用户改用 root runtime。 修正方向： 在 中等
问题： 新增的 触发条件： 使用 实际影响： 会话记忆 checkpoint 静默丢失：重启后 修正方向： 在 |
| def __init__( | ||
| self, | ||
| config: AdvancedAutoCompactSummarizerConfig | None = None, | ||
| *, | ||
| model: LLMModel | None = None, | ||
| session_memory_extractor: SessionMemoryExtractor | None = None, | ||
| ) -> None: | ||
| """Initialize the compressor, summary generator, and session locks.""" | ||
| self._model = model | ||
| self._runtime = AdvancedAutoCompactSummarizerRuntime(config=config or AdvancedAutoCompactSummarizerConfig()) | ||
| self._auto_compact_config = config.auto_compact |
There was a problem hiding this comment.
问题： __init__ 声明 config: AdvancedAutoCompactSummarizerConfig | None = None 并在第 127 行用 config or AdvancedAutoCompactSummarizerConfig() 兜底构造了 runtime,但第 128 行 self._auto_compact_config = config.auto_compact 直接对原始 config 对象本身即未做 None 判空,None 直接抛 AttributeError`。
触发条件： 任何以 AdvancedAutoCompactSummarizer() 或 AdvancedAutoCompactSummarizer(None) 构造(默认参数合法公开 API,config=None 是签名允许的值)都会在实例化时崩溃;_runtime 已正确 fallback 而 _auto_compact_config 未 fallback,两者行为不一致。
实际影响： 构造即抛 AttributeError: 'NoneType' object has no attribute 'auto_compact',服务启动失败;已有的 tests/sessions/compact/test_session_compact.py 均传入完整配置所以未暴露此缺口。
修正方向： 统一从已兜底的 runtime 取值:self._auto_compact_config = self._runtime.config.auto_compact(或 (config or AdvancedAutoCompactSummarizerConfig()).auto_compact),并补充一个 config=None 的构造测试。
| def _runtime_for_context(self, tool_context: Any | None) -> Any: | ||
| """Resolve storage from the authenticated session, never tool arguments.""" | ||
| if tool_context is None: | ||
| return self._runtime | ||
| session = getattr(tool_context, "session", None) | ||
| return self._runtime.for_session(session) |
There was a problem hiding this comment.
问题： _runtime_for_context 在 tool_context 携带 session 时无条件调用 self._runtime.for_session(session),但 ScopedAdvancedMemoryRuntime(由 AdvancedMemoryRuntime.for_scope() 返回且公开导出)没有 for_session 方法,只有 root runtime 具备。
触发条件： 用户将 AdvancedMemoryToolSet(scoped_runtime) 或 create_advanced_memory_toolset(scoped_runtime) 装入 agent 后,任何携带 session 的工具调用(save_memory/read_memory/list_memory_index)都会抛 AttributeError: 'ScopedAdvancedMemoryRuntime' object has no attribute 'for_session';tool_context=None 时反而正常返回 scoped runtime,行为不一致,SESSION 级工具在 scoped 模式下直接不可用。
实际影响： 使用 scoped runtime 的集成在主路径上工具调用崩溃,且错误信息无法提示用户改用 root runtime。
修正方向： 在 _runtime_for_context 中先判断 hasattr(self._runtime, "for_session"),缺失时直接返回 self._runtime(scoped 已按租户隔离,无需再按 session 细分),并补一个 scoped runtime 的工具调用测试。
| async def update_session_state( | ||
| self, | ||
| session: Session, | ||
| state_delta: dict[str, Any], | ||
| ) -> None: | ||
| """Patch stored session state without replacing its Event window.""" | ||
| if not state_delta: | ||
| return | ||
| session.state.update(state_delta) | ||
|
|
||
| app_sessions = self._sessions.get(session.app_name) | ||
| user_sessions = app_sessions.get(session.user_id) if app_sessions else None | ||
| stored = user_sessions.get(session.id) if user_sessions else None | ||
| if stored is None: |
There was a problem hiding this comment.
问题： 新增的 InMemorySessionService.update_session_state 只更新内存缓存 _sessions 与调用方 session.state,不调用 update_session 持久化钩子。ClawSessionService 继承 InMemorySessionService 但通过覆写 update_session 将会话保存到磁盘 JSONL;任何走 update_session_state 的状态增量(session memory checkpoint)完成后既不在缓存对应的磁盘文件里,注入到 _sessions 缓存的状态也永远不会触发 save_session。
触发条件： 使用 ClawSessionService(openclaw 服务端)且启用 advanced compact 的 session memory 提取时,_persist_checkpoint 每次经 update_session_state 写 checkpoint;进程重启或缓存逐出后从磁盘加载的 Session 丢失 SESSION_MEMORY_STATE_KEY。
实际影响： 会话记忆 checkpoint 静默丢失：重启后 _latest_session_memory_record 读不到旧 checkpoint,已压缩的历史事件被当作新事件重新提取,边界签名失效,压缩边界从 active window 中重新选取,可能重复压缩已归档内容;与 update_session(会持久化)路径行为不一致。
修正方向： 在 InMemorySessionService.update_session_state 更新缓存后调用 await self.update_session(session)(或在 ClawSessionService 覆写 update_session_state 并追加磁盘保存),确保状态增量与 update_session 一致落盘。
AI Code Review审查结论不通过 审查范围：base f05797d..head f26d550（Feature/advanced memory 重构，8 个提交、111 个文件、+6191/-6574），重点核对高级自动压缩（AdvancedAutoCompactSummarizer）、令牌预算（TokenContextTracker）、会话记忆抽取与 Redis/SQL/内存三种会话服务的状态持久化、兼容别名层及新增测试。 计划符合性：总体方向符合计划（advanced memory 与上下文解耦、压缩持久化、重试幂等、命令行兼容别名），主要模块均有实现与配套测试。 主要风险：发现 2 个中等级别缺陷——(1) 令牌预算的请求上下文指纹链断裂：写入端与消费端 key 名称不一致，真实请求事件上从不携带指纹字段，usage 基线退化为纯内容匹配，预算评估失真；(2) 压缩幂等契约不可达：compaction_id 每次压缩生成新 UUID，重试永不命中 compact_events 的幂等早返回，持久化失败重试会重复压缩并污染历史。另有 2 个低级别问题：Redis 会话状态为整包读改写覆写、并发窗口内可能丢失事件；压缩失败路径与幂等行为缺少针对新增逻辑的测试覆盖。 测试充分性：新增测试覆盖了状态持久化、历史事件保留、指纹手工 fixture 等主路径，但端到端指纹链路、重试幂等、边界映射失败等异常路径缺失。 门禁结论：存在中等级别功能缺陷，建议修复后合入。 发现的问题中等
问题： 触发条件： 每次 实际影响： 指纹守卫形同虚设，usage 基线匹配实际退化为纯 修正方向： 在响应 Event 落库处（如 Runner 生成 Response 事件的位置）读取 state 中的 pending 指纹，写入该 Event 的 中等
问题： 触发条件： 实际影响： 违反 修正方向： 将压缩 ID 改为基于会话与边界的确定性派生（例如以 较低
问题： 新增的 触发条件： 同一会话存在并发写入时（例如 deferred 后台线程写入长期记忆与主流程追加事件交错、或多个进程共享同一 Redis 实例），两个读改写窗口重叠。 实际影响： 竞态窗口内先完成的写入（如新追加的 Event 或状态字段）会被后完成的整包覆写覆盖丢失，表现为偶发的事件丢失或状态回退，会话事件历史在并发下不稳定。 修正方向： 对 state 做增量写入（如用 Redis 哈希或 Lua 脚本保证读改写原子性），或对整包覆写加乐观版本号校验并重试冲突写入；并在并发写入场景补充测试。 较低
问题： 触发条件： 会话事件被部分清理（如历史全量清空或事件被外部删除）导致 实际影响： 容错与幂等行为缺乏回归保障，后续改动容易悄然破坏「映射失败→重试→阻断」的降级链而不被发现；高级压缩在异常路径上的行为会变得不可预期。 修正方向： 为边界映射失败、 |
| @classmethod | ||
| def record_request_context( | ||
| self, | ||
| request: "LlmRequest", | ||
| ctx: "InvocationContext | None", | ||
| cls, | ||
| request: LlmRequest, | ||
| ctx: InvocationContext, | ||
| ) -> None: | ||
| """Stage the final request fingerprint for persistence on the response Event.""" | ||
| session = getattr(ctx, "session", None) if ctx is not None else None | ||
| state = getattr(session, "state", None) | ||
| session = ctx.session | ||
| state = session.state | ||
| if isinstance(state, dict): | ||
| state["advanced_memory_pending_request_context_fingerprint"] = _request_static_fingerprint(request) |
There was a problem hiding this comment.
问题： record_request_context 把 advanced_memory_pending_request_context_fingerprint 写入 session.state，但整条链路上没有任何代码把这枚指纹转存到响应 Event 的 custom_metadata。全仓搜索确认：消费端读取的 advanced_memory_request_context_fingerprint 除测试手工构造的 fixture 外，再无任何生产代码写入；两枚 key 一写一读、名称不同，永不交汇。
触发条件： 每次 create_session_summary_by_request 未被阻断时都会执行该状态写入；下一次请求做预算评估时，_latest_usage_baseline（约 160-193 行）遍历已有 Event 的 custom_metadata，metadata.get("advanced_memory_request_context_fingerprint") 对真实请求产生的 Event 恒为 None，指纹比对分支永远不排除任何事件。
实际影响： 指纹守卫形同虚设，usage 基线匹配实际退化为纯 _content_fingerprint 内容匹配。请求内容一旦因拼接、裁剪或改写而变化，基线匹配失败并回退到启发式估算，令牌预算失真，可能更早触发 auto-compact 或将本可避免的请求误判为需要阻塞，使 advanced memory 的令牌预算与压缩行为偏离设计预期。
修正方向： 在响应 Event 落库处（如 Runner 生成 Response 事件的位置）读取 state 中的 pending 指纹，写入该 Event 的 custom_metadata["advanced_memory_request_context_fingerprint"] 并清除待处理字段；或删除该状态写入、保留确定性校验，并为真实写入链路补充端到端测试。
| return AdvancedAutoCompactRecord( | ||
| boundary_signature, | ||
| boundary_occurrence, | ||
| summary, | ||
| source, | ||
| boundary_event_id, | ||
| f"advanced_autocompact:{uuid.uuid4().hex}", | ||
| ) |
There was a problem hiding this comment.
问题： _compact_with_summary 每次压缩都为 record.compaction_id 生成全新的 advanced_autocompact:{uuid.uuid4().hex}，而 compact_events（_session.py 约 140-186 行）的幂等早返回依据 event.custom_metadata.get("session_compaction_id") == compaction_id 精确匹配整个 ID。新生成的 UUID 永远无法与已持久化 Summary Event 上记录的旧 ID 相等，幂等保护实际不可达。
触发条件： _persist_session_compaction 在 compact_events 已修改 ctx.session.events 后调用 update_session 失败并抛异常，随后对同一请求重试，或后续请求再次命中同一边界进行压缩时，重复执行 compact_events，幂等早返回永不触发。
实际影响： 违反 compact_events 声明的"compaction_id 使重试幂等"契约：持久化成功但调用方未观测到结果时，重试会再次压缩并把又一份 Summary Event 追加进 historical_events，历史记录出现重复压缩产物并随时间累积，长期污染会话记忆。
修正方向： 将压缩 ID 改为基于会话与边界的确定性派生（例如以 boundary_event_id 或边界签名为稳定输入），使同一边界的重试能命中幂等早返回；并为「持久化失败后重试不产生重复摘要」补充单元测试。
| async def update_session_state( | ||
| self, | ||
| session: Session, | ||
| state_delta: dict[str, Any], | ||
| ) -> None: | ||
| """Persist session-scoped state without replacing the caller's Event window.""" | ||
| if not state_delta: | ||
| return | ||
| session.state.update(state_delta) | ||
|
|
||
| async with self._redis_storage.create_db_session() as redis_session: | ||
| key = session_key(session.app_name, session.user_id, session.id) | ||
| storage_session = await self._get_session(redis_session, key) | ||
| if not storage_session: | ||
| logger.warning( | ||
| "Session %s not found in Redis while updating state", | ||
| session.id, | ||
| ) | ||
| return | ||
| storage_session.state.update(state_delta) | ||
| await self._set_session(redis_session, storage_session) |
There was a problem hiding this comment.
问题： 新增的 update_session_state（254-274 行）在 Redis 后端按「读取完整会话 JSON → 更新 state → 整包覆写」实现，_set_session 用 SET 全量替换会话 JSON；同会话的其他并发写入（如 append_event、update_session 或后台记忆写入线程）也走同一完整覆写路径，后写覆盖先写，无合并、无 CAS 保护。
触发条件： 同一会话存在并发写入时（例如 deferred 后台线程写入长期记忆与主流程追加事件交错、或多个进程共享同一 Redis 实例），两个读改写窗口重叠。
实际影响： 竞态窗口内先完成的写入（如新追加的 Event 或状态字段）会被后完成的整包覆写覆盖丢失，表现为偶发的事件丢失或状态回退，会话事件历史在并发下不稳定。
修正方向： 对 state 做增量写入（如用 Redis 哈希或 Lua 脚本保证读改写原子性），或对整包覆写加乐观版本号校验并重试冲突写入；并在并发写入场景补充测试。
| boundary_event_id = record.boundary_event_id or self._resolve_boundary_event_id( | ||
| ctx, | ||
| record.boundary_signature, | ||
| record.boundary_occurrence, | ||
| ) | ||
| if boundary_event_id is None: | ||
| raise ValueError("Cannot map the AdvancedAutoCompactSummarizer boundary to an active Session Event") | ||
|
|
||
| compaction_id = record.compaction_id or f"advanced_autocompact:{uuid.uuid4().hex}" | ||
| summary_event = Event( | ||
| invocation_id="summary", | ||
| author="system", | ||
| content=self._summary_content(record.summary), | ||
| custom_metadata={ | ||
| "session_compaction_source": record.source, | ||
| "session_compaction_boundary_signature": record.boundary_signature, | ||
| "session_compaction_boundary_occurrence": record.boundary_occurrence, | ||
| }, | ||
| ) | ||
| active_before = list(ctx.session.events) | ||
| historical_before = list(ctx.session.historical_events) | ||
| last_update_before = ctx.session.last_update_time | ||
| try: | ||
| changed = compact_events( | ||
| summary_event, | ||
| boundary_event_id, | ||
| compaction_id=compaction_id, | ||
| ) | ||
| if changed: | ||
| await ctx.session_service.update_session(ctx.session) | ||
| except Exception: | ||
| ctx.session.events = active_before | ||
| ctx.session.historical_events = historical_before | ||
| ctx.session.last_update_time = last_update_before | ||
| raise |
There was a problem hiding this comment.
问题： _persist_session_compaction 的失败路径与边界映射是本变更引入的关键容错逻辑（420-454 行），但本变更新增的测试未覆盖：边界无法映射时抛出 ValueError 并计入 consecutive_failures、达到上限后请求被阻断、以及 compact_events 幂等早返回路径，均无对应测试。
触发条件： 会话事件被部分清理（如历史全量清空或事件被外部删除）导致 _resolve_boundary_event_id 在严格的“签名+出现次数”匹配下返回 None 时，生产代码会触发 raise；该路径没有测试保护。
实际影响： 容错与幂等行为缺乏回归保障，后续改动容易悄然破坏「映射失败→重试→阻断」的降级链而不被发现；高级压缩在异常路径上的行为会变得不可预期。
修正方向： 为边界映射失败、consecutive_failures 达到 max_failures 阻断、以及同 compaction_id 重复调用返回 False 三条路径补充单元测试。
Advanced Memory 服务化
解耦 advanced compact 和 advanced long-term memory
支持 redis & sql