diff --git a/backend/app/database/db.py b/backend/app/database/db.py index 8183929..9df5381 100644 --- a/backend/app/database/db.py +++ b/backend/app/database/db.py @@ -26,8 +26,28 @@ def _load_extension(conn: sqlite3.Connection) -> None: def connect() -> sqlite3.Connection: settings = get_settings() - settings.db_path.parent.mkdir(parents=True, exist_ok=True) - conn = sqlite3.connect(settings.db_path) + return _connect_path(settings.db_path) + + +def connect_knowledge() -> sqlite3.Connection: + """Desktop projections never share note or vector rows between Vaults.""" + settings = get_settings() + if settings.environment != 'desktop': + return connect() + from app import host_bridge + from app.errors import ApiError + from uuid import UUID + try: + vault = str(UUID(host_bridge.vault_id.get() or '')) + except ValueError: + raise ApiError(409, 'WORKSPACE_NOT_OPEN', '请先打开授权工作区。') from None + # This database also holds durable logical records (tasks); never delete it as a cache. + return _connect_path(settings.data_dir / 'vault-state' / vault / 'core.sqlite3') + + +def _connect_path(path) -> sqlite3.Connection: + path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(path) conn.row_factory = sqlite3.Row # 关闭 Python sqlite3 的隐式事务,提交时机由 transaction() 或显式 commit 控制。 conn.isolation_level = None diff --git a/backend/app/repository.py b/backend/app/repository.py index bb8213c..fa82787 100644 --- a/backend/app/repository.py +++ b/backend/app/repository.py @@ -14,7 +14,7 @@ from dataclasses import dataclass, field from datetime import datetime from app.contracts import NoteBlock -from app.database.db import connect, transaction +from app.database.db import connect_knowledge as connect, transaction from app.textutils import segment diff --git a/backend/app/retrieval/engine.py b/backend/app/retrieval/engine.py index 82f1008..9ee4eec 100644 --- a/backend/app/retrieval/engine.py +++ b/backend/app/retrieval/engine.py @@ -55,6 +55,10 @@ class RetrievalEngine: @track_search async def search(self, request: SearchRequest) -> SearchResponse: + from app.config import get_settings + if get_settings().environment == 'desktop': + from app.services.desktop_projection import refresh + await refresh() if request.mode == SearchMode.fts: return self._search_fts(request) diff --git a/backend/app/retrieval/routed_vectors.py b/backend/app/retrieval/routed_vectors.py index ab959be..4233e21 100644 --- a/backend/app/retrieval/routed_vectors.py +++ b/backend/app/retrieval/routed_vectors.py @@ -17,7 +17,7 @@ import sqlite3 from dataclasses import dataclass from typing import Protocol -from app.database.db import connect, transaction +from app.database.db import connect_knowledge as connect, transaction from app.errors import ApiError from app.operation_logs import log_event from app.retrieval.vectorstore import VectorHit diff --git a/backend/app/retrieval/vectorstore.py b/backend/app/retrieval/vectorstore.py index 58a4a1c..fbfad85 100644 --- a/backend/app/retrieval/vectorstore.py +++ b/backend/app/retrieval/vectorstore.py @@ -13,7 +13,7 @@ from typing import Protocol, runtime_checkable import sqlite_vec -from app.database.db import connect, transaction +from app.database.db import connect_knowledge as connect, transaction @dataclass diff --git a/backend/app/services/desktop_projection.py b/backend/app/services/desktop_projection.py new file mode 100644 index 0000000..8f9a709 --- /dev/null +++ b/backend/app/services/desktop_projection.py @@ -0,0 +1,65 @@ +"""Rebuildable per-Vault FTS projection, sourced only through the Host broker.""" +from __future__ import annotations +import asyncio +from app import repository +from app.database.db import connect_knowledge, transaction +from app.knowledge.parser import parse_note +from app.services import desktop_notes +from app.services.coordination import vault_mutation_lock + + +def entries(): + result, offset = [], 0 + while True: + page = desktop_notes.call('list', offset=offset, limit=1000) + result.extend(page['items']) + offset += len(page['items']) + if offset >= page['total'] or not page['items']: return result + + +def _refresh(): + current = entries() # Always validates authorization, including when the cache is current. + conn = connect_knowledge() + try: + conn.execute('CREATE TABLE IF NOT EXISTS host_projection (file_id TEXT PRIMARY KEY, hash TEXT NOT NULL, path TEXT NOT NULL)') + old = {row['file_id']: (row['hash'], row['path']) for row in conn.execute('SELECT * FROM host_projection')} + changed = [] + for entry in current: + if old.get(entry['file_id']) == (entry['hash'], entry['path']): continue + document = desktop_notes.call('read', file_id=entry['file_id']) + note = desktop_notes.note_from_document(document) + parsed = parse_note(markdown=note.markdown, file_path=note.file_path, + folder=note.file_path.rpartition('/')[0], note_id=note.note_id, + tags=note.tags, created_at=note.created_at, updated_at=note.updated_at) + changed.append((document, parsed)) + removed = set(old) - {entry['file_id'] for entry in current} + # Content is verified before starting the projection transaction. No model/network IO inside. + with transaction(conn): + for file_id in removed: + for block_id in repository.delete_note(file_id, conn=conn): + conn.execute('DELETE FROM vec_blocks WHERE block_id=?', [block_id]) + conn.execute('DELETE FROM host_projection WHERE file_id=?', [file_id]) + for document, parsed in changed: + old_ids = repository.replace_note_metadata(conn=conn, note_id=parsed.note_id, title=parsed.title, + file_path=parsed.file_path, folder=parsed.folder, tags=parsed.tags, created_at=parsed.created_at, + updated_at=parsed.updated_at, blocks=parsed.blocks) + for block_id in old_ids: + conn.execute('DELETE FROM vec_blocks WHERE block_id=?', [block_id]) + conn.execute('UPDATE blocks SET embedding_local_only=? WHERE note_id=?', (int(parsed.embedding_local_only), parsed.note_id)) + conn.execute('INSERT OR REPLACE INTO host_projection VALUES (?,?,?)', (parsed.note_id, document['hash'], parsed.file_path)) + if removed or changed: + repository.set_index_meta({'workspace_vectors_pending': '1'}, conn=conn) + finally: + conn.close() + + +async def refresh(): + async with vault_mutation_lock(): + work = asyncio.create_task(asyncio.to_thread(_refresh)) + # Keep the projection gate until the worker has finished even if the request is cancelled. + cancelled = False + while not work.done(): + try: await asyncio.shield(work) + except asyncio.CancelledError: cancelled = True + work.result() + if cancelled: raise asyncio.CancelledError diff --git a/backend/app/services/index_service.py b/backend/app/services/index_service.py index db80a59..5b38e0b 100644 --- a/backend/app/services/index_service.py +++ b/backend/app/services/index_service.py @@ -16,7 +16,7 @@ from app.contracts import IndexJob, IndexRebuildRequest, IndexStatus from app.errors import ApiError from app.knowledge.parser import parse_note from app.services.note_service import index_note, prepare_note_index -from app.database.db import connect, transaction +from app.database.db import connect_knowledge as connect, transaction from app.services.coordination import vault_mutation_lock from app.retrieval.vectorstore import SqliteVecStore from app.local_models.runtime import LocalEmbedding @@ -47,6 +47,14 @@ def _scan_vault() -> list[tuple[str, str, str, datetime, datetime]]: 先读入内存:若文件读取失败,rebuild 尚未清空旧索引,不会造成数据损失。 """ + if get_settings().environment == 'desktop': + from app.services.desktop_projection import entries + from app.services import desktop_notes + result = [] + for entry in entries(): + note = desktop_notes.note_from_document(desktop_notes.call('read', file_id=entry['file_id'])) + result.append((note.file_path, note.file_path.rpartition('/')[0], note.markdown, note.created_at, note.updated_at)) + return result vault = get_settings().vault_path.resolve() result: list[tuple[str, str, str, datetime, datetime]] = [] if not vault.exists(): @@ -80,8 +88,12 @@ async def rebuild(request: IndexRebuildRequest) -> IndexJob: {"scope": request.scope, "note_ids": request.note_ids}, ) + if get_settings().environment == 'desktop': + from app.services.desktop_projection import refresh + await refresh() docs = _scan_vault() - saved_records = {key: repository.get_note_record(key) for key in _pending_notes()} + record_ids = [entry.note_id for entry in repository.list_note_locations()] if get_settings().environment == 'desktop' else _pending_notes() + saved_records = {key: repository.get_note_record(key) for key in record_ids} saved_paths = {record.file_path: record for record in saved_records.values() if record is not None} _active_job_id = job_id @@ -117,7 +129,8 @@ async def rebuild(request: IndexRebuildRequest) -> IndexJob: # All network/model awaits precede the transaction. The concrete SQLite # methods below complete synchronously despite their async interfaces. async with vault_mutation_lock(): - if _scan_vault() != docs or saved_records != {key: repository.get_note_record(key) for key in _pending_notes()}: + current_ids = [entry.note_id for entry in repository.list_note_locations()] if get_settings().environment == 'desktop' else _pending_notes() + if _scan_vault() != docs or saved_records != {key: repository.get_note_record(key) for key in current_ids}: raise ApiError(409, "INDEX_SNAPSHOT_CHANGED", "笔记在计算期间发生变化,稍后重新计算。") conn = connect() try: diff --git a/backend/app/services/note_service.py b/backend/app/services/note_service.py index a7a7901..497e38a 100644 --- a/backend/app/services/note_service.py +++ b/backend/app/services/note_service.py @@ -14,7 +14,7 @@ from uuid import uuid4 from app import repository from app.contracts import Note, NoteBlock, NoteSummary -from app.database.db import connect, transaction +from app.database.db import connect_knowledge as connect, transaction from app.errors import ApiError from app.knowledge.parser import ParsedNote, parse_note from app.local_models.runtime import LocalEmbedding, background_embeddings diff --git a/backend/app/services/task_service.py b/backend/app/services/task_service.py index 830acb4..c981274 100644 --- a/backend/app/services/task_service.py +++ b/backend/app/services/task_service.py @@ -9,7 +9,7 @@ from weakref import WeakKeyDictionary from app import repository from app.contracts import Task, TaskStatus -from app.database.db import connect, transaction +from app.database.db import connect_knowledge as connect, transaction from app.errors import ApiError from app.operation_logs import log_event @@ -39,6 +39,13 @@ def _now() -> datetime: return datetime.now(timezone.utc) +def _prepare_note_link(note_id: str | None) -> None: + from app.config import get_settings + if note_id and get_settings().environment == 'desktop': + from app.services.desktop_projection import _refresh + _refresh() + + def _task_from_row(row) -> Task: return Task( task_id=row["task_id"], @@ -56,6 +63,7 @@ def create_task( *, title: str, description: str = "", note_id: str | None = None, due_at: datetime | None = None, ) -> Task: + _prepare_note_link(note_id) if note_id and repository.get_note_record(note_id) is None: raise ApiError(404, "RESOURCE_NOT_FOUND", "note not found", {"note_id": note_id}) task_id = f"task_{uuid4().hex}" @@ -109,6 +117,7 @@ def update_task(task_id: str, values: dict[str, object]) -> Task: raise ApiError(404, "RESOURCE_NOT_FOUND", "task not found", {"task_id": task_id}) if "note_id" in values and values["note_id"]: note_id = str(values["note_id"]) + _prepare_note_link(note_id) if repository.get_note_record(note_id) is None: raise ApiError(404, "RESOURCE_NOT_FOUND", "note not found", {"note_id": note_id}) if values.get("title") is None: diff --git a/backend/tests/test_desktop_projection.py b/backend/tests/test_desktop_projection.py new file mode 100644 index 0000000..42f1a9f --- /dev/null +++ b/backend/tests/test_desktop_projection.py @@ -0,0 +1,76 @@ +import asyncio +from dataclasses import replace +from hashlib import sha256 +from uuid import uuid4 + +from app import host_bridge, repository +from app.config import get_settings +from app.database import db +from app.contracts import SearchRequest, SearchMode +from app.retrieval.engine import engine +from app.services import desktop_notes, desktop_projection, index_service, note_service +from app.retrieval.embedding import HashEmbeddingProvider + + +def test_projection_isolates_same_path_and_refreshes_changed_deleted_content(tmp_path, monkeypatch): + settings = replace(get_settings(), environment='desktop', data_dir=tmp_path, db_path=tmp_path/'global.sqlite3') + monkeypatch.setattr(db, 'get_settings', lambda: settings) + monkeypatch.setattr('app.config.get_settings', lambda: settings) + first, second = str(uuid4()), str(uuid4()) + documents = {first: {'file_id': 'file-a', 'path': 'same.md', 'content': 'uniquefirsttoken', 'created_at': 0, 'updated_at': 1}, + second: {'file_id': 'file-b', 'path': 'same.md', 'content': 'uniquesecondtoken', 'created_at': 0, 'updated_at': 1}} + def call(method, **params): + doc = documents.get(host_bridge.vault_id.get()) + if doc: doc['hash'] = sha256(doc['content'].encode()).hexdigest() + if method == 'list': return {'items': [doc] if doc else [], 'total': int(doc is not None)} + assert method == 'read' and doc['file_id'] == params['file_id'] + return doc + monkeypatch.setattr(desktop_notes, 'call', call) + def search(vault, query): + token = host_bridge.vault_id.set(vault) + try: return asyncio.run(engine.search(SearchRequest(query=query, mode=SearchMode.fts))) + finally: host_bridge.vault_id.reset(token) + assert 'file-a' in search(first, 'uniquefirsttoken').model_dump_json() + assert 'file-a' not in search(second, 'uniquefirsttoken').model_dump_json() + assert 'file-b' in search(second, 'uniquesecondtoken').model_dump_json() + documents[first]['content'] = 'replacementtoken' + assert 'file-a' not in search(first, 'uniquefirsttoken').model_dump_json() + assert 'file-a' in search(first, 'replacementtoken').model_dump_json() + del documents[first] + assert 'file-a' not in search(first, 'replacementtoken').model_dump_json() + assert (tmp_path/'vault-state'/first/'core.sqlite3').is_file() + assert (tmp_path/'vault-state'/second/'core.sqlite3').is_file() + from app.services import task_service + token = host_bridge.vault_id.set(second) + try: + task = task_service.create_task(title='Scoped task', note_id='file-b') + assert task.note_id == 'file-b' + finally: + host_bridge.vault_id.reset(token) + token = host_bridge.vault_id.set(first) + try: + assert task_service.get_task(task.task_id) is None + finally: + host_bridge.vault_id.reset(token) + + +def test_desktop_semantic_rebuild_preserves_host_file_id(tmp_path, monkeypatch): + settings = replace(get_settings(), environment='desktop', data_dir=tmp_path, db_path=tmp_path/'global.sqlite3') + monkeypatch.setattr(db, 'get_settings', lambda: settings) + monkeypatch.setattr('app.config.get_settings', lambda: settings) + monkeypatch.setattr(index_service, 'get_settings', lambda: settings) + document = {'file_id': 'stable-host-id', 'path': 'same.md', 'hash': sha256(b'test note').hexdigest(), + 'content': 'test note', 'created_at': 0, 'updated_at': 1} + monkeypatch.setattr(desktop_notes, 'call', lambda method, **params: {'items': [document], 'total': 1} if method == 'list' else document) + monkeypatch.setattr(note_service, 'embedding', HashEmbeddingProvider()) + async def no_remote(*args, **kwargs): return None + monkeypatch.setattr('app.retrieval.routed_vectors.embed_remote', no_remote) + from app.contracts import IndexRebuildRequest + token = host_bridge.vault_id.set(str(uuid4())) + try: + result = asyncio.run(index_service.rebuild(IndexRebuildRequest())) + assert result.status == 'completed' + assert repository.get_note_record('stable-host-id') is not None + assert [record.note_id for record in repository.list_note_locations()] == ['stable-host-id'] + finally: + host_bridge.vault_id.reset(token) diff --git a/docs/contracts/Host-v1契约.md b/docs/contracts/Host-v1契约.md index 231c0ff..6591eb9 100644 --- a/docs/contracts/Host-v1契约.md +++ b/docs/contracts/Host-v1契约.md @@ -36,7 +36,9 @@ Windows 使用独立隐藏窗口注册 WTS 会话通知,锁屏、注销及本 Workspace schema 2 新增持久操作回执。`workspace_write` 可选 operation_id(UUID),同 ID、同载荷重放返回原提交条目;不同载荷返回 OPERATION_PAYLOAD_CONFLICT。回执与文件元数据、outbox 在同一 SQLite 事务提交,即使之后再次编辑,查询仍返回对应操作的原始结果。schema 1 升级前用 SQLite VACUUM INTO 保存一致备份;旧 schema 1 Host 拒绝写入 schema 2。 -Core 的 `workspace.list/read/write/mutate/operation` RPC 通过受控管道转发,严格拒绝未知字段并校验 Host HTTP 请求捕获的 vault_id。切换或撤销工作区后旧任务返回 VAULT_PERMISSION_CHANGED。普通 HTTP 与 SSE 均由 Rust 添加工作区头,WebView 不能指定。Core 笔记 CRUD 不再写入 unbound-vault,也不从旧 Core 笔记索引回退读取;Markdown 元数据随正文保存,笔记 ID 使用 Rust file_id。当前 Core 笔记 RPC 单篇上限 1 MiB,管道帧上限 8 MiB;大媒体传输与检索投影另行实现,不宣称已满足完整 A-03。 +Core 的 `workspace.list/read/write/mutate/operation` RPC 通过受控管道转发,严格拒绝未知字段并校验 Host HTTP 请求捕获的 vault_id。切换或撤销工作区后旧任务返回 VAULT_PERMISSION_CHANGED。普通 HTTP 与 SSE 均由 Rust 添加工作区头,WebView 不能指定。Core 笔记 CRUD 不再写入 unbound-vault,也不从旧 Core 笔记索引回退读取;Markdown 元数据随正文保存,笔记 ID 使用 Rust file_id。当前 Core 笔记 RPC 单篇上限 1 MiB,管道帧上限 8 MiB;大媒体传输与完整取消提交确认另行实现,不宣称已满足完整 A-03。 + +桌面检索与任务使用 Core 数据目录中的 `vault-state//core.sqlite3`,不同 Vault 不共享笔记/向量/任务行。该库含持久任务,不能作为缓存整体删除。搜索前通过 Host 扫描摘要并事务更新全文投影,删除与修改使旧向量失效;向量重建通过 Host 读取正文并保持稳定 file_id。模型配置和凭据仍属设备配置,不随笔记库路由。没有授权 Vault 的知识请求明确失败,不回退全局索引。旧版无归属的 Core 任务不会自动指派给任意 Vault,其迁移入口仍待完成。 OS 文件锁配合 Rust Mutex 维持单实例 Vault 写入。Web `serialized_vault_mutation` 使用同一 OS 文件锁;发现 `.ainote/host.sqlite3` 后拒绝 Web 写入,不自动降级。已有个人 Vault 不会被测试读取或迁移。 diff --git a/docs/development/OpenNexus生产化实施进度-2026-09-08.md b/docs/development/OpenNexus生产化实施进度-2026-09-08.md index a761c0f..dc3045d 100644 --- a/docs/development/OpenNexus生产化实施进度-2026-09-08.md +++ b/docs/development/OpenNexus生产化实施进度-2026-09-08.md @@ -6,6 +6,9 @@ ## 持续实施增量 +- 桌面全文/向量投影按 Vault 隔离,搜索前经 Host 对账文件摘要;向量重建从 Host 读取正文并保留 file_id。任务和笔记关联使用同一 Vault 的持久库。新增测试覆盖同路径双 Vault 隔离、变更/删除刷新、稳定 ID 和任务跨 Vault 不可见;真实 Core 测试增加全文搜索与删除后的检索验证。 +- 检索隔离增量最终后端全量 899 项通过,真实 Core/Vault 搜索集成和 Rust desktop 全目标 Clippy 通过。语义重建的稳定 ID 测试使用显式测试 Embedding,不作为真实模型质量或性能证据。 + - Core 笔记 CRUD 已经由受控管道接入当前授权 Rust Vault,保留稳定 file_id,支持 CAS 和跨 Vault 拒绝。新增 schema 2 操作回执,记录写入/移动/删除提交结果,升级前保存 schema 1 一致备份。同一写入操作重放 100 次不新增 outbox,也不覆盖后续编辑。 - 真实 Python Core + 临时 Rust Vault 集成测试覆盖创建、列表、修改、并发冲突、移动、删除、跨 Vault 拒绝和提交查询;Rust desktop 全目标 35 项通过。Core 检索投影、后台任务取消的完整提交边界、大媒体和 UI 提交确认仍需继续实现,不能将 CRUD 接通等同整个 AI 工作流验收完成。 - 此增量后端全量 897 项、前端全量 94 文件/506 项通过;Rust desktop 全目标 Clippy 与文档链接检查通过。旧 Core 打包产物尚未重新构建,不将开发解释器进程测试作为最新安装包证据。 diff --git a/frontend/src-tauri/tests/core_workspace.rs b/frontend/src-tauri/tests/core_workspace.rs index a94be1a..3466a35 100644 --- a/frontend/src-tauri/tests/core_workspace.rs +++ b/frontend/src-tauri/tests/core_workspace.rs @@ -71,6 +71,17 @@ async fn real_core_notes_roundtrip_only_through_bound_host_and_confirm_commits() let original = workspace.lock().unwrap().read("Core fixture.md").unwrap(); assert_eq!(original.entry.file_id, file_id); assert!(original.content.contains("original")); + let (status, search) = request( + &mut core, + "POST", + "/api/search", + &vault, + &uuid::Uuid::new_v4().to_string(), + Some(json!({"query":"original","mode":"fts"})), + ) + .await; + assert_eq!(status, 200, "{search}"); + assert!(search.to_string().contains(file_id)); assert_eq!( workspace .lock() @@ -146,6 +157,17 @@ async fn real_core_notes_roundtrip_only_through_bound_host_and_confirm_commits() .await; assert_eq!(status, 200, "{deleted}"); assert!(!root.join("nested/Core fixture.md").exists()); + let (status, search) = request( + &mut core, + "POST", + "/api/search", + &vault, + &uuid::Uuid::new_v4().to_string(), + Some(json!({"query":"original","mode":"fts"})), + ) + .await; + assert_eq!(status, 200, "{search}"); + assert!(!search.to_string().contains(file_id)); assert_eq!(workspace.lock().unwrap().pending_count().unwrap(), 4); assert!(!temp .path()