From 15b2380c43f5c89dfb6d0d70f03a804020c91070 Mon Sep 17 00:00:00 2001 From: Ziliang_Zhou Date: Wed, 15 Jul 2026 15:40:06 +0800 Subject: [PATCH 001/291] fix(memory): degrade to FTS5-only search when embedding is unconfigured MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs: #1320 **What type of PR is this?** /kind bug **Self-checklist**:（**请自检，在[ ]内打上x，我们将检视你的完成情况，否则会导致pr无法合入**） + - [ ] **设计**：PR对应的方案是否已经经过Maintainer评审，方案检视意见是否均已答复并完成方案修改 + - [ ] **测试**：PR中的代码是否已有UT/ST测试用例进行充分的覆盖，新增测试用例是否随本PR一并上库或已经上库 + - [ ] **验证**：PR描述信息中是否已包含对该PR对应的Feature、Refactor、Bugfix的预期目标达成情况的详细验证结果描述 + - [ ] **接口**：是否涉及对外接口变更，相应变更已得到接口评审组织的通过，API对应的注释信息已经刷新正确 + - [ ] **文档**：是否涉及官网文档修改，如果涉及请及时提交资料到Doc仓 --- openjiuwen/core/memory/lite/manager.py | 34 ++++++++++++++++++++++---- 1 file changed, 29 insertions(+), 5 deletions(-) diff --git a/openjiuwen/core/memory/lite/manager.py b/openjiuwen/core/memory/lite/manager.py index 6a9abaf52..7f2392a84 100644 --- a/openjiuwen/core/memory/lite/manager.py +++ b/openjiuwen/core/memory/lite/manager.py @@ -362,6 +362,14 @@ def _ensure_schema(self) -> None: async def _initialize_provider(self) -> None: """Initialize embedding provider.""" + if self.embedding_config is None or not getattr(self.embedding_config, "api_key", None): + self.provider = None + self.provider_key = "none:no-embedding" + logger.info( + "Embedding provider not configured (no embedding_config / api_key); " + "memory will use FTS5 keyword search only." + ) + return try: self.provider = await create_embedding_provider( model=self.settings.model, @@ -600,10 +608,13 @@ async def _should_full_reindex(self) -> bool: meta = json.loads(row["value"]) - if meta.get("provider") != self.provider.id: + provider_id = self.provider.id if self.provider else None + provider_model = self.provider.model if self.provider else None + + if meta.get("provider") != provider_id: return True - if meta.get("model") != self.provider.model: + if meta.get("model") != provider_model: return True if meta.get("chunkTokens") != self.settings.chunking.get("tokens"): @@ -633,8 +644,8 @@ async def _run_reindex(self) -> None: await self._sync_session_files() meta = { - "provider": self.provider.id, - "model": self.provider.model, + "provider": self.provider.id if self.provider else None, + "model": self.provider.model if self.provider else None, "providerKey": self.provider_key, "chunkTokens": self.settings.chunking.get("tokens"), "chunkOverlap": self.settings.chunking.get("overlap"), @@ -762,6 +773,8 @@ async def _index_chunk( embedding = await self._get_embedding(chunk.text) + model_name = self.provider.model if self.provider else None + cursor = self.db.execute(""" INSERT OR REPLACE INTO chunks (id, path, source, start_line, end_line, hash, model, text, embedding, updated_at) @@ -769,7 +782,7 @@ async def _index_chunk( RETURNING rowid """, ( chunk_id, file_path, source, chunk.start_line, chunk.end_line, - chunk_hash, self.provider.model, chunk.text, + chunk_hash, model_name, chunk.text, vector_to_blob(embedding) if embedding else None, int(asyncio.get_event_loop().time()) if self._event_loop else 0 )) @@ -914,6 +927,13 @@ async def search( if r["score"] >= min_score ][:max_results] + # if not embedding, Skip the rerank and use the raw keyword scores directly instead. + if not has_vector: + return [ + r for r in keyword_results + if r["score"] >= min_score + ][:max_results] + merged = _merge_hybrid_results( vector_results, keyword_results, @@ -933,6 +953,8 @@ async def _search_vector( return await self._search_vector_fallback(query_vec, limit) if not self.vector_dims: + if not self.provider: + return [] sample = await self.provider.embed_query("sample") self._ensure_vector_table(len(sample)) else: @@ -1122,6 +1144,8 @@ def _build_source_filter(self) -> tuple: async def _embed_query_with_timeout(self, query: str) -> List[float]: """Embed query with timeout.""" + if not self.provider: + return [] try: timeout = 60.0 return await asyncio.wait_for( From bb1ffa20a30c670a856703a0db81fc3f21ce0e39 Mon Sep 17 00:00:00 2001 From: hzx99 Date: Wed, 15 Jul 2026 17:11:19 +0800 Subject: [PATCH 002/291] fix(memory): resolve issue where re-indexing fails after embedding configuration changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs: #1319 **What type of PR is this?** /kind bug **Self-checklist**:（**请自检，在[ ]内打上x，我们将检视你的完成情况，否则会导致pr无法合入**） + - [ ] **设计**：PR对应的方案是否已经经过Maintainer评审，方案检视意见是否均已答复并完成方案修改 + - [ ] **测试**：PR中的代码是否已有UT/ST测试用例进行充分的覆盖，新增测试用例是否随本PR一并上库或已经上库 + - [ ] **验证**：PR描述信息中是否已包含对该PR对应的Feature、Refactor、Bugfix的预期目标达成情况的详细验证结果描述 + - [ ] **接口**：是否涉及对外接口变更，相应变更已得到接口评审组织的通过，API对应的注释信息已经刷新正确 + - [ ] **文档**：是否涉及官网文档修改，如果涉及请及时提交资料到Doc仓 --- openjiuwen/core/memory/lite/manager.py | 300 ++++++++++++++++++++++--- 1 file changed, 268 insertions(+), 32 deletions(-) diff --git a/openjiuwen/core/memory/lite/manager.py b/openjiuwen/core/memory/lite/manager.py index 7f2392a84..dfbffaf29 100644 --- a/openjiuwen/core/memory/lite/manager.py +++ b/openjiuwen/core/memory/lite/manager.py @@ -7,11 +7,78 @@ import sqlite3 import struct import asyncio +import contextlib from datetime import datetime, timedelta, timezone from typing import List, Optional, Dict, Any, Set, TYPE_CHECKING from dataclasses import dataclass from openjiuwen.core.common.logging import memory_logger as logger + +@contextlib.contextmanager +def _cross_process_lock(lock_path: str, timeout: float = 30.0): + """Cross-process exclusive lock via a lock file (blocks until acquired). + + Serializes the config-change rebuild across concurrently starting memory + manager instances, so only one process drops & rebuilds the index tables + while the others wait — then the waiters re-check and find the rebuild + already done. Without this, concurrent rebuilds race on the same DB and + fail with "database is locked", leaving a half-built (empty) vector table. + + Windows: msvcrt.locking (LK_LOCK, blocks with retry). POSIX: fcntl.flock. + + Raises ``TimeoutError`` if the lock is not acquired within ``timeout`` — + the caller must NOT run the rebuild without the lock, so we fail loudly + rather than silently proceeding and racing with another rebuild. + """ + import time + + ensure_dir(os.path.dirname(lock_path) or ".") + fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o644) + acquired = False + try: + deadline = time.monotonic() + timeout + if os.name == "nt": + import msvcrt + + while time.monotonic() < deadline: + try: + msvcrt.locking(fd, msvcrt.LK_LOCK, 1) + acquired = True + break + except OSError: + time.sleep(0.1) + else: + import fcntl + + while time.monotonic() < deadline: + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + acquired = True + break + except OSError: + time.sleep(0.1) + if not acquired: + raise TimeoutError(f"Could not acquire rebuild lock within {timeout}s: {lock_path}") + yield + finally: + if acquired: + try: + if os.name == "nt": + import msvcrt + + # Unlock failures are non-fatal (the fd is closed below and + # byte-range locks release on close); let the outer except + # log them at debug rather than silently swallowing here. + os.lseek(fd, 0, os.SEEK_SET) + msvcrt.locking(fd, msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(fd, fcntl.LOCK_UN) + except Exception as e: + logger.debug(f"Failed to release rebuild lock: {e}") + os.close(fd) + if TYPE_CHECKING: from openjiuwen.core.sys_operation.sys_operation import SysOperation from openjiuwen.harness.workspace.workspace import Workspace @@ -33,6 +100,13 @@ INDEX_CACHE: Dict[str, 'MemoryIndexManager'] = {} +# Module-level lock shared across all initialize() calls in this process so the +# rebuild (clear tables + reindex) is serialized per-process — distinct from the +# cross-process file lock, which serializes across processes. A per-call +# asyncio.Lock() would be a fresh instance each time and serialize nothing. +# Python 3.10+ defers loop binding to first await, so defining at import is safe. +_REBUILD_INTRAPROCESS_LOCK = asyncio.Lock() + @dataclass class SessionDeltaState: @@ -68,8 +142,11 @@ def _open_database(db_path: str) -> sqlite3.Connection: """Open SQLite database.""" ensure_dir(os.path.dirname(db_path) or ".") - conn = sqlite3.connect(db_path, check_same_thread=False) + conn = sqlite3.connect(db_path, timeout=5.0, check_same_thread=False) conn.row_factory = sqlite3.Row + # busy_timeout lets SQLite wait (up to ms) for a lock instead of failing + # immediately with "database is locked" when another process is writing. + conn.execute("PRAGMA busy_timeout=10000") conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA foreign_keys=ON") @@ -241,7 +318,29 @@ async def initialize(self) -> None: await self._initialize_provider() await self._load_vector_extension() - await self.sync(reason="initial") + # Embedding 配置变化时，清空索引表数据后重新索引，而不是在脏库上做增量 reindex。 + # 原因：旧库的向量表里残留着旧模型/旧维度的向量行，且其 rowid 与重新写入的 chunks + # rowid 错位（更新路径只 DELETE chunks 不清向量表），导致搜索时 query 向量与库里 + # 向量对不上 rowid、返回空。清空各表并重置 rowid 序列，让重新索引在干净的表上 + # 从 rowid 1 连续递增、维度统一，彻底消除脏状态。源记忆文件(.md)仍在，可完整重建。 + # 注意：必须在 _load_vector_extension 之后执行，清向量虚拟表需要扩展已加载。 + # + # 多个 manager 实例可能在不同进程里同时初始化（agent_server / gateway 各起一个， + # 或重启时新旧进程短暂并存）。若它们同时检测到配置变化并清表重建，会抢 memory.db + # 的写锁互相覆盖、写一半失败，留下空的 1024 维表。用跨进程文件锁把"检测→清表→ + # 全量重建"串行化：持锁者重建，其他进程等锁后重新检测，发现已重建好就直接跳过。 + if await self._needs_rebuild_on_config_change(): + lock_path = self._rebuild_lock_path() + async with _REBUILD_INTRAPROCESS_LOCK: # 本进程内串行（多个 manager 实例同进程时） + with _cross_process_lock(lock_path, timeout=120.0): + # 持锁后再检测一次：可能在等锁期间，别的进程已完成重建、meta 已更新。 + if await self._needs_rebuild_on_config_change(): + self._clear_index_tables() + # 在锁内一次跑完整个重建，释放锁时表已是完整的新索引，避免别的 + # 进程看到半成品（chunks 有数据但向量表空）。 + await self.sync(reason="initial") + else: + await self.sync(reason="initial") if self.settings.sync.get("watch", True): self._setup_file_watcher() @@ -250,6 +349,108 @@ async def initialize(self) -> None: logger.info(f"Memory manager initialized for agent: {self.agent_id}") + def _rebuild_lock_path(self) -> str: + """Path to the cross-process rebuild lock file (next to the DB).""" + return self.db_path + ".rebuild.lock" + + async def _needs_rebuild_on_config_change(self) -> bool: + """Whether the embedding config changed since the last index (=> must rebuild). + + Mirrors ``_should_full_reindex``'s config-comparison logic but is evaluated + at init, so a rebuild can clear the index tables rather than mutate a dirty, + mismatched index in place. Returns False when there is no prior index — in + that case ``sync`` builds the index normally, nothing to rebuild. + """ + try: + cursor = self.db.execute( + "SELECT value FROM meta WHERE key = ?", (META_KEY,) + ) + row = cursor.fetchone() + if not row: + return False + meta = json.loads(row["value"]) + if meta.get("provider") != self.provider.id: + return True + if meta.get("model") != self.provider.model: + return True + if meta.get("providerKey") != self.provider_key: + return True + if meta.get("chunkTokens") != self.settings.chunking.get("tokens"): + return True + return False + except Exception as e: + logger.warning(f"Failed to check meta for rebuild: {e}") + return False + + def _clear_index_tables(self) -> None: + """Clear index tables for rebuild, keeping the DB file. + + Called when embedding config changed. The vector virtual table is DROPPED + and the base tables cleared, so the subsequent ``sync`` rebuilds the vector + index under the new model/dims from scratch. + + Drop (not DELETE) is essential for the vector table: a stale ``chunks_vec`` + created under the old model's dims (e.g. float[1024]) would survive a + DELETE, and ``_ensure_vector_table``'s ``CREATE ... IF NOT EXISTS`` would + then no-op — leaving a 1024-dim table that rejects 2560-dim Qwen vectors + with a "Dimension mismatch" error that is swallowed at debug level, + silently producing an empty vector table. + + The FTS table, by contrast, is DELETEd (not DROPPed): it has no dimension + constraint, so a plain DELETE clears it for re-indexing. And unlike the + vector table it is never recreated after init — ``_ensure_schema`` (the + only place that creates it) runs once *before* this method; the later + ``sync``/``_index_chunk`` path only INSERTs into it. Dropping it here + would leave ``fts_available`` True but the table gone, so every FTS + insert (debug-swallowed) and keyword search (returns [] silently) fails + for the rest of this instance's life. DELETE keeps the structure so + ``_index_chunk`` can repopulate it during the rebuild. + """ + if not self.db: + return + logger.info( + f"Embedding config changed, clearing index tables for rebuild: {self.db_path}" + ) + drops = [ + (f"DROP TABLE IF EXISTS {VECTOR_TABLE}", "vector"), + ] + for sql, label in drops: + try: + self.db.execute(sql) + except Exception as e: + logger.debug(f"Skipping {label} drop: {e}") + + # Clear base tables (DELETE keeps schema; rowids reset below). The FTS + # virtual table is cleared here too (DELETE, not DROP) — see docstring. + deletions = [ + ("DELETE FROM chunks", "chunks"), + ("DELETE FROM files", "files"), + (f"DELETE FROM {FTS_TABLE}", "fts"), + (f"DELETE FROM {EMBEDDING_CACHE_TABLE}", "embedding_cache"), + ("DELETE FROM meta", "meta"), + ] + for sql, label in deletions: + try: + self.db.execute(sql) + except Exception as e: + logger.debug(f"Skipping {label} clear: {e}") + + # Reset autoincrement so re-indexed chunks get rowids from 1, matching the + # freshly written vector rows 1:1 (search joins chunks.rowid <-> vec.rowid). + for table in ("chunks", VECTOR_TABLE): + try: + self.db.execute( + "DELETE FROM sqlite_sequence WHERE name = ?", (table,) + ) + except Exception as e: + logger.debug(f"Skipping {table} sqlite_sequence reset: {e}") + + # Forget the old dims so _ensure_vector_table recreates the vec table under + # the new model's dims on first write. Combined with the DROP above, the first + # chunk write builds a correctly-dimensioned vector table from scratch. + self.vector_dims = None + self.db.commit() + def _resolve_db_path(self) -> str: """Resolve database path. @@ -414,8 +615,8 @@ def _ensure_vector_table(self, dims: int) -> bool: if self.vector_dims is not None and self.vector_dims != dims: try: self.db.execute(f"DROP TABLE IF EXISTS {VECTOR_TABLE}") - except sqlite3.Error: - pass + except sqlite3.Error as e: + logger.debug(f"Skipping {VECTOR_TABLE} drop before recreate: {e}") self.db.execute(f""" CREATE VIRTUAL TABLE IF NOT EXISTS {VECTOR_TABLE} USING vec0( @@ -635,13 +836,19 @@ async def _should_full_reindex(self) -> bool: return True async def _run_reindex(self) -> None: - """Run full reindex.""" + """Run full reindex. + + Forces re-indexing of every file regardless of hash so that chunks are + re-embedded under the current model/dims — this is what "re-index on + vector config switch" must mean, otherwise unchanged files keep stale + embeddings from the old model and search returns empty results. + """ if "memory" in self.settings.sources: - await self._sync_memory_files() + await self._sync_memory_files(force=True) self.dirty = False if "sessions" in self.settings.sources: - await self._sync_session_files() + await self._sync_session_files(force=True) meta = { "provider": self.provider.id if self.provider else None, @@ -655,15 +862,20 @@ async def _run_reindex(self) -> None: self._write_meta(meta) - async def _sync_memory_files(self) -> None: + async def _sync_memory_files(self, force: bool = False) -> None: """Sync memory files. All session files (YYYY-MM-DD.md) are indexed for search. Recent session files (today + yesterday) are also loaded for context. + + When ``force`` is True (e.g. embedding config changed -> full reindex), + every file is re-indexed even if its hash is unchanged, so that chunks + get re-embedded under the new model/dims. This is the intended meaning + of "re-index already-indexed files on vector config switch" — not skip. """ files = list_memory_files(self.workspace, node_name=self.node_name) - logger.debug(f"Syncing {len(files)} memory files") + logger.debug(f"Syncing {len(files)} memory files (force={force})") active_paths = set() @@ -672,14 +884,15 @@ async def _sync_memory_files(self) -> None: entry = await build_file_entry(filepath, base_dir) active_paths.add(entry["path"]) - cursor = self.db.execute( - "SELECT hash FROM files WHERE path = ? AND source = ?", - (entry["path"], "memory") - ) - row = cursor.fetchone() + if not force: + cursor = self.db.execute( + "SELECT hash FROM files WHERE path = ? AND source = ?", + (entry["path"], "memory") + ) + row = cursor.fetchone() - if row and row["hash"] == entry["hash"]: - continue + if row and row["hash"] == entry["hash"]: + continue await self._index_file(entry, "memory") @@ -695,8 +908,11 @@ def _get_base_dir_for_file(self, filepath: str) -> str: return str(self.workspace.root_path) return self.memory_dir - async def _sync_session_files(self) -> None: - """Sync session transcript files.""" + async def _sync_session_files(self, force: bool = False) -> None: + """Sync session transcript files. + + See ``_sync_memory_files`` for the meaning of ``force``. + """ sessions_dir = os.path.join(self.memory_dir, "sessions") if not os.path.exists(sessions_dir): return @@ -707,21 +923,22 @@ async def _sync_session_files(self) -> None: if f.endswith(".jsonl"): session_files.append(os.path.join(root, f)) - logger.debug(f"Syncing {len(session_files)} session files") + logger.debug(f"Syncing {len(session_files)} session files (force={force})") active_paths = set() for session_file in session_files: entry = await build_file_entry(session_file, self.memory_dir) active_paths.add(entry["path"]) - cursor = self.db.execute( - "SELECT hash FROM files WHERE path = ? AND source = ?", - (entry["path"], "sessions") - ) - row = cursor.fetchone() + if not force: + cursor = self.db.execute( + "SELECT hash FROM files WHERE path = ? AND source = ?", + (entry["path"], "sessions") + ) + row = cursor.fetchone() - if row and row["hash"] == entry["hash"]: - continue + if row and row["hash"] == entry["hash"]: + continue await self._index_file(entry, "sessions") @@ -740,12 +957,31 @@ async def _index_file(self, entry: Dict[str, Any], source: str) -> None: logger.error("no available sys_operation when _index_file") chunks = chunk_markdown(content, self.settings.chunking) + # Before deleting the old chunks, drop their vector rows too — + # otherwise re-indexing a file (force reindex / hash change) leaves + # orphan vectors in chunks_vec whose rowids no longer match any chunk + # (the new chunks get fresh rowids), polluting the index and breaking + # the rowid join used by search. + old_rowids = [] + for r in self.db.execute( + "SELECT rowid FROM chunks WHERE path = ?", (entry["path"],) + ).fetchall(): + old_rowids.append(r["rowid"]) + if self.vector_available: + for rid in old_rowids: + try: + self.db.execute( + f"DELETE FROM {VECTOR_TABLE} WHERE rowid = ?", (rid,) + ) + except sqlite3.Error as e: + logger.debug(f"Skipping orphan vector row delete (rowid={rid}): {e}") + self.db.execute("DELETE FROM chunks WHERE path = ?", (entry["path"],)) if self.fts_available: try: self.db.execute(f"DELETE FROM {FTS_TABLE} WHERE path = ?", (entry["path"],)) - except sqlite3.Error: - pass + except sqlite3.Error as e: + logger.debug(f"Skipping FTS delete for {entry['path']}: {e}") for chunk in chunks: await self._index_chunk(entry["path"], source, chunk) @@ -818,16 +1054,16 @@ def _remove_file_from_index(self, file_path: str) -> None: for row in cursor.fetchall(): try: self.db.execute(f"DELETE FROM {VECTOR_TABLE} WHERE rowid = ?", (row["rowid"],)) - except sqlite3.Error: - pass + except sqlite3.Error as e: + logger.debug(f"Skipping vector row delete for {file_path} (rowid={row['rowid']}): {e}") if self.fts_available: cursor = self.db.execute("SELECT rowid FROM chunks WHERE path = ?", (file_path,)) for row in cursor.fetchall(): try: self.db.execute(f"DELETE FROM {FTS_TABLE} WHERE rowid = ?", (row["rowid"],)) - except sqlite3.Error: - pass + except sqlite3.Error as e: + logger.debug(f"Skipping FTS row delete for {file_path} (rowid={row['rowid']}): {e}") self.db.execute("DELETE FROM chunks WHERE path = ?", (file_path,)) self.db.execute("DELETE FROM files WHERE path = ?", (file_path,)) From 3d8462a5c455188dd260d3282c425e351e93e0fa Mon Sep 17 00:00:00 2001 From: l_x_d Date: Thu, 16 Jul 2026 09:28:05 +0800 Subject: [PATCH 003/291] fix(context): add switchable refactored processors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs: #1249 **What type of PR is this?** /kind