Feature/advanced memory service2 - #330
Conversation
Please enter the commit message for your changes. Lines starting
AI Code Review审查结论通过 审查范围：base_commit f05797d..head_commit 3acb32a（Feature/advanced memory service2，98 个文件，+4746/-4880）。本次变更实现 Advanced Memory 服务化（本地/Redis/SQL 三种 storage_backend）、Session Compact 新机制（分级 before_model_callback、compact_events 事件压缩与幂等）、后端状态补丁（patch_session_state）、TokenContextTracker 用量基线、分布式写锁与 TTL 注册表、延迟 post-turn 处理线程等。计划符合性：主体功能均已实现并有对应测试（tests/sessions 下新增 patch/compact/token_budget 覆盖主路径，SQL/Redis 后端接口齐全）；但存在四处高置信缺陷：示例配置传参错误导致旗舰示例启动即崩溃、SQL 长期记忆索引过期后永久为空且与主题 TTL 状态发散、指纹防陈旧机制为半成品（生产者写入的 key 与消费者读取的 key 不一致）、压缩持久化经 update_session 全量重写在跨进程并发下可能丢失事件。测试充分性：新增测试覆盖了单进程主路径与指纹读取路径（test_token_budget.py 显式断言了陈旧指纹跳过逻辑），但删除约 2000 行旧测试后未完全等价补齐，且无 SQL/Redis 后端集成测试覆盖索引过期与跨进程压缩并发场景。主要风险集中在 SQL 后端长期记忆 TTL 一致性、示例可运行性与多用例部署下的压缩持久化竞争。门禁结论：无 SEVERE 级问题，state=PASSED；建议合入前修复示例参数错误并按注释修正方向补齐 SQL 索引过期处理。 发现的问题中等
问题： 旗舰示例 触发条件： 按示例文档运行该示例（本地/SQL 任一 storage_backend），构造 实际影响： 这是本特性对外演示的旗舰示例，启动即崩溃，用户无法运行长期记忆服务化的最小演示，直接影响新特性上手与验证。 修正方向： 删除第 37 行的 中等
问题： 触发条件： 索引行过期而某些主题行尚未过期（主题在过期前被再次写入或读取得到续期，索引因长期未被访问而先过期）时发生。 实际影响： 记忆索引返回永为空字符串：注入阶段读不到任何长期记忆条目，尽管主题数据仍存活且被持续续期；索引与主题的 TTL 状态发散且无收敛路径（SQL 清理任务只会删除真正过期的行）。表现为“内容还在但索引永久为空”的长期记忆失效，用户长期记忆无法被注入。 修正方向： 中等
问题： 触发条件： 任意一次模型请求后， 实际影响： 指纹防陈旧机制实际失效：当新请求内容与旧事件内容指纹恰好相同（如相同提示词重放）时，误用旧事件的 usage 作为基线，即使用户上下文已因其它会话操作发生变化；直接后果是 token 预算估计偏低，可能推迟触发 autocompact，导致模型输入超窗口被截断或拒绝。 修正方向： 统一指纹链路：将生产者改为在输出事件的 较低
问题： 触发条件： 多 worker 同时服务同一会话（或压缩持久化与新的 实际影响： 压缩瞬间丢失并发追加的对话事件，或压缩结果被并发写覆盖，历史记录不完整；该缺陷由本次引入的压缩持久化新路径首次可达（原 Summarizer 路径未涉及压缩期间事件覆盖）。 修正方向： 压缩持久化复用与 较低
问题： 触发条件： 使用自定义/未实现 实际影响： Session Memory 摘要永久无法持久化，且异常在 post-turn 异步路径（ 修正方向： |
| config = AdvancedMemoryServiceConfig( | ||
| root_dir=Path(__file__).resolve().parent, | ||
| memory_ttl_seconds=int(memory_ttl) if memory_ttl else None, | ||
| session_ttl_seconds=session_ttl_seconds or None, | ||
| memory_focus_instruction=("特别关注并主动记住用户长期稳定的兴趣爱好、" |
There was a problem hiding this comment.
问题： 旗舰示例 examples/memory_service_with_advanced_memory/run_agent.py 第 37 行向 AdvancedMemoryServiceConfig 构造函数传入了不存在的关键字参数 session_ttl_seconds。全仓库检索确认：_config.py 中 AdvancedMemoryServiceConfig 仅定义 memory_ttl_seconds 等字段，SDK 与示例中均无 session_ttl_seconds 字段；该配置真正归属处是第 47 行 _make_session_service 的 ttl_seconds=session_ttl_seconds。
触发条件： 按示例文档运行该示例（本地/SQL 任一 storage_backend），构造 AdvancedMemoryServiceConfig 即抛 TypeError: __init__() got an unexpected keyword argument 'session_ttl_seconds'。
实际影响： 这是本特性对外演示的旗舰示例，启动即崩溃，用户无法运行长期记忆服务化的最小演示，直接影响新特性上手与验证。
修正方向： 删除第 37 行的 session_ttl_seconds=... 参数（会话 TTL 已通过第 47 行 ttl_seconds 配置到 SessionService），或在 _config.py 中补充该字段并接线。
| async def read_index(self) -> str: | ||
| async with self._storage.create_db_session() as db: | ||
| row = await self._storage.get(db, SqlKey(key=(self._app_name, self._user_id), storage_cls=SqlMemoryIndex)) | ||
| if row is None or self._expired(row.expires_at): | ||
| return "" | ||
| await self._refresh_memory_scope(db) | ||
| await self._storage.commit(db) | ||
| content = row.content | ||
| lines, used_bytes = [], 0 | ||
| for line in content.splitlines(keepends=True)[:self._config.memory_index_max_lines]: | ||
| size = len(line.encode(self._config.encoding)) | ||
| if used_bytes + size > self._config.memory_index_max_bytes: | ||
| break | ||
| lines.append(line) | ||
| used_bytes += size | ||
| return "".join(lines) | ||
|
|
||
| async def write_index(self, entries: list[MemoryIndexEntry]) -> None: | ||
| content = "\n".join(entry.to_markdown() for entry in entries) | ||
| if content: | ||
| content += "\n" | ||
| async with self._storage.create_db_session() as db: |
There was a problem hiding this comment.
问题： SqlLongTermMemoryStore.read_index（_sql_stores.py:126-141）在索引行不存在或已过期时直接 return ""，既不删除过期行也不回写，跳过了刷新逻辑；而 read_topic/list_topics/write_topic/write_index 每次访问都会把该作用域内所有未过期主题行的 expires_at 刷新为 now + ttl。
触发条件： 索引行过期而某些主题行尚未过期（主题在过期前被再次写入或读取得到续期，索引因长期未被访问而先过期）时发生。
实际影响： 记忆索引返回永为空字符串：注入阶段读不到任何长期记忆条目，尽管主题数据仍存活且被持续续期；索引与主题的 TTL 状态发散且无收敛路径（SQL 清理任务只会删除真正过期的行）。表现为“内容还在但索引永久为空”的长期记忆失效，用户长期记忆无法被注入。
修正方向： read_index 遇过期索引时回写给空内容并刷新 expires_at（与 write_index 的续期语义一致），或直接删除过期索引行并触发 _refresh_memory_scope。
| def set_session_service( | ||
| self, | ||
| session_service: "SessionServiceABC", | ||
| force: bool = False, | ||
| ) -> None: | ||
| """Bind the manager to the original persistence service.""" | ||
| if self._session_service is not None and self._session_service is not session_service and not force: | ||
| raise ValueError("AdvancedSessionCompactManager is already bound to another SessionService") | ||
| session_config = getattr(session_service, "session_config", None) | ||
| if session_config is None or not getattr(session_config, "store_historical_events", False): | ||
| raise ValueError("Advanced Session Compact requires " | ||
| "SessionServiceConfig(store_historical_events=True)") | ||
| self._session_service = session_service | ||
| if self._session_memory_extractor is not None: | ||
| self._session_memory_extractor.attach_session_service(session_service) | ||
|
|
There was a problem hiding this comment.
问题： AdvancedSessionCompactManager.set_session_service（_manager.py:97-112）只校验 store_historical_events 配置，不校验目标 SessionService 是否实现 patch_session_state（_session_service.py:128-138 默认 raise NotImplementedError）；SessionMemoryExtractor._persist_checkpoint（_session_memory.py:736）在写入 _trpc_agent:summary 时不捕获该异常。
触发条件： 使用自定义/未实现 patch_session_state 的第三方 SessionService（仅实现 store_historical_events）时，首个 post-turn 的 create_session_summary 在关闭会话/定时任务中抛 NotImplementedError。
实际影响： Session Memory 摘要永久无法持久化，且异常在 post-turn 异步路径（_PostTurnWorkerThread）中被吞掉或打断关闭流程，用户感知不到原因；patch_session_state 的语义契约未在绑定期校验，缺陷延迟到运行时才暴露。
修正方向： set_session_service 增加 hasattr(session_service, "patch_session_state") 且实现非默认抛错的能力校验（或在 create_session_summary 中捕获 NotImplementedError 并降级为跳过摘要持久化），并在文档注明该能力要求。
3acb32a to
23ea179
Compare
AI Code Review审查结论通过 审查范围：base_commit f05797d 至 head_commit 23ea179，共 98 个变更文件（+4825/−4894，含 18 个测试文件），覆盖 Advanced Memory 服务化（local/redis/sql 三存储后端）、Session Compact 子系统（AutoCompact/HistorySnip/Microcompact/ToolResultBudget/TokenContextTracker）、会话服务层原子状态更新（patch_session_state）与并发协调（SessionOperationCoordinator/CrossLoopLock）。 计划符合性：核心需求均已实现且调用链可达——Runner.bind 依次调用 memory_service.bind(agent, session_service) 与 compact_manager.setup(agent)，staged callbacks 按 stage 5/10/20/30/40 注册进 before-model 回调链，工具侧通过 _runtime_for_context 解析 scoped runtime；compact_events 幂等归档与 checkpoint 持久化（_trpc_agent:summary）设计完备。 主要风险：5 个 MODERATE——(1) storage/_sql.py 的 get_for_update 无条件 FOR UPDATE，SQLite 后端行锁语义静默失效；(2) Redis 会话服务新增原子 patch_session_state 与整对象 update_session/append_event 并发时更新被覆盖丢失；(3) AutoCompact 边界解析失败抛异常并计入熔断，长期失败使压缩停摆；(4) 删除 AdvancedMemoryService/AdvancedMemorySessionService/AdvancedMemoryConfig 等旧公开名称且无兼容别名，破坏升级兼容；(5) SQL 存储首次写入时 SELECT FOR UPDATE 锁不到不存在的行，并发首写仍触发唯一约束冲突。另有 4 个 LOW：read_topic 缺 TTL 过期检查、无 scope 工具调用下 redis/sql storage_reference 抛错并产生部分写、scoped runtime 重复启动 SQL 清理循环、compact 子系统测试覆盖不足。 测试充分性：新增了 patch_session_state（三后端）、compact_events 幂等与 TokenContextTracker 等测试，但 AutoCompact 全链路、边界解析失败降级、熔断恢复及 stage 10/20/30 回调在真实 Runner 下的行为缺少测试；SQLite 测试环境无法暴露 FOR UPDATE 失效问题。 门禁结论：无 SEVERE 级别问题，state 为 PASSED。建议合入前修复 5 个 MODERATE 问题（尤其 Redis 写竞争与 SQL 首写互斥），LOW 项可随后续迭代处理。 发现的问题中等
问题： 新增的 触发条件： 后端为 SQLite（如 实际影响： "先锁行再修改"的互斥与原子性在 SQLite 上静默不生效，并发写入仍可发生唯一约束冲突或覆盖更新；且现有 SQLite 测试无法暴露该问题，缺陷仅在 MySQL/PostgreSQL 部署下可避免。 修正方向： 在 中等
问题： 本提交新增的原子 触发条件： 同一 session 上 实际影响： 内存检查点 修正方向： 将 中等
问题： 触发条件： 边界事件已被更早的 staged callback（history snip stage 20、microcompact stage 30）从 实际影响： AutoCompact 每次触发都失败并累加熔断计数，压缩长期无法执行，上下文持续膨胀直至触发阻塞阈值或压缩流程整体停摆。 修正方向： 边界解析失败时优雅降级（跳过本次压缩但不计入连续失败，或退化为不指定边界的压缩），而不是抛异常进入熔断路径。 中等
问题： 本提交删除了 触发条件： 既有用户代码按 1.1.20 及更早版本的方式执行 实际影响： 升级后模块 修正方向： 在 中等
问题： 触发条件： 同一 memory scope 首次持久化时发生并发写入（多协程或多进程共享同一 SQL 后端，且尚无任何 index 行）。 实际影响： 两个写入方同时通过"无行"检查并插入相同主键行，触发唯一约束冲突（ 修正方向： 先执行幂等 upsert（如 较低
问题： 触发条件： topic 内容已超过 TTL 存活期但清理任务尚未运行（或清理按 index 过期判定而 index 条目仍新鲜）时，读取该记忆项。 实际影响： 已过期记忆仍被读回并注入模型上下文，产生陈旧/失效记忆，与 local 后端"按 TTL 清理"的语义不一致。 修正方向： 在 较低
问题： 触发条件： 在未绑定 session scope 的上下文中调用高级记忆工具且后端为 redis/sql。 实际影响： 工具调用直接失败，或发生写入了错误作用域/错误后端后再报错的部分写，导致读写作用域不一致、数据落错位置。 修正方向： 在发生任何写入副作用之前校验 scope 可解析（redis/sql 后端要求 scope），未解析时先报错终止，或显式回退到根作用域的合法路径并保持一致。 较低
问题： 触发条件： 每个 scoped runtime（每 scope/每 session 创建）走 实际影响： SQL 清理循环被多个任务重复运行，造成冗余清理与资源浪费；若 修正方向： 将 较低
问题： 本次新增的 AutoCompact 全链路、边界解析（boundary signature/occurrence）、HistorySnip、Microcompact、TokenContextTracker 与 staged callback 排序等核心路径缺少针对性测试—— 触发条件： 上述路径在后续迭代中发生回归时。 实际影响： 已确认的边界解析失败熔断、回调顺序等缺陷无法在 CI 中暴露，回归风险高。 修正方向： 补充 AutoCompact 全链路测试（含边界事件被裁剪/归档后的失败降级）、熔断恢复测试及各 staged callback 按 stage 排序执行的断言。 |
| async def get_for_update(self, db: SqlSession, key: SqlKey) -> Any: | ||
| """Get one row while holding a database row lock until commit.""" | ||
| stmt = select(key.storage_cls) | ||
| for column, value in zip(inspect(key.storage_cls).primary_key, key.key): | ||
| stmt = stmt.where(column == value) | ||
| stmt = stmt.with_for_update() | ||
| if isinstance(db, AsyncSession): | ||
| result = await db.execute(stmt) | ||
| else: | ||
| result = db.execute(stmt) | ||
| return result.scalars().first() |
There was a problem hiding this comment.
问题： 新增的 get_for_update 无条件对查询附加 FOR UPDATE 行锁，未按数据库方言区分；SQLite 不支持该子句，行锁语义在后端为 SQLite 时直接失效。
触发条件： 后端为 SQLite（如 sqlite:///:memory:，测试与本地部署常见）时，任何调用 get_for_update 的路径被触发，包括 SqlSessionService.append_event/patch_session_state 以及 SqlLongTermMemoryStore.write_index/write_topic。
实际影响： "先锁行再修改"的互斥与原子性在 SQLite 上静默不生效，并发写入仍可发生唯一约束冲突或覆盖更新；且现有 SQLite 测试无法暴露该问题，缺陷仅在 MySQL/PostgreSQL 部署下可避免。
修正方向： 在 get_for_update 内判定方言（SQLite 时不附加 FOR UPDATE），对 SQLite 改用事务 + 唯一约束/冲突重试或应用级锁维持互斥语义，并补充 SQLite 下的并发写入测试。
| @override | ||
| async def patch_session_state( | ||
| self, | ||
| session: Session, | ||
| state_delta: dict[str, Any], | ||
| ) -> None: | ||
| """Atomically merge state while preserving concurrently written Events.""" | ||
| script = """ | ||
| local raw = redis.call('GET', KEYS[1]) | ||
| if not raw then | ||
| return false | ||
| end | ||
| local value = cjson.decode(raw) | ||
| local delta = cjson.decode(ARGV[1]) | ||
| if not value.state then | ||
| value.state = {} | ||
| end | ||
| for key, item in pairs(delta) do | ||
| value.state[key] = item | ||
| end | ||
| if type(value.events) == 'table' and next(value.events) == nil then | ||
| value.events = cjson.empty_array | ||
| end | ||
| if type(value.historical_events) == 'table' and next(value.historical_events) == nil then | ||
| value.historical_events = cjson.empty_array | ||
| end | ||
| if type(value.historicalEvents) == 'table' and next(value.historicalEvents) == nil then | ||
| value.historicalEvents = cjson.empty_array | ||
| end | ||
| local timestamp = tonumber(ARGV[2]) | ||
| if value.last_update_time ~= nil then | ||
| value.last_update_time = timestamp | ||
| end | ||
| if value.lastUpdateTime ~= nil then | ||
| value.lastUpdateTime = timestamp | ||
| end | ||
| local encoded = cjson.encode(value) | ||
| local ttl = tonumber(ARGV[3]) | ||
| if ttl > 0 then | ||
| redis.call('SET', KEYS[1], encoded, 'EX', ttl) | ||
| else | ||
| redis.call('SET', KEYS[1], encoded) | ||
| end | ||
| return encoded | ||
| """ | ||
| timestamp = time.time() | ||
| ttl = (int(self._session_config.ttl.ttl_seconds) if self._session_config.ttl.need_ttl_expire() else 0) | ||
| key = session_key(session.app_name, session.user_id, session.id) | ||
| async with self._redis_storage.create_db_session() as redis_session: | ||
| result = await self._redis_storage.execute_command( | ||
| redis_session, | ||
| RedisCommand( | ||
| method="eval", | ||
| args=( | ||
| script, | ||
| 1, | ||
| key, | ||
| json.dumps(state_delta, default=str), | ||
| timestamp, | ||
| ttl, | ||
| ), | ||
| ), | ||
| ) | ||
| if not result: | ||
| raise ValueError(f"Session {session.id} was not found") | ||
| stored_session = _session_from_storage_json(result) | ||
| session.state.update(state_delta) | ||
| session.last_update_time = stored_session.last_update_time | ||
|
|
There was a problem hiding this comment.
问题： 本提交新增的原子 patch_session_state（Lua 合并写）与同文件既有的非原子整对象 update_session/append_event（GET→SET 全量覆盖）构成两条并发写入路径：后者基于旧快照的整对象 SET 会覆盖前者刚刚合并写入的 state 变更。
触发条件： 同一 session 上 patch_session_state（记忆提取检查点、AutoCompact 落盘等回调）与 update_session/append_event（post-turn 处理、事件追加）并发执行，彼此的 GET 与 SET 交错。
实际影响： 内存检查点 session.state["_trpc_agent:summary"]、last_update_time 等更新被旧快照覆盖而丢失/回退，可能导致记忆重复提取、压缩记录错位；Lua 路径虽原子执行但结果随后被静默覆盖。
修正方向： 将 update_session/append_event 的状态写入也改为原子 Lua 合并（而非整对象覆盖），或引入版本号/条件更新（CAS）保证同一 key 只有一条原子写入路径。
| compact_events = getattr(ctx.session, "compact_events", None) | ||
| if not callable(compact_events): | ||
| # AutoCompact 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 AutoCompact boundary to an active Session Event") | ||
|
|
There was a problem hiding this comment.
问题： _persist_session_compaction 在 boundary_event_id 缺失时通过 _resolve_boundary_event_id 按 signature+occurrence 在 session.events 中查找边界，找不到即 raise ValueError，该异常随后被 _apply_scoped 计入 consecutive_failures 触发熔断。
触发条件： 边界事件已被更早的 staged callback（history snip stage 20、microcompact stage 30）从 session.events 裁剪/折叠，或已在上一次压缩中被归档到 historical_events，导致边界解析必然失败。
实际影响： AutoCompact 每次触发都失败并累加熔断计数，压缩长期无法执行，上下文持续膨胀直至触发阻塞阈值或压缩流程整体停摆。
修正方向： 边界解析失败时优雅降级（跳过本次压缩但不计入连续失败，或退化为不指定边界的压缩），而不是抛异常进入熔断路径。
| async def write_index(self, entries: list[MemoryIndexEntry]) -> None: | ||
| content = "\n".join(entry.to_markdown() for entry in entries) | ||
| if content: | ||
| content += "\n" | ||
| async with self._storage.create_db_session() as db: | ||
| # Keep the tenant's lock row locked until this transaction commits. | ||
| await self._storage.get_for_update( | ||
| db, | ||
| SqlKey(key=(self._app_name, self._user_id), storage_cls=SqlMemoryIndex), | ||
| ) | ||
| key = SqlKey(key=(self._app_name, self._user_id), storage_cls=SqlMemoryIndex) | ||
| row = await self._storage.get(db, key) | ||
| if row is None: | ||
| row = SqlMemoryIndex(app_name=self._app_name, user_id=self._user_id) | ||
| await self._storage.add(db, row) | ||
| row.content = content | ||
| row.expires_at = self._expiry(self._config.memory_ttl_seconds) | ||
| await self._refresh_memory_scope(db) | ||
| await self._storage.commit(db) |
There was a problem hiding this comment.
问题： write_index/write_topic 依赖"用 get_for_update 锁住 index 行作为租户互斥"，但当记录首次写入时锁行尚不存在，SELECT ... FOR UPDATE 不会锁定任何行，后续 row is None 分支直接 add 新行，互斥失效。
触发条件： 同一 memory scope 首次持久化时发生并发写入（多协程或多进程共享同一 SQL 后端，且尚无任何 index 行）。
实际影响： 两个写入方同时通过"无行"检查并插入相同主键行，触发唯一约束冲突（IntegrityError）抛给调用方，或产生重复写入，记忆数据写入失败。
修正方向： 先执行幂等 upsert（如 INSERT ... ON CONFLICT DO NOTHING）确保锁行存在后再 SELECT ... FOR UPDATE，或捕获唯一约束冲突后重试加锁流程。
| async def read_topic(self, topic_name: str) -> str | None: | ||
| path = self._paths.memory_topic_path(topic_name) | ||
| return await asyncio.to_thread(lambda: path.read_text(encoding=self._config.encoding) | ||
| if path.exists() else None) |
There was a problem hiding this comment.
问题： read_topic 直接读取 topic 文件内容，未像 read_index 路径那样先做 _is_expired 过期校验。
触发条件： topic 内容已超过 TTL 存活期但清理任务尚未运行（或清理按 index 过期判定而 index 条目仍新鲜）时，读取该记忆项。
实际影响： 已过期记忆仍被读回并注入模型上下文，产生陈旧/失效记忆，与 local 后端"按 TTL 清理"的语义不一致。
修正方向： 在 read_topic 返回前增加与 read_index 一致的过期检查，过期视为不存在（可顺带删除文件），并补充过期读取的测试。
| memory_type=resolved_type, | ||
| content=content, | ||
| ) | ||
| async with self._index_lock: | ||
| path = await self._runtime.long_term_memory.write_topic( | ||
| runtime = self._runtime_for_context(tool_context) | ||
| async with self._index_lock(runtime): | ||
| path = await runtime.long_term_memory.write_topic( | ||
| filename, | ||
| document, | ||
| ) | ||
| entries = _parse_index(await self._runtime.long_term_memory.read_index()) | ||
| entries = _parse_index(await runtime.long_term_memory.read_index()) | ||
| new_entry = MemoryIndexEntry( | ||
| name=name, | ||
| filename=path.name, | ||
| summary=summary, | ||
| ) | ||
| entries = [entry for entry in entries if entry.filename != new_entry.filename] | ||
| entries.insert(0, new_entry) | ||
| await self._runtime.long_term_memory.write_index(entries) | ||
| updated_at = parse_memory_updated_at(await self._runtime.long_term_memory.read_topic(filename) or "") | ||
| await runtime.long_term_memory.write_index(entries) |
There was a problem hiding this comment.
问题： _runtime_for_context 在 tool_context is None（未携带 scope）时解析到根（unscoped）runtime；redis/sql 后端下 storage_reference 因 self.scope is None 抛出 ValueError，而此时 topic 等数据可能已经写入非预期作用域的后端或已产生部分写副作用。
触发条件： 在未绑定 session scope 的上下文中调用高级记忆工具且后端为 redis/sql。
实际影响： 工具调用直接失败，或发生写入了错误作用域/错误后端后再报错的部分写，导致读写作用域不一致、数据落错位置。
修正方向： 在发生任何写入副作用之前校验 scope 可解析（redis/sql 后端要求 scope），未解析时先报错终止，或显式回退到根作用域的合法路径并保持一致。
| if self.config.storage_backend == "sql" and self.root._sql_cleanup is not None: | ||
| await self.root._sql_cleanup.start() | ||
| if self.config.storage_backend == "local" and self.root._local_cleanup is not None: | ||
| await self.root._local_cleanup.start() | ||
| if self.config.storage_backend == "sql" and self.root._sql_cleanup is not None: | ||
| await self.root._sql_cleanup.start() |
There was a problem hiding this comment.
问题： ScopedAdvancedMemoryRuntime.initialize 中重复执行 await self.root._sql_cleanup.start()，而根 runtime 初始化时已启动过该清理循环，start() 未保证幂等。
触发条件： 每个 scoped runtime（每 scope/每 session 创建）走 initialize 时。
实际影响： SQL 清理循环被多个任务重复运行，造成冗余清理与资源浪费；若 start() 对重复启动抛错，scoped 初始化将直接失败。
修正方向： 将 SqlAdvancedMemoryCleanup 的启动收敛到根 runtime 统一初始化，或在 start() 内保证幂等（已运行则直接返回），并补充重复初始化测试。
| autocompact = setup_autocompact( | ||
| agent, | ||
| runtime, | ||
| self._summary_generator, | ||
| model=self._compact_model, | ||
| ) | ||
| autocompact.attach_session_memory_extractor(extractor) | ||
| extractor.attach_session_service(self._session_service) | ||
| self._runtime = runtime | ||
| self._session_memory_extractor = extractor | ||
|
|
There was a problem hiding this comment.
问题： 本次新增的 AutoCompact 全链路、边界解析（boundary signature/occurrence）、HistorySnip、Microcompact、TokenContextTracker 与 staged callback 排序等核心路径缺少针对性测试——tests/sessions/compact/ 仅覆盖 manager 装配与少量回调行为，未覆盖 _persist_session_compaction 的边界解析失败降级、熔断计数与恢复，以及 stage 10/20/30 回调在真实 Runner 下的执行。
触发条件： 上述路径在后续迭代中发生回归时。
实际影响： 已确认的边界解析失败熔断、回调顺序等缺陷无法在 CI 中暴露，回归风险高。
修正方向： 补充 AutoCompact 全链路测试（含边界事件被裁剪/归档后的失败降级）、熔断恢复测试及各 staged callback 按 stage 排序执行的断言。
23ea179 to
fafcb5d
Compare
AI Code Review审查结论不通过 审查范围： 发现的问题严重
问题： head 提交 触发条件： 运行 实际影响： 本次变更带来的测试套件收集失败——CI 全量测试无法运行；该文件正处本次重构（ 修正方向： 删除 14-20 行的冲突块，按功能需要保留 中等
问题： 本行段（ 触发条件： 启用 token 模式（配置了 实际影响： usage 基线失效校验形同虚设：指令/工具变更后 token 计数仍按旧 usage 高估，autocompact 可能在远低于真实阈值的时机被错误触发（或相反被压制），导致不必要的上下文压缩或漏压缩；另外 修正方向： 二选一：① 在三个 SessionService 的 中等
问题： 触发条件： 启用 实际影响： 清理任务永久死亡且无任何补偿：后续 SQL 过期行不再删除、本地过期文件不再清理——记忆数据随时间无限膨胀（redis 后端因 TTL 由 redis 自身过期不受影响，sql/local 后端受影响）。 修正方向： 在 中等
问题： 本次重构将原 触发条件： 本次变更删除的测试与 实际影响： 上下文压缩是数据敏感操作—— 修正方向： 在 较低
问题： 触发条件： 启用 sql 或 redis 后端且 实际影响： 热路径延迟被放大：sql 后端每次请求的写事务增加锁竞争与吞吐开销（ 修正方向： 将 prune 从读路径移出，收敛到写路径（ |
| <<<<<<< HEAD | ||
| from trpc_agent_sdk.memory.advanced_memory import MemoryIndexEntry | ||
| ======= | ||
| from trpc_agent_sdk.memory.advanced_memory import MemoryDocument | ||
| from trpc_agent_sdk.memory.advanced_memory import MemoryIndexEntry | ||
| from trpc_agent_sdk.memory.advanced_memory import MemoryType | ||
| >>>>>>> 23ea179 (feature: Advanced memory 长期记忆实现优化) |
There was a problem hiding this comment.
问题： head 提交 fafcb5d 在修改 tests/advanced_memory/test_memory_context.py 时把未解决的 git 合并冲突标记直接合进了提交内容：14-20 行残留 <<<<<<< HEAD / ======= / >>>>>>> 23ea179 三个冲突行。同时 HEAD 一侧丢弃了 MemoryDocument 与 MemoryType 的导入，而该文件 92-95 行（本次变更新增的 runtime.long_term_memory.write_topic 调用）仍在显式使用这两个符号。
触发条件： 运行 pytest（pyproject.toml 的 testpaths = ["tests"] 会收集该文件）。ast.parse 在 14 行即报 SyntaxError: invalid syntax，整个测试会话在收集阶段被中断；即使手工删掉冲突标记按 HEAD 侧解析，MemoryDocument/MemoryType 也是未定义名称，该测试必然 NameError。
实际影响： 本次变更带来的测试套件收集失败——CI 全量测试无法运行；该文件正处本次重构（trpc_agent_sdk/advanced_memory 迁移到 trpc_agent_sdk/memory/advanced_memory）的核心测试路径上。
修正方向： 删除 14-20 行的冲突块，按功能需要保留 MemoryDocument、MemoryIndexEntry、MemoryType 三行导入（三者均已由 trpc_agent_sdk/memory/advanced_memory/__init__.py 导出），并重新运行该文件及相关测试确认可收集、可执行。
| if self._session_memory_extractor is not None: | ||
| await self._session_memory_extractor.extract_if_needed( | ||
| ctx.session, | ||
| ctx, | ||
| force=True, | ||
| ) | ||
| session_memory = await self._latest_session_memory_record( | ||
| session_id, | ||
| ctx, | ||
| ) |
There was a problem hiding this comment.
问题： 本行段（_apply_scoped 的 compaction 主路径，本次范围新增）所在会话的每次模型请求都会经由 AutoCompactCallback.__call__ 调用 TokenContextTracker.record_request_context，它把静态指纹写进 session.state["advanced_memory_pending_request_context_fingerprint"]；而指纹的消费端 _latest_usage_baseline（_token_budget.py:186）读取的是 Event 的 custom_metadata["advanced_memory_request_context_fingerprint"]。本提交范围把唯一负责把 pending 指纹迁移到 Event custom_metadata 的实现（2219dbf 删除的 sessions/compact/_session_service.py 的 append_event：usage_metadata 非空时拷贝 pending 指纹到 event custom_metadata）整文件删除，且未在任何其他位置重建该迁移；head 全仓库搜索只剩写入（_token_budget.py:249）与读取（_token_budget.py:186）两处引用，中间没有任何衔接。
触发条件： 启用 token 模式（配置了 model_context_window_tokens）后运行任意会话：新请求基于上一次带 usage_metadata 的 Event 计算 usage 基线；当系统指令或工具集发生变化（静态指纹理应不同，test_changed_recorded_system_or_tool_fingerprint_falls_back 正是为此设计）时，由于 Event 上没有指纹，recorded_fingerprint 恒为 None，旧的 usage 被无条件采纳。
实际影响： usage 基线失效校验形同虚设：指令/工具变更后 token 计数仍按旧 usage 高估，autocompact 可能在远低于真实阈值的时机被错误触发（或相反被压制），导致不必要的上下文压缩或漏压缩；另外 session.state 每请求残留一个按键，随事件 state_delta 持久化到 redis/sql。范围内删除迁移逻辑使该缺陷由"功能不完整"变成"确定失效"。
修正方向： 二选一：① 在三个 SessionService 的 append_event 中恢复迁移：把 session.state["advanced_memory_pending_request_context_fingerprint"] 写入带 usage_metadata 的 Event 的 custom_metadata["advanced_memory_request_context_fingerprint"] 并清除该 state 键；② 让 _latest_usage_baseline 直接读 session.state 中由 record_request_context 记录的指纹，与消费端使用同一个键。
| async def _run(self) -> None: | ||
| if self._stop_event is None: | ||
| return | ||
| try: | ||
| while not self._stop_event.is_set(): | ||
| try: | ||
| await asyncio.wait_for( | ||
| self._stop_event.wait(), | ||
| timeout=self._config.sql_cleanup_interval_seconds, | ||
| ) | ||
| except asyncio.TimeoutError: | ||
| await self.cleanup_once() | ||
| except asyncio.CancelledError: | ||
| raise | ||
|
|
There was a problem hiding this comment.
问题： SqlAdvancedMemoryCleanup._run 的 while 循环体内只捕获 asyncio.TimeoutError；cleanup_once()（254-282 行）对过期行做 delete、对每个 index 读全部 topic 并 prune_memory_index，任一数据库异常（连接断开、锁等待超时、约束冲突等）都会直接传播出 while 循环，销毁 self._task。_storage.py:204-217 的 LocalAdvancedMemoryCleanup._run 存在完全相同的结构问题（cleanup_once 里的文件遍历/unlink 异常同样会终止循环）。
触发条件： 启用 memory_ttl_seconds（两个 cleanup 的 start() 均要求它非 None）且进程持续运行；某次周期清理遇到一次可恢复的存储层异常即可触发。
实际影响： 清理任务永久死亡且无任何补偿：后续 SQL 过期行不再删除、本地过期文件不再清理——记忆数据随时间无限膨胀（redis 后端因 TTL 由 redis 自身过期不受影响，sql/local 后端受影响）。close() 因 self._task.done() 为 True 而跳过 cancel，不会报错也不会有日志提示任务已死。
修正方向： 在 await self.cleanup_once() 外层增加兜底：except Exception: logger.exception("...cleanup failed...") 后继续循环（或指数退避），确保单次失败不终止后台任务；可参照项目中其他后台清理循环的容错模式。
| async def read_index(self) -> str: | ||
| async with self._storage.create_db_session() as db: | ||
| row = await self._storage.get(db, SqlKey(key=(self._app_name, self._user_id), storage_cls=SqlMemoryIndex)) | ||
| if row is None or self._expired(row.expires_at): | ||
| return "" | ||
| await self._refresh_memory_scope(db) | ||
| content = row.content | ||
| valid_topics = await self._storage.query( | ||
| db, | ||
| SqlKey(key=(self._app_name, self._user_id), storage_cls=SqlMemoryTopic), | ||
| SqlCondition(filters=[ | ||
| SqlMemoryTopic.app_name == self._app_name, | ||
| SqlMemoryTopic.user_id == self._user_id, | ||
| SqlMemoryTopic.expires_at.is_(None) | (SqlMemoryTopic.expires_at > self._now()), | ||
| ]), | ||
| ) | ||
| valid_filenames = {topic.topic_name for topic in valid_topics} | ||
| pruned_content = prune_memory_index(content, valid_filenames) | ||
| if pruned_content != content: | ||
| row.content = pruned_content | ||
| content = pruned_content | ||
| await self._storage.commit(db) |
There was a problem hiding this comment.
问题： SqlLongTermMemoryStore.read_index（128-149 行；read_topic/list_topics 同构）在读事务里调用 _refresh_memory_scope 并把 prune 后的内容直接写回 row.content 后 commit——即每次请求至少一次写事务；_redis_stores.py:136-161 的 read_index 则表现为每读一次对 index 逐条 EXISTS 探测每个 topic（N+1 次往返）。
触发条件： 启用 sql 或 redis 后端且 LongTermMemoryContext.apply/list_memory_index 工具在每次模型请求前触发（_memory_context.py:46 每次调用 read_index）：长会话、多 topic 场景下每次请求都产生写事务或 N+1 次 redis 往返。
实际影响： 热路径延迟被放大：sql 后端每次请求的写事务增加锁竞争与吞吐开销（_refresh_memory_scope 还会对全部 topic 行更新 expires_at，触发大量行级更新）；redis 后端 topic 较多时每请求多出 O(topic数) 次往返。属于性能问题，不改变语义。
修正方向： 将 prune 从读路径移出，收敛到写路径（write_index/write_topic 或独立的清理周期任务内）；redis 侧对 index 的逐条 EXISTS 可合并为一次 pipeline/mget，或仅在清理周期内对 registry 成员做探测。
fafcb5d to
bcdb021
Compare
AI Code Review审查结论不通过 审查范围：base f05797d..head bcdb021，98 个文件（+4839/−4961），主题为 Advanced Memory 服务化重构。计划符合性：独立 memory 包（local/redis/sql 后端）、Session Compact 由 SessionService 持有、删除旧 AdvancedMemorySessionService 包装器、删除磁盘 transcript 持久化等关键目标均已实现，但引入多项正确性与兼容性风险。 主要风险（按严重度）：1) SEVERE——SessionMemoryExtractor.extract_if_needed 的跨进程锁 guard 不传超时且配置项 session_memory_wait_timeout_seconds 全库无消费方，defer 模式下主循环请求可能无限阻塞；2) Token 基线静态指纹保护为死代码（record_request_context 写入的 pending key 与 _latest_usage_baseline 读取的事件 key 不一致，事件侧无写入方），指令/工具变更后旧 usage 基线仍被命中导致 token 低估、压缩阈值错位；3) autocompact 持久化在模型请求中途调用 update_session，SQL 后端 DELETE+重插整表覆盖，与并发写者竞态可丢失事件/state（该文件在 eb29c64..bcdb021 间的改动为压缩路径新增了 update_session 调用点，故锚定到本 diff 新文件 _autocompact.py）；4) boundary 解析失败经 ValueError 计入连续失败，达阈值后模型输出被 AUTOCOMPACT_BLOCKED_MESSAGE 硬阻断；5) summarizer 与 compact manager 互斥导致旧组合用法升级即启动失败；另有 manager setup 早退静默缺回调、patch_session_state 实现未校验、预算/压缩状态仅进程内存（配合磁盘持久化被删）、local read_topic 不检查 TTL、标准记忆 API no-op 静默空行为、测试覆盖大幅削弱等中低风险项。 测试充分性：本 diff 删除约 2100 行测试仅补充约 128 行，新模块关键路径（提取、持久化、锁、指纹）无覆盖，上述多数缺陷无法被 CI 捕获。 门禁结论：存在 SEVERE 级缺陷（跨进程锁无超时且配置无效，可无限阻塞请求），判定为 FAILED，建议修复锁超时与指纹链路后再合入。 发现的问题严重
问题： 触发条件： defer 模式下 实际影响： 主循环请求在该点无限阻塞，整个 Agent 请求被 hang 住；下游无任何超时兜底，配置了 修正方向： 在 中等
问题： Token 基线校验的静态指纹保护是死代码： 触发条件： 会话中途修改 system prompt 或工具集（指令/工具变更改变上下文）后，后续请求的尾部内容恰好与历史 usage 事件的内容指纹相同（reuse 场景）。 实际影响： 设计意图（见 修正方向： 打通写入链：让 中等
问题： 触发条件： 多进程或长请求场景下，另一写者（其他请求的 实际影响： 并发写入的事件或 state 更新被覆盖丢失，历史事件表短暂不一致（DELETE+重插非原子）；SQL 后端下为数据丢失级竞态，且失败时仅回滚内存对象，存储端已被部分写入。 修正方向： 压缩持久化改为增量写：用 中等
问题： 分支 触发条件： 边界事件被 实际影响： 非阻塞性质疑场景演进为硬阻断：用户收到占位消息而非模型输出，且 修正方向： 对 boundary 解析失败走降级而非抛异常：跳过本轮压缩（返回未变更结果、不递增失败计数），或将其排除在 中等
问题： 触发条件： 存量集成同时启用了 Summarizer（ 实际影响： 兼容性回归：旧配置在升级后启动失败，且异常发生在框架内部初始化阶段，用户难以定位；计划要求"删除旧包装器"但不要求废除与其并存的 summarizer 用法。 修正方向： 让两者可共存（例如各自管理独立状态槽位：compact manager 负责 较低
问题： 触发条件： 两个及以上 agent 共享同一个 实际影响： 第二个及以后的 agent 静默缺少 tool-budget/history-snip/microcompact/autocompact 全部回调，长会话上下文无限增长且无压缩保护，问题仅在运行期显现。 修正方向： 对 较低
问题： 触发条件： 使用自定义 SessionService 且未实现 实际影响： 压缩/记忆提取的 checkpoint 持久化在运行时抛 修正方向： 在校验阶段对 较低
问题： tool-budget 的 触发条件： 多 worker 部署或进程重启后同一 session 继续对话；已替换过的工具结果为不同 worker 保留， 实际影响： 重启后工具结果替换与压缩边界重复执行、失败计数清零，行为与持续运行进程不一致；在 Redis/SQL 后端下本可持久化却未持久化。 修正方向： 将状态写入 session state 或后端存储（如经 较低
问题： local 存储的 触发条件： 记忆条目超过 TTL 但索引尚未被读（因此文件未被清理）时，用户通过 实际影响： 修正方向： 较低
问题： 触发条件： 既有集成以标准方式调用 实际影响： 静默功能缺失：标准记忆路径的写入被丢弃、检索恒为空，调用方无任何错误提示，表现为"记忆功能失效"类缺陷，排查困难。 修正方向： 在 no-op 分支记录 warning 日志并在文档中明确该不兼容，或让 standard API 委托到 advanced 后端（二者本是同一存储），避免静默空行为。 较低
问题： 本 diff 删除约 2100 行旧测试（test_autocompact、test_session_memory_extractor、test_tool_result_budget、test_storage、test_history_snip、test_microcompact、test_transcript_session_service 共 7 个文件全部删除），仅新增约 128 行测试（tests/sessions/compact/test_session_compact.py 共 4 个用例），新模块的关键路径（Session Memory 提取、compact 回调管道、Redis/SQL 持久化竞态、TTL 语义、锁超时）均无测试覆盖。 触发条件： 后续改动经测试门禁时，所有被删除路径的回归无法被 CI 捕获。 实际影响： 本次重构（98 个文件、+4839/−4961）的核心行为回归风险无法被现有测试暴露，特别是压缩边界解析、锁超时、指纹校验等本清单已识别的缺陷。 修正方向： 为 SessionMemoryExtractor、AutoCompact 持久化与边界解析、TokenContextTracker 指纹链路、工具结果预算状态持久化补充针对性单元测试，至少覆盖本次缺陷清单中的各触发路径。 |
| session_key = runtime.session_key(session.id) | ||
| async with self._runtime.coordination.guard(session_key) as acquired: | ||
| if not acquired: |
There was a problem hiding this comment.
问题： extract_if_needed 通过 self._runtime.coordination.guard(session_key) 获取跨进程写锁时不传 timeout，而 CrossLoopLock.guard(timeout=None) 会无限等待（每 10ms 轮询直到拿到锁）。配置项 session_memory_wait_timeout_seconds（_config.py:70，默认 15.0）已声明并转发，但全代码库没有任何消费方（grep 仅命中定义与转发两处），超时配置完全无效。
触发条件： defer 模式下 _PostTurnWorkerThread 持锁进行提取 LLM 调用（HTTP 超时可达 600s）期间，主循环的下一请求进入 extract_if_needed 等其他持 guard 的路径且锁未释放，或另一进程的提取任务因 LLM 慢/挂起长时间占锁。
实际影响： 主循环请求在该点无限阻塞，整个 Agent 请求被 hang 住；下游无任何超时兜底，配置了 session_memory_wait_timeout_seconds 也无法改变该行为。
修正方向： 在 guard() 调用处传入 self._runtime.config.session_memory_wait_timeout_seconds 作为超时参数，未获锁时按现有 "coordination-timeout" 分支优雅降级（跳过本次提取，交给下一轮）。
| def estimate_request_tokens(self, request: "LlmRequest") -> int: | ||
| """Estimate a complete request without applying a usage baseline.""" | ||
| return self._estimate_request(request) | ||
|
|
There was a problem hiding this comment.
问题： Token 基线校验的静态指纹保护是死代码：_latest_usage_baseline 从事件 custom_metadata["advanced_memory_request_context_fingerprint"] 读取指纹，但整个代码库没有任何地方向事件元数据写入该 key；record_request_context 写入的是 session.state["advanced_memory_pending_request_context_fingerprint"]（_token_budget.py:249），该 key 无任何读取方，且未在响应事件上持久化。
触发条件： 会话中途修改 system prompt 或工具集（指令/工具变更改变上下文）后，后续请求的尾部内容恰好与历史 usage 事件的内容指纹相同（reuse 场景）。
实际影响： 设计意图（见 test_token_budget.py 的 test_changed_recorded_system_or_tool_fingerprint_falls_back）是在指令/工具变更后丢弃旧 usage 基线，但旧基线仍会被命中 → token 低估 → 提前/错误触发 autocompact 或 blocking 阈值判断，压缩边界错位。
修正方向： 打通写入链：让 record_request_context 记录的指纹随请求写入模型响应事件（写入 Event.custom_metadata），并统一两个 key 名称，确保 _latest_usage_baseline 读取的是模型实际执行时请求的静态指纹而非从未存在的事件字段。
| async def _persist_session_compaction( | ||
| self, | ||
| ctx: "InvocationContext", | ||
| record: AutoCompactRecord, | ||
| ) -> None: | ||
| """Persist the compacted active window through the original SessionService.""" | ||
| compact_events = getattr(ctx.session, "compact_events", None) | ||
| if not callable(compact_events): | ||
| # AutoCompact 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 AutoCompact boundary to an active Session Event") | ||
|
|
||
| compaction_id = record.compaction_id or f"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: |
There was a problem hiding this comment.
问题： _persist_session_compaction（_autocompact.py:478-481）在模型请求调用链中途调用 ctx.session_service.update_session(ctx.session)，SQL 后端 update_session（_sql_session_service.py:648 起）会 DELETE 该 session 的全部事件行再逐条重插，并整行覆盖 state/historical_events/conversation_count；而 ctx.session 是 run 启动/上次刷新时加载的视图，可能已过期。
触发条件： 多进程或长请求场景下，另一写者（其他请求的 append_event/patch_session_state）在本会话两次读写之间提交了新数据，随后本请求触发 autocompact 并整表重建。
实际影响： 并发写入的事件或 state 更新被覆盖丢失，历史事件表短暂不一致（DELETE+重插非原子）；SQL 后端下为数据丢失级竞态，且失败时仅回滚内存对象，存储端已被部分写入。
修正方向： 压缩持久化改为增量写：用 patch_session_state 只写压缩元数据（如 _trpc_agent:summary、last_compaction），并让 compact_events 的归档动作经服务端原子操作（单条 append + 条件删除），避免整体 DELETE+重插覆盖并发写入。
| 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 AutoCompact boundary to an active Session Event") |
There was a problem hiding this comment.
问题： 分支 boundary_event_id is None 时 _persist_session_compaction 直接 raise ValueError，该异常被 _apply_scoped 捕获后递增 consecutive_failures（_autocompact.py:709-711）；当失败次数达 autocompact_max_failures（默认 3）且达到 blocking 阈值时，AutoCompactCallback.__call__ 返回 AUTOCOMPACT_BLOCKED_MESSAGE 替换模型真实响应。
触发条件： 边界事件被 filter_events/compact_events 归档出活跃窗口后 _resolve_boundary_event_id 返回 None，或 _legacy_boundary_event_id 在 len(content_events) <= 1 时返回 None——连续三轮触发即达阻断阈值。
实际影响： 非阻塞性质疑场景演进为硬阻断：用户收到占位消息而非模型输出，且 consecutive_failures 仅存进程内存，重启后才复位；在 SQL/Redis 后端下阻塞成因（边界事件被归档）不会自愈。
修正方向： 对 boundary 解析失败走降级而非抛异常：跳过本轮压缩（返回未变更结果、不递增失败计数），或将其排除在 consecutive_failures 统计之外，避免该错误模式触达模型阻断。
| summarizer_manager: The summarizer manager to use | ||
| force: Whether to force update even if already set | ||
| """ | ||
| if self._session_compact_manager is not None: | ||
| raise ValueError("SummarizerSessionManager and BaseSessionCompactManager are mutually exclusive") | ||
| if not self._summarizer_manager or force: | ||
| self._summarizer_manager = summarizer_manager | ||
| self._summarizer_manager.set_session_service(self) | ||
|
|
||
| def set_session_compact_manager( | ||
| self, | ||
| compact_manager: "BaseSessionCompactManager", | ||
| force: bool = False, | ||
| ) -> None: | ||
| """Attach Session Compact through the native manager lifecycle.""" | ||
| if self._summarizer_manager is not None: | ||
| raise ValueError("SummarizerSessionManager and BaseSessionCompactManager are mutually exclusive") | ||
| if self._session_compact_manager is not None and not force: | ||
| if self._session_compact_manager is compact_manager: | ||
| return | ||
| raise ValueError("A Session Compact manager is already configured") | ||
| self._session_compact_manager = compact_manager | ||
| compact_manager.set_session_service(self, force=force) | ||
|
|
||
| @override | ||
| async def append_event(self, session: Session, event: Event) -> Event: | ||
| """Appends an event to a session object.""" |
There was a problem hiding this comment.
问题： set_summarizer_manager 与 set_session_compact_manager 互斥（任何一方已设置即抛 ValueError，_base_session_service.py:96-97），本 diff 把 Advanced Memory 从旧 AdvancedMemorySessionService 包装器迁到 AdvancedSessionCompactManager 后，原本合法的组合用法被破坏。
触发条件： 存量集成同时启用了 Summarizer（SummarizerSessionManager）与 Advanced Memory（compact manager），升级本版本后 bind()/setup() 路径抛 ValueError。
实际影响： 兼容性回归：旧配置在升级后启动失败，且异常发生在框架内部初始化阶段，用户难以定位；计划要求"删除旧包装器"但不要求废除与其并存的 summarizer 用法。
修正方向： 让两者可共存（例如各自管理独立状态槽位：compact manager 负责 _trpc_agent:summary，summarizer 管理常规摘要事件），或在新版文档/迁移说明中明确该组合不再支持并提供迁移路径。
| raise RuntimeError("Session Compact manager must be bound to a SessionService first") | ||
| if self._runtime is not None: | ||
| return |
There was a problem hiding this comment.
问题： AdvancedSessionCompactManager.setup（_manager.py:51-84）在 self._runtime is not None 时直接 return，且不校验已安装回调是否针对同一 agent；复用同一 SessionService 创建第二个 Runner（框架支持多 agent 共用服务）时，第二个 agent 不会注册任何压缩回调，且无任何告警。
触发条件： 两个及以上 agent 共享同一个 AdvancedMemoryService/compact manager 实例并各自调用 setup(agent)。
实际影响： 第二个及以后的 agent 静默缺少 tool-budget/history-snip/microcompact/autocompact 全部回调，长会话上下文无限增长且无压缩保护，问题仅在运行期显现。
修正方向： 对 self._runtime is not None 且目标 agent 不是首次注册的实例抛出明确异常，或按 agent 维护独立的 runtime/回调集合。
| ) -> None: | ||
| """Attach Session Compact through the native manager lifecycle.""" | ||
| if self._summarizer_manager is not None: | ||
| raise ValueError("SummarizerSessionManager and BaseSessionCompactManager are mutually exclusive") | ||
| if self._session_compact_manager is not None and not force: | ||
| if self._session_compact_manager is compact_manager: | ||
| return | ||
| raise ValueError("A Session Compact manager is already configured") | ||
| self._session_compact_manager = compact_manager | ||
| compact_manager.set_session_service(self, force=force) | ||
|
|
||
| @override | ||
| async def append_event(self, session: Session, event: Event) -> Event: | ||
| """Appends an event to a session object.""" |
There was a problem hiding this comment.
问题： set_session_compact_manager（_base_session_service.py:105-111 起）在绑定前仅校验存储端 store_historical_events=True，不校验 patch_session_state 是否被服务端实现（SessionABC/SessionServiceBase 的默认实现为 raise NotImplementedError），SQL/Redis 之外的第三方实现可能不满足压缩持久化前提。
触发条件： 使用自定义 SessionService 且未实现 patch_session_state 的集成，或直接以仅内存的服务绑定 compact manager 后运行（如本 diff 新增的测试使用 InMemorySessionService）。
实际影响： 压缩/记忆提取的 checkpoint 持久化在运行时抛 NotImplementedError，经 _apply_scoped 全捕获后计入失败计数，最终表现为模型输出被 AUTOCOMPACT_BLOCKED_MESSAGE 替换，根因难以定位。
修正方向： 在校验阶段对 session_service.patch_session_state 做实现检查（如 is_implemented 标志或能力探测），不满足时明确报错而不是留到运行期。
| async def _load_state(self, session_id: str) -> ToolResultBudgetState: | ||
| """Restore frozen results and historical replacements from the transcript.""" | ||
| state = self._states.get(session_id) | ||
| """Return process-local state for the current Session.""" | ||
| state_key = self._runtime.session_key(session_id) if hasattr(self._runtime, "session_key") else session_id | ||
| state = self._states.get(state_key) | ||
| if state is not None: | ||
| return state | ||
| records = await self._runtime.transcripts.read_all(session_id) | ||
| seen_ids: set[str] = set() | ||
| replacements: dict[str, dict[str, Any]] = {} | ||
| result_hashes: dict[str, str] = {} | ||
| for record in records: | ||
| if record.get("kind") not in { | ||
| "content-replacement", | ||
| "content-replacement-decision", | ||
| }: | ||
| continue | ||
| result_id = record.get("result_id") | ||
| replacement = record.get("replacement_response") | ||
| original_sha256 = record.get("original_sha256") | ||
| if isinstance(result_id, str): | ||
| seen_ids.add(result_id) | ||
| if isinstance(original_sha256, str): | ||
| result_hashes[result_id] = original_sha256 | ||
| if record.get("kind") == "content-replacement" and isinstance(replacement, dict): | ||
| replacements[result_id] = replacement | ||
| state = ToolResultBudgetState( | ||
| seen_ids=seen_ids, | ||
| replacements=replacements, | ||
| result_hashes=result_hashes, | ||
| seen_ids=set(), | ||
| replacements={}, | ||
| result_hashes={}, | ||
| ) | ||
| self._states[session_id] = state | ||
| self._states[state_key] = state | ||
| return state |
There was a problem hiding this comment.
问题： tool-budget 的 seen_ids/replacements/result_hashes、autocompact 的 consecutive_failures/latest_compaction 等状态仅存进程内存（_tool_result_budget.py:136-148 _load_state 新建空状态），本 diff 同时删除了磁盘 transcript/工具结果全文持久化（SessionCompactRuntime 不再持有 transcripts/tool_results/paths，新测试 test_compact_runtime_has_no_external_storage 明确印证该删除），跨进程与跨重启后这些状态全部丢失。
触发条件： 多 worker 部署或进程重启后同一 session 继续对话；已替换过的工具结果为不同 worker 保留，consecutive_failures 复位。
实际影响： 重启后工具结果替换与压缩边界重复执行、失败计数清零，行为与持续运行进程不一致；在 Redis/SQL 后端下本可持久化却未持久化。
修正方向： 将状态写入 session state 或后端存储（如经 patch_session_state 落库），至少在分布式后端上消除跨进程不一致。
| async def read_topic(self, topic_name: str) -> str | None: | ||
| path = self._paths.memory_topic_path(topic_name) | ||
| return await asyncio.to_thread(lambda: path.read_text(encoding=self._config.encoding) | ||
| if path.exists() else None) |
There was a problem hiding this comment.
问题： local 存储的 read_topic 不做 TTL 过期检查直接读取文件，而同一文件内的 _read_index_sync 与索引剪枝逻辑都按 memory_ttl_seconds 判定过期并删除 .md 文件（_storage.py:91-124）；两处对"过期"的语义不一致。
触发条件： 记忆条目超过 TTL 但索引尚未被读（因此文件未被清理）时，用户通过 read_memory 工具读取该 topic。
实际影响： read_memory 返回已过期的记忆内容，长期记忆工具结果与 TTL 语义不一致，用户会读到本应失效的旧记忆。
修正方向： read_topic/read_topic_frontmatter 加与 _read_index_sync 相同的 _is_expired 检查，过期时返回 None（并顺带清理）。
| async def store_session( | ||
| self, | ||
| session: Session, | ||
| session: SessionABC, | ||
| agent_context: Optional[AgentContext] = None, | ||
| ) -> None: | ||
| """Keep the standard Runner post-turn contract without duplicating work. | ||
| """Keep the standard hook side-effect free. | ||
|
|
||
| The wrapped session service performs session-memory extraction from | ||
| ``create_session_summary`` before Runner reaches this method. | ||
| Advanced Memory is model-directed: the Agent decides what is durable | ||
| and calls ``save_memory``. Automatically storing every Session here | ||
| would mix transient conversation history with long-term memory. | ||
| """ | ||
| return None | ||
|
|
||
| @override | ||
| async def search_memory( | ||
| self, | ||
| key: str, | ||
| query: str, | ||
| limit: int = 10, | ||
| agent_context: Optional[AgentContext] = None, | ||
| ) -> SearchMemoryResponse: | ||
| """Return an empty legacy-style response. | ||
| """Return the standard empty response for compatibility. | ||
|
|
||
| Advanced long-term memory is intentionally accessed through its | ||
| ``save_memory``, ``read_memory``, and ``list_memory_index`` tools. | ||
| """ | ||
| return SearchMemoryResponse() |
There was a problem hiding this comment.
问题： AdvancedMemoryService.store_session（95-107）与 search_memory 为 no-op（直接 return None/返回空 SearchMemoryResponse()），但标准记忆 API 消费方（如内存工具加载路径 _load_memory_tool.py 经 search_memory 检索、runner post-turn 经 store_session 持久化）不受 advanced 模式影响，调用方拿到的始终是空应答。
触发条件： 既有集成以标准方式调用 session_service.store_session 或工具链中的 search_memory，同时启用了 Advanced Memory 后端。
实际影响： 静默功能缺失：标准记忆路径的写入被丢弃、检索恒为空，调用方无任何错误提示，表现为"记忆功能失效"类缺陷，排查困难。
修正方向： 在 no-op 分支记录 warning 日志并在文档中明确该不兼容，或让 standard API 委托到 advanced 后端（二者本是同一存储），避免静默空行为。
| def test_compact_runtime_has_no_external_storage() -> None: | ||
| runtime = SessionCompactRuntime.create(AdvancedCompactConfig()) | ||
|
|
||
| assert not hasattr(runtime, "transcripts") | ||
| assert not hasattr(runtime, "tool_results") | ||
| assert not hasattr(runtime, "paths") | ||
|
|
||
|
|
||
| @pytest.mark.asyncio | ||
| async def test_session_service_accepts_a_configured_compact_manager() -> None: | ||
| manager = AdvancedSessionCompactManager(config=AdvancedCompactConfig()) | ||
| service = InMemorySessionService( | ||
| session_config=SessionServiceConfig(store_historical_events=True), | ||
| session_compact_manager=manager, |
There was a problem hiding this comment.
问题： 本 diff 删除约 2100 行旧测试（test_autocompact、test_session_memory_extractor、test_tool_result_budget、test_storage、test_history_snip、test_microcompact、test_transcript_session_service 共 7 个文件全部删除），仅新增约 128 行测试（tests/sessions/compact/test_session_compact.py 共 4 个用例），新模块的关键路径（Session Memory 提取、compact 回调管道、Redis/SQL 持久化竞态、TTL 语义、锁超时）均无测试覆盖。
触发条件： 后续改动经测试门禁时，所有被删除路径的回归无法被 CI 捕获。
实际影响： 本次重构（98 个文件、+4839/−4961）的核心行为回归风险无法被现有测试暴露，特别是压缩边界解析、锁超时、指纹校验等本清单已识别的缺陷。
修正方向： 为 SessionMemoryExtractor、AutoCompact 持久化与边界解析、TokenContextTracker 指纹链路、工具结果预算状态持久化补充针对性单元测试，至少覆盖本次缺陷清单中的各触发路径。
2 participants
No description provided.