feat: 按 Vault 隔离桌面检索投影与任务链接

This commit is contained in:
2026-09-08 15:12:25 +08:00
parent bc1403bc85
commit 495cc4a24e
11 changed files with 219 additions and 10 deletions
+22 -2
View File
@@ -26,8 +26,28 @@ def _load_extension(conn: sqlite3.Connection) -> None:
def connect() -> sqlite3.Connection: def connect() -> sqlite3.Connection:
settings = get_settings() settings = get_settings()
settings.db_path.parent.mkdir(parents=True, exist_ok=True) return _connect_path(settings.db_path)
conn = sqlite3.connect(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 conn.row_factory = sqlite3.Row
# 关闭 Python sqlite3 的隐式事务,提交时机由 transaction() 或显式 commit 控制。 # 关闭 Python sqlite3 的隐式事务,提交时机由 transaction() 或显式 commit 控制。
conn.isolation_level = None conn.isolation_level = None
+1 -1
View File
@@ -14,7 +14,7 @@ from dataclasses import dataclass, field
from datetime import datetime from datetime import datetime
from app.contracts import NoteBlock 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 from app.textutils import segment
+4
View File
@@ -55,6 +55,10 @@ class RetrievalEngine:
@track_search @track_search
async def search(self, request: SearchRequest) -> SearchResponse: 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: if request.mode == SearchMode.fts:
return self._search_fts(request) return self._search_fts(request)
+1 -1
View File
@@ -17,7 +17,7 @@ import sqlite3
from dataclasses import dataclass from dataclasses import dataclass
from typing import Protocol 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.errors import ApiError
from app.operation_logs import log_event from app.operation_logs import log_event
from app.retrieval.vectorstore import VectorHit from app.retrieval.vectorstore import VectorHit
+1 -1
View File
@@ -13,7 +13,7 @@ from typing import Protocol, runtime_checkable
import sqlite_vec import sqlite_vec
from app.database.db import connect, transaction from app.database.db import connect_knowledge as connect, transaction
@dataclass @dataclass
@@ -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
+16 -3
View File
@@ -16,7 +16,7 @@ from app.contracts import IndexJob, IndexRebuildRequest, IndexStatus
from app.errors import ApiError from app.errors import ApiError
from app.knowledge.parser import parse_note from app.knowledge.parser import parse_note
from app.services.note_service import index_note, prepare_note_index 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.services.coordination import vault_mutation_lock
from app.retrieval.vectorstore import SqliteVecStore from app.retrieval.vectorstore import SqliteVecStore
from app.local_models.runtime import LocalEmbedding from app.local_models.runtime import LocalEmbedding
@@ -47,6 +47,14 @@ def _scan_vault() -> list[tuple[str, str, str, datetime, datetime]]:
先读入内存若文件读取失败rebuild 尚未清空旧索引不会造成数据损失 先读入内存若文件读取失败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() vault = get_settings().vault_path.resolve()
result: list[tuple[str, str, str, datetime, datetime]] = [] result: list[tuple[str, str, str, datetime, datetime]] = []
if not vault.exists(): if not vault.exists():
@@ -80,8 +88,12 @@ async def rebuild(request: IndexRebuildRequest) -> IndexJob:
{"scope": request.scope, "note_ids": request.note_ids}, {"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() 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} saved_paths = {record.file_path: record for record in saved_records.values() if record is not None}
_active_job_id = job_id _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 # All network/model awaits precede the transaction. The concrete SQLite
# methods below complete synchronously despite their async interfaces. # methods below complete synchronously despite their async interfaces.
async with vault_mutation_lock(): 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", "笔记在计算期间发生变化,稍后重新计算。") raise ApiError(409, "INDEX_SNAPSHOT_CHANGED", "笔记在计算期间发生变化,稍后重新计算。")
conn = connect() conn = connect()
try: try:
+1 -1
View File
@@ -14,7 +14,7 @@ from uuid import uuid4
from app import repository from app import repository
from app.contracts import Note, NoteBlock, NoteSummary 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.errors import ApiError
from app.knowledge.parser import ParsedNote, parse_note from app.knowledge.parser import ParsedNote, parse_note
from app.local_models.runtime import LocalEmbedding, background_embeddings from app.local_models.runtime import LocalEmbedding, background_embeddings
+10 -1
View File
@@ -9,7 +9,7 @@ from weakref import WeakKeyDictionary
from app import repository from app import repository
from app.contracts import Task, TaskStatus 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.errors import ApiError
from app.operation_logs import log_event from app.operation_logs import log_event
@@ -39,6 +39,13 @@ def _now() -> datetime:
return datetime.now(timezone.utc) 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: def _task_from_row(row) -> Task:
return Task( return Task(
task_id=row["task_id"], task_id=row["task_id"],
@@ -56,6 +63,7 @@ def create_task(
*, title: str, description: str = "", note_id: str | None = None, *, title: str, description: str = "", note_id: str | None = None,
due_at: datetime | None = None, due_at: datetime | None = None,
) -> Task: ) -> Task:
_prepare_note_link(note_id)
if note_id and repository.get_note_record(note_id) is None: 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}) raise ApiError(404, "RESOURCE_NOT_FOUND", "note not found", {"note_id": note_id})
task_id = f"task_{uuid4().hex}" 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}) raise ApiError(404, "RESOURCE_NOT_FOUND", "task not found", {"task_id": task_id})
if "note_id" in values and values["note_id"]: if "note_id" in values and values["note_id"]:
note_id = str(values["note_id"]) note_id = str(values["note_id"])
_prepare_note_link(note_id)
if repository.get_note_record(note_id) is None: if repository.get_note_record(note_id) is None:
raise ApiError(404, "RESOURCE_NOT_FOUND", "note not found", {"note_id": note_id}) raise ApiError(404, "RESOURCE_NOT_FOUND", "note not found", {"note_id": note_id})
if values.get("title") is None: if values.get("title") is None:
+76
View File
@@ -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)
@@ -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(); let original = workspace.lock().unwrap().read("Core fixture.md").unwrap();
assert_eq!(original.entry.file_id, file_id); assert_eq!(original.entry.file_id, file_id);
assert!(original.content.contains("original")); 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!( assert_eq!(
workspace workspace
.lock() .lock()
@@ -146,6 +157,17 @@ async fn real_core_notes_roundtrip_only_through_bound_host_and_confirm_commits()
.await; .await;
assert_eq!(status, 200, "{deleted}"); assert_eq!(status, 200, "{deleted}");
assert!(!root.join("nested/Core fixture.md").exists()); 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_eq!(workspace.lock().unwrap().pending_count().unwrap(), 4);
assert!(!temp assert!(!temp
.path() .path()