diff --git a/hoard/cli/main.py b/hoard/cli/main.py index 83e8ea4..8cf7d1b 100644 --- a/hoard/cli/main.py +++ b/hoard/cli/main.py @@ -75,7 +75,13 @@ def _call_mcp(config: dict, method: str, params: dict, *, token_override: str | if isinstance(payload, dict) and payload.get("error"): message = payload["error"].get("message", payload["error"]) raise click.ClickException(str(message)) - return payload.get("result", payload) + result = payload.get("result", payload) + if method == "tools/call" and isinstance(result, dict) and "content" in result: + try: + return json.loads(result["content"][0]["text"]) + except (KeyError, IndexError, json.JSONDecodeError): + pass + return result def _call_admin(config: dict, method: str, params: dict) -> dict: @@ -1946,7 +1952,14 @@ def _call_mcp(method: str, params: dict) -> dict: def _call_tool(name: str, arguments: dict) -> dict: - return _call_mcp("tools/call", {"name": name, "arguments": arguments}) + resp = _call_mcp("tools/call", {"name": name, "arguments": arguments}) + result = resp.get("result", resp) + if isinstance(result, dict) and "content" in result: + try: + return json.loads(result["content"][0]["text"]) + except (KeyError, IndexError, json.JSONDecodeError): + pass + return result def search(query: str, limit: int = 10) -> dict: diff --git a/hoard/core/config.py b/hoard/core/config.py index d983345..70140fd 100644 --- a/hoard/core/config.py +++ b/hoard/core/config.py @@ -130,6 +130,9 @@ "top_k": 5, "contradiction_threshold": 0.7, }, + "duplicates": { + "similarity_threshold": 0.85, + }, "sensitivity": { "sensitive_max_ttl_days": 90, "restricted_max_ttl_days": 30, diff --git a/hoard/core/mcp/server.py b/hoard/core/mcp/server.py index c1d3b1f..4cd7005 100644 --- a/hoard/core/mcp/server.py +++ b/hoard/core/mcp/server.py @@ -186,7 +186,8 @@ def _dispatch_jsonrpc(self, payload: Dict[str, Any], conn, limiter, token): chunks_returned=count_chunks(response), bytes_returned=len(response_bytes), ) - return {"jsonrpc": "2.0", "id": msg_id, "result": response} + content_result = _wrap_tool_result(response) + return {"jsonrpc": "2.0", "id": msg_id, "result": content_result} except ValueError as exc: self._log_access( tool=tool_name, @@ -325,6 +326,10 @@ def run_server( server.serve_forever() +def _wrap_tool_result(response: Dict[str, Any]) -> Dict[str, Any]: + return {"content": [{"type": "text", "text": json.dumps(response)}]} + + def _jsonrpc_error(payload: Dict[str, Any], code: int, message: str) -> Dict[str, Any]: msg_id = payload.get("id") if isinstance(payload, dict) else None return {"jsonrpc": "2.0", "id": msg_id, "error": {"code": code, "message": message}} diff --git a/hoard/core/mcp/stdio.py b/hoard/core/mcp/stdio.py index ed26d14..e292f87 100644 --- a/hoard/core/mcp/stdio.py +++ b/hoard/core/mcp/stdio.py @@ -145,7 +145,8 @@ def _handle_tools_call(self, msg_id: Any, params: Dict[str, Any]) -> Dict[str, A response = dispatch_tool(tool_name, arguments, conn, self.config, token) response_bytes = json.dumps(response).encode("utf-8") limiter.check_quota(token.name, count_chunks(response), len(response_bytes)) - return self._build_result(msg_id, response) + content_result = {"content": [{"type": "text", "text": json.dumps(response)}]} + return self._build_result(msg_id, content_result) except AuthError as exc: return self._build_error(msg_id, -32001, str(exc)) except ScopeError as exc: diff --git a/hoard/core/memory/model_cache.py b/hoard/core/memory/model_cache.py new file mode 100644 index 0000000..a99d140 --- /dev/null +++ b/hoard/core/memory/model_cache.py @@ -0,0 +1,13 @@ +from __future__ import annotations + +from typing import Any + +_st_cache: dict[str, Any] = {} + + +def get_sentence_transformer(model_name: str): + """Return a cached SentenceTransformer instance for the given model name.""" + if model_name not in _st_cache: + from sentence_transformers import SentenceTransformer + _st_cache[model_name] = SentenceTransformer(model_name) + return _st_cache[model_name] diff --git a/hoard/core/memory/v2/store.py b/hoard/core/memory/v2/store.py index 095634e..d5adf2a 100644 --- a/hoard/core/memory/v2/store.py +++ b/hoard/core/memory/v2/store.py @@ -7,9 +7,13 @@ from typing import Any, Dict, Iterable, List, Optional from hoard.core.memory.predicates import active_memory_conditions +from hoard.core.search.bm25 import _sanitize_fts_query from hoard.core.security.auth import TokenInfo +from hoard.core.memory.model_cache import get_sentence_transformer + + class MemoryError(Exception): pass @@ -236,7 +240,7 @@ def memory_write( _insert_event(conn, memory_id=memory_id, event_type="created", actor=actor or source_agent) - _enqueue_job(conn, job_type="embed_memory", memory_id=memory_id, priority=0) + _enqueue_job(conn, job_type="embed_memory", memory_id=memory_id, priority=10) _enqueue_job(conn, job_type="detect_duplicates", memory_id=memory_id, priority=0) _enqueue_job(conn, job_type="detect_conflicts", memory_id=memory_id, priority=0) @@ -547,7 +551,7 @@ def memory_query( ORDER BY bm25(memories_fts) LIMIT ? """, - [query, *base_params, *tag_params, union_limit * 5], + [_sanitize_fts_query(query), *base_params, *tag_params, union_limit * 5], ).fetchall() fts_scores: dict[str, float] = {} @@ -564,12 +568,11 @@ def memory_query( ) if vectors_enabled: try: - from sentence_transformers import SentenceTransformer import numpy as np model_cfg = config.get("write", {}).get("embeddings", {}).get("active_model", {}) model_name = model_cfg.get("name", "sentence-transformers/all-MiniLM-L6-v2") - model = SentenceTransformer(model_name) + model = get_sentence_transformer(model_name) query_vec = model.encode([query], normalize_embeddings=True)[0] emb_rows = conn.execute( diff --git a/hoard/core/search/bm25.py b/hoard/core/search/bm25.py index 7ad6f18..e6132fb 100644 --- a/hoard/core/search/bm25.py +++ b/hoard/core/search/bm25.py @@ -3,6 +3,14 @@ from typing import Any, Dict, List +def _sanitize_fts_query(query: str) -> str: + """Escape FTS5 metacharacters by wrapping each token in double quotes.""" + tokens = query.split() + if not tokens: + return query + return " ".join('"' + token.replace('"', '""') + '"' for token in tokens) + + def search_chunks_flat( conn, query: str, @@ -27,7 +35,7 @@ def search_chunks_flat( " AND entities.tombstoned_at IS NULL" ) - params: list[Any] = [query] + params: list[Any] = [_sanitize_fts_query(query)] if source: sql += " AND entities.source = ?" params.append(source) diff --git a/hoard/core/search/hybrid.py b/hoard/core/search/hybrid.py index 36837aa..c6ef826 100644 --- a/hoard/core/search/hybrid.py +++ b/hoard/core/search/hybrid.py @@ -6,6 +6,14 @@ from hoard.core.search.bm25 import search_chunks_flat from hoard.core.search.vector import vector_search +_model_cache: dict[str, EmbeddingModel] = {} + + +def _get_model(model_name: str) -> EmbeddingModel: + if model_name not in _model_cache: + _model_cache[model_name] = EmbeddingModel(model_name) + return _model_cache[model_name] + def hybrid_search( conn, @@ -51,7 +59,7 @@ def hybrid_search( if vectors_enabled: try: - model = EmbeddingModel(model_name) + model = _get_model(model_name) query_vector = model.encode([query])[0] vector_results = vector_search( conn, diff --git a/hoard/core/worker.py b/hoard/core/worker.py index cb6005b..d614150 100644 --- a/hoard/core/worker.py +++ b/hoard/core/worker.py @@ -11,6 +11,8 @@ from hoard.core.db.connection import connect from hoard.core.db.writer import WriteCoordinator from hoard.core.memory.embeddings import encode_embedding +from hoard.core.memory.model_cache import get_sentence_transformer +from hoard.core.memory.predicates import active_memory_conditions class Worker: @@ -193,9 +195,7 @@ def _process_embed(self, memory_id: str) -> None: model_version = model_cfg.get("version", "2.0.0") dimensions = int(model_cfg.get("dimensions", 384)) - from sentence_transformers import SentenceTransformer - - model = SentenceTransformer(model_name) + model = get_sentence_transformer(model_name) vector = model.encode([content], normalize_embeddings=True)[0] blob = encode_embedding(vector, dimensions) embedded_at = datetime.utcnow().isoformat(timespec="seconds") @@ -232,9 +232,208 @@ def _store(conn): self._writer.submit(_store) def _process_duplicates(self, memory_id: str) -> None: - # Placeholder: detection not implemented yet - return + embeddings_enabled = bool( + self._config.get("write", {}).get("embeddings", {}).get("enabled", False) + ) + if not embeddings_enabled: + return + + threshold = float( + self._config.get("write", {}).get("duplicates", {}).get("similarity_threshold", 0.85) + ) + model_cfg = self._config.get("write", {}).get("embeddings", {}).get("active_model", {}) + model_name = model_cfg.get("name", "sentence-transformers/all-MiniLM-L6-v2") + dimensions = int(model_cfg.get("dimensions", 384)) + + conn = connect(self._db_path) + try: + emb_row = conn.execute( + "SELECT embedding, model_name, dimensions FROM memory_embeddings WHERE memory_id = ?", + (memory_id,), + ).fetchone() + if not emb_row: + return + if emb_row["model_name"] != model_name or emb_row["dimensions"] != dimensions: + return + + from array import array as float_array + + query_arr = float_array("f") + query_arr.frombytes(emb_row["embedding"]) + if len(query_arr) != dimensions: + return + query_vec = list(query_arr) + + now = datetime.utcnow().isoformat(timespec="seconds") + src_conditions, src_params = active_memory_conditions(now, table_alias="m") + src_conditions.append("m.id = ?") + src_params.append(memory_id) + src_where = " AND ".join(src_conditions) + mem_row = conn.execute( + f"SELECT m.scope_type, m.scope_id FROM memories m WHERE {src_where}", + src_params, + ).fetchone() + if not mem_row: + return + conditions, params = active_memory_conditions(now) + conditions.append("m.id != ?") + params.append(memory_id) + if mem_row["scope_id"] is None: + conditions.append("m.scope_type = ?") + params.append(mem_row["scope_type"]) + conditions.append("m.scope_id IS NULL") + else: + conditions.append("m.scope_type = ?") + params.append(mem_row["scope_type"]) + conditions.append("m.scope_id = ?") + params.append(mem_row["scope_id"]) + + where_clause = " AND ".join(conditions) + candidates = conn.execute( + f""" + SELECT e.memory_id, e.embedding, e.model_name, e.dimensions + FROM memory_embeddings e + JOIN memories m ON m.id = e.memory_id + WHERE {where_clause} + """, + params, + ).fetchall() + + duplicates = [] + for cand in candidates: + if cand["model_name"] != model_name or cand["dimensions"] != dimensions: + continue + cand_arr = float_array("f") + cand_arr.frombytes(cand["embedding"]) + if len(cand_arr) != dimensions: + continue + similarity = sum(a * b for a, b in zip(query_vec, cand_arr)) + if similarity >= threshold: + duplicates.append((cand["memory_id"], similarity)) + finally: + conn.close() + + if not duplicates: + return + + def _store_duplicates(conn): + for other_id, similarity in duplicates: + already = conn.execute( + """ + SELECT d.id FROM memory_duplicates d + JOIN duplicate_members dm1 ON dm1.duplicate_id = d.id + JOIN duplicate_members dm2 ON dm2.duplicate_id = d.id + WHERE dm1.memory_id = ? AND dm2.memory_id = ? + AND d.resolved_at IS NULL + """, + (memory_id, other_id), + ).fetchone() + if already: + continue + dup_id = str(uuid.uuid4()) + now_ts = datetime.utcnow().isoformat(timespec="seconds") + conn.execute( + "INSERT INTO memory_duplicates (id, detected_at, similarity) VALUES (?, ?, ?)", + (dup_id, now_ts, similarity), + ) + conn.execute( + "INSERT INTO duplicate_members (duplicate_id, memory_id, is_canonical) VALUES (?, ?, 1)", + (dup_id, other_id), + ) + conn.execute( + "INSERT INTO duplicate_members (duplicate_id, memory_id, is_canonical) VALUES (?, ?, 0)", + (dup_id, memory_id), + ) + + self._writer.submit(_store_duplicates) def _process_conflicts(self, memory_id: str) -> None: - # Placeholder: detection not implemented yet - return + conn = connect(self._db_path) + try: + now = datetime.utcnow().isoformat(timespec="seconds") + src_conditions, src_params = active_memory_conditions(now, table_alias="m") + src_conditions.append("m.id = ?") + src_params.append(memory_id) + src_where = " AND ".join(src_conditions) + mem_row = conn.execute( + f"SELECT m.slot, m.scope_type, m.scope_id FROM memories m WHERE {src_where}", + src_params, + ).fetchone() + if not mem_row or not mem_row["slot"]: + return + + slot = mem_row["slot"] + scope_type = mem_row["scope_type"] + scope_id = mem_row["scope_id"] + + now = datetime.utcnow().isoformat(timespec="seconds") + conditions, params = active_memory_conditions(now) + conditions.append("m.id != ?") + params.append(memory_id) + conditions.append("m.slot = ?") + params.append(slot) + conditions.append("m.scope_type = ?") + params.append(scope_type) + if scope_id is None: + conditions.append("m.scope_id IS NULL") + else: + conditions.append("m.scope_id = ?") + params.append(scope_id) + + where_clause = " AND ".join(conditions) + others = conn.execute( + f"SELECT m.id FROM memories m WHERE {where_clause}", + params, + ).fetchall() + + if not others: + return + + other_ids = [row["id"] for row in others] + + existing_conflict_id = None + for other_id in other_ids: + row = conn.execute( + """ + SELECT cm.conflict_id FROM conflict_members cm + JOIN memory_conflicts c ON c.id = cm.conflict_id + WHERE cm.memory_id = ? AND c.resolved_at IS NULL + AND c.slot = ? AND c.scope_type = ? + """, + (other_id, slot, scope_type), + ).fetchone() + if row: + existing_conflict_id = row["conflict_id"] + break + finally: + conn.close() + + def _store_conflict(conn): + now_ts = datetime.utcnow().isoformat(timespec="seconds") + if existing_conflict_id: + already = conn.execute( + "SELECT 1 FROM conflict_members WHERE conflict_id = ? AND memory_id = ?", + (existing_conflict_id, memory_id), + ).fetchone() + if not already: + conn.execute( + "INSERT INTO conflict_members (conflict_id, memory_id, nli_label, added_at) VALUES (?, ?, NULL, ?)", + (existing_conflict_id, memory_id, now_ts), + ) + else: + conflict_id = str(uuid.uuid4()) + conn.execute( + "INSERT INTO memory_conflicts (id, slot, scope_type, scope_id, detected_at) VALUES (?, ?, ?, ?, ?)", + (conflict_id, slot, scope_type, scope_id, now_ts), + ) + for mid in other_ids: + conn.execute( + "INSERT INTO conflict_members (conflict_id, memory_id, nli_label, added_at) VALUES (?, ?, NULL, ?)", + (conflict_id, mid, now_ts), + ) + conn.execute( + "INSERT INTO conflict_members (conflict_id, memory_id, nli_label, added_at) VALUES (?, ?, NULL, ?)", + (conflict_id, memory_id, now_ts), + ) + + self._writer.submit(_store_conflict) diff --git a/tests/test_cli_http_parity.py b/tests/test_cli_http_parity.py index e335338..4a82cca 100644 --- a/tests/test_cli_http_parity.py +++ b/tests/test_cli_http_parity.py @@ -58,4 +58,5 @@ def test_cli_write_http_read(tmp_path: Path, mcp_server) -> None: {"name": "memory_get", "arguments": {"key": "shared_key"}}, ) assert status == 200 - assert resp["result"]["memory"]["content"] == "from cli" + content = json.loads(resp["result"]["content"][0]["text"]) + assert content["memory"]["content"] == "from cli" diff --git a/tests/test_http_mcp_concurrency.py b/tests/test_http_mcp_concurrency.py index 2d708b4..75a99e6 100644 --- a/tests/test_http_mcp_concurrency.py +++ b/tests/test_http_mcp_concurrency.py @@ -72,7 +72,8 @@ def worker() -> None: "tools/call", {"name": "search", "arguments": {"query": "Hoard", "limit": 1}}, ) - assert resp["result"]["results"] + content = json.loads(resp["result"]["content"][0]["text"]) + assert content["results"] except Exception as exc: # pragma: no cover - diagnostic with lock: errors.append(str(exc)) diff --git a/tests/test_http_mcp_integration.py b/tests/test_http_mcp_integration.py index 797612c..ca6c903 100644 --- a/tests/test_http_mcp_integration.py +++ b/tests/test_http_mcp_integration.py @@ -71,7 +71,8 @@ def test_http_mcp_tools_list_and_search(tmp_path: Path, mcp_server) -> None: "tools/call", {"name": "search", "arguments": {"query": "Hoard", "limit": 1}}, ) - assert search["result"]["results"] + content = json.loads(search["result"]["content"][0]["text"]) + assert content["results"] finally: pass @@ -143,7 +144,8 @@ def test_http_mcp_inbox_put_and_sync(tmp_path: Path, mcp_server) -> None: "tools/call", {"name": "inbox_put", "arguments": {"content": "Inbox HTTP content"}}, ) - assert "path" in inbox_resp["result"] + inbox_content = json.loads(inbox_resp["result"]["content"][0]["text"]) + assert "path" in inbox_content _, search = _call_mcp( url, @@ -151,7 +153,8 @@ def test_http_mcp_inbox_put_and_sync(tmp_path: Path, mcp_server) -> None: "tools/call", {"name": "search", "arguments": {"query": "Inbox HTTP", "limit": 1}}, ) - assert search["result"]["results"] + search_content = json.loads(search["result"]["content"][0]["text"]) + assert search_content["results"] finally: pass finally: diff --git a/tests/test_orchestrator_mcp.py b/tests/test_orchestrator_mcp.py index 08d5d6c..575b3a1 100644 --- a/tests/test_orchestrator_mcp.py +++ b/tests/test_orchestrator_mcp.py @@ -54,7 +54,8 @@ def test_orchestrator_mcp_task_flow(tmp_path: Path, mcp_server, monkeypatch) -> }, }, ) - agent_token = register_resp["result"]["token"] + register_content = json.loads(register_resp["result"]["content"][0]["text"]) + agent_token = register_content["token"] _, task_resp = _call_mcp( url, @@ -62,17 +63,21 @@ def test_orchestrator_mcp_task_flow(tmp_path: Path, mcp_server, monkeypatch) -> "tools/call", {"name": "task.create", "arguments": {"name": "Demo task", "requires_capability": "cap.basic"}}, ) - task_id = task_resp["result"]["id"] + task_content = json.loads(task_resp["result"]["content"][0]["text"]) + task_id = task_content["id"] _, poll_resp = _call_mcp(url, agent_token, "tools/call", {"name": "task.poll", "arguments": {"limit": 5}}) - tasks = poll_resp["result"]["tasks"] + poll_content = json.loads(poll_resp["result"]["content"][0]["text"]) + tasks = poll_content["tasks"] assert tasks and tasks[0]["id"] == task_id _, claim_resp = _call_mcp(url, agent_token, "tools/call", {"name": "task.claim", "arguments": {"task_id": task_id}}) - assert claim_resp["result"]["task"]["status"] == "claimed" + claim_content = json.loads(claim_resp["result"]["content"][0]["text"]) + assert claim_content["task"]["status"] == "claimed" _, start_resp = _call_mcp(url, agent_token, "tools/call", {"name": "task.start", "arguments": {"task_id": task_id}}) - assert start_resp["result"]["success"] is True + start_content = json.loads(start_resp["result"]["content"][0]["text"]) + assert start_content["success"] is True _, complete_resp = _call_mcp( url, @@ -80,7 +85,8 @@ def test_orchestrator_mcp_task_flow(tmp_path: Path, mcp_server, monkeypatch) -> "tools/call", {"name": "task.complete", "arguments": {"task_id": task_id, "output_summary": "done"}}, ) - assert complete_resp["result"]["success"] is True + complete_content = json.loads(complete_resp["result"]["content"][0]["text"]) + assert complete_content["success"] is True _, artifact_resp = _call_mcp( url, @@ -96,7 +102,8 @@ def test_orchestrator_mcp_task_flow(tmp_path: Path, mcp_server, monkeypatch) -> }, }, ) - artifact_id = artifact_resp["result"]["artifact"]["artifact_id"] + artifact_content = json.loads(artifact_resp["result"]["content"][0]["text"]) + artifact_id = artifact_content["artifact"]["artifact_id"] _, artifact_get_resp = _call_mcp( url, @@ -104,10 +111,12 @@ def test_orchestrator_mcp_task_flow(tmp_path: Path, mcp_server, monkeypatch) -> "tools/call", {"name": "artifact.get", "arguments": {"artifact_id": artifact_id, "include_content": True}}, ) - assert artifact_get_resp["result"]["artifact"]["content"] == "hello" + artifact_get_content = json.loads(artifact_get_resp["result"]["content"][0]["text"]) + assert artifact_get_content["artifact"]["content"] == "hello" _, event_resp = _call_mcp(url, agent_token, "tools/call", {"name": "event.poll", "arguments": {"limit": 50}}) - event_types = [event["event_type"] for event in event_resp["result"]["events"]] + event_content = json.loads(event_resp["result"]["content"][0]["text"]) + event_types = [event["event_type"] for event in event_content["events"]] assert "task.completed" in event_types diff --git a/tests/test_stdio_mcp.py b/tests/test_stdio_mcp.py index 8b7efef..a0e8cff 100644 --- a/tests/test_stdio_mcp.py +++ b/tests/test_stdio_mcp.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json from pathlib import Path from hoard.connectors.local_files.connector import LocalFilesConnector @@ -80,11 +81,12 @@ def test_stdio_tools_list_and_search(tmp_path: Path) -> None: } ) assert search_resp - results = search_resp["result"]["results"] + search_content = json.loads(search_resp["result"]["content"][0]["text"]) + results = search_content["results"] assert results - assert search_resp["result"]["next_cursor"] is not None + assert search_content["next_cursor"] is not None - cursor = search_resp["result"]["next_cursor"] + cursor = search_content["next_cursor"] next_resp = server._handle_single_message( { "jsonrpc": "2.0", @@ -97,7 +99,8 @@ def test_stdio_tools_list_and_search(tmp_path: Path) -> None: } ) assert next_resp - assert next_resp["result"]["results"] + next_content = json.loads(next_resp["result"]["content"][0]["text"]) + assert next_content["results"] write_resp = server._handle_single_message( { diff --git a/tests/test_worker.py b/tests/test_worker.py index 24120b5..bb13cd0 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -1,5 +1,7 @@ from __future__ import annotations +import uuid +from array import array from datetime import datetime, timedelta from pathlib import Path @@ -20,6 +22,63 @@ def _config() -> dict: } +def _detection_config(embeddings_enabled: bool = True) -> dict: + return { + "write": { + "worker": { + "poll_interval_ms": 10, + "job_timeout_seconds": 1, + "lease_duration_seconds": 60, + }, + "embeddings": { + "enabled": embeddings_enabled, + "active_model": { + "name": "test-model", + "version": "1.0", + "dimensions": 4, + }, + }, + "duplicates": { + "similarity_threshold": 0.85, + }, + } + } + + +def _make_embedding(values: list[float]) -> bytes: + """Encode a float vector as little-endian float32 bytes.""" + arr = array("f", values) + return arr.tobytes() + + +def _insert_memory(conn, memory_id: str, content: str, slot: str | None = None, + scope_type: str = "user", scope_id: str | None = None) -> None: + now = datetime.utcnow().isoformat(timespec="seconds") + conn.execute( + """ + INSERT INTO memories (id, content, memory_type, slot, scope_type, scope_id, + source_agent, created_at, sensitivity) + VALUES (?, ?, 'fact', ?, ?, ?, 'test', ?, 'normal') + """, + (memory_id, content, slot, scope_type, scope_id, now), + ) + conn.execute("INSERT INTO memory_counters (memory_id) VALUES (?)", (memory_id,)) + conn.commit() + + +def _insert_embedding(conn, memory_id: str, embedding: bytes, model_name: str = "test-model", + dimensions: int = 4) -> None: + now = datetime.utcnow().isoformat(timespec="seconds") + conn.execute( + """ + INSERT INTO memory_embeddings (memory_id, embedding, model_name, model_version, dimensions, embedded_at) + VALUES (?, ?, ?, '1.0', ?, ?) + """, + (memory_id, embedding, model_name, dimensions, now), + ) + conn.commit() + + def test_worker_lease(tmp_path: Path) -> None: db_path = tmp_path / "hoard.db" conn = connect(db_path) @@ -61,3 +120,462 @@ def test_stuck_recovery(tmp_path: Path) -> None: conn.close() writer.stop() + + +def test_duplicate_detection(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid1 = str(uuid.uuid4()) + mid2 = str(uuid.uuid4()) + vec = [0.5, 0.5, 0.5, 0.5] # identical vectors → similarity 1.0 + blob = _make_embedding(vec) + + _insert_memory(conn, mid1, "duplicate content A") + _insert_memory(conn, mid2, "duplicate content A") + _insert_embedding(conn, mid1, blob) + _insert_embedding(conn, mid2, blob) + + worker._process_duplicates(mid2) + + dup_rows = conn.execute("SELECT * FROM memory_duplicates").fetchall() + assert len(dup_rows) == 1 + assert dup_rows[0]["similarity"] >= 0.99 + + members = conn.execute( + "SELECT memory_id, is_canonical FROM duplicate_members ORDER BY is_canonical DESC" + ).fetchall() + assert len(members) == 2 + member_ids = {m["memory_id"] for m in members} + assert mid1 in member_ids + assert mid2 in member_ids + + conn.close() + writer.stop() + + +def test_duplicate_detection_idempotent(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid1 = str(uuid.uuid4()) + mid2 = str(uuid.uuid4()) + blob = _make_embedding([0.5, 0.5, 0.5, 0.5]) + + _insert_memory(conn, mid1, "same content") + _insert_memory(conn, mid2, "same content") + _insert_embedding(conn, mid1, blob) + _insert_embedding(conn, mid2, blob) + + worker._process_duplicates(mid2) + worker._process_duplicates(mid2) + + dup_rows = conn.execute("SELECT * FROM memory_duplicates").fetchall() + assert len(dup_rows) == 1 + + conn.close() + writer.stop() + + +def test_duplicate_detection_skips_without_embedding(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid = str(uuid.uuid4()) + _insert_memory(conn, mid, "no embedding content") + + # Should complete without error (no-op) + worker._process_duplicates(mid) + + dup_rows = conn.execute("SELECT * FROM memory_duplicates").fetchall() + assert len(dup_rows) == 0 + + conn.close() + writer.stop() + + +def test_duplicate_detection_skips_mismatched_model(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid1 = str(uuid.uuid4()) + mid2 = str(uuid.uuid4()) + blob = _make_embedding([0.5, 0.5, 0.5, 0.5]) + + _insert_memory(conn, mid1, "content A") + _insert_memory(conn, mid2, "content B") + _insert_embedding(conn, mid1, blob, model_name="other-model") + _insert_embedding(conn, mid2, blob) + + worker._process_duplicates(mid2) + + dup_rows = conn.execute("SELECT * FROM memory_duplicates").fetchall() + assert len(dup_rows) == 0 + + conn.close() + writer.stop() + + +def test_duplicate_detection_null_scope_id(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid1 = str(uuid.uuid4()) + mid2 = str(uuid.uuid4()) + mid3 = str(uuid.uuid4()) + blob = _make_embedding([0.5, 0.5, 0.5, 0.5]) + + # mid1 and mid2: user scope (NULL scope_id) + _insert_memory(conn, mid1, "user content", scope_type="user", scope_id=None) + _insert_memory(conn, mid2, "user content", scope_type="user", scope_id=None) + # mid3: project scope (different scope) — should NOT match + _insert_memory(conn, mid3, "project content", scope_type="project", scope_id="proj1") + _insert_embedding(conn, mid1, blob) + _insert_embedding(conn, mid2, blob) + _insert_embedding(conn, mid3, blob) + + worker._process_duplicates(mid2) + + members = conn.execute("SELECT memory_id FROM duplicate_members").fetchall() + member_ids = {m["memory_id"] for m in members} + assert mid1 in member_ids + assert mid2 in member_ids + assert mid3 not in member_ids + + conn.close() + writer.stop() + + +def test_conflict_detection(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid1 = str(uuid.uuid4()) + mid2 = str(uuid.uuid4()) + + _insert_memory(conn, mid1, "preferred dark mode", slot="pref:theme", scope_type="user") + _insert_memory(conn, mid2, "preferred light mode", slot="pref:theme", scope_type="user") + + worker._process_conflicts(mid2) + + conflict_rows = conn.execute("SELECT * FROM memory_conflicts").fetchall() + assert len(conflict_rows) == 1 + assert conflict_rows[0]["slot"] == "pref:theme" + assert conflict_rows[0]["scope_type"] == "user" + assert conflict_rows[0]["scope_id"] is None + + members = conn.execute("SELECT memory_id FROM conflict_members").fetchall() + member_ids = {m["memory_id"] for m in members} + assert mid1 in member_ids + assert mid2 in member_ids + + conn.close() + writer.stop() + + +def test_conflict_detection_idempotent(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid1 = str(uuid.uuid4()) + mid2 = str(uuid.uuid4()) + + _insert_memory(conn, mid1, "value A", slot="fact:test.key", scope_type="user") + _insert_memory(conn, mid2, "value B", slot="fact:test.key", scope_type="user") + + worker._process_conflicts(mid2) + worker._process_conflicts(mid2) + + conflict_rows = conn.execute("SELECT * FROM memory_conflicts").fetchall() + assert len(conflict_rows) == 1 + + members = conn.execute("SELECT memory_id FROM conflict_members").fetchall() + assert len(members) == 2 + + conn.close() + writer.stop() + + +def test_conflict_appends_to_existing(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid1 = str(uuid.uuid4()) + mid2 = str(uuid.uuid4()) + mid3 = str(uuid.uuid4()) + + _insert_memory(conn, mid1, "value A", slot="fact:test.key", scope_type="user") + _insert_memory(conn, mid2, "value B", slot="fact:test.key", scope_type="user") + + worker._process_conflicts(mid2) + + conflict_rows = conn.execute("SELECT * FROM memory_conflicts").fetchall() + assert len(conflict_rows) == 1 + + # Third memory with same slot: should append to existing conflict, not create new + _insert_memory(conn, mid3, "value C", slot="fact:test.key", scope_type="user") + worker._process_conflicts(mid3) + + conflict_rows = conn.execute("SELECT * FROM memory_conflicts").fetchall() + assert len(conflict_rows) == 1 + + members = conn.execute("SELECT memory_id FROM conflict_members").fetchall() + member_ids = {m["memory_id"] for m in members} + assert mid1 in member_ids + assert mid2 in member_ids + assert mid3 in member_ids + assert len(members) == 3 + + conn.close() + writer.stop() + + +def test_conflict_detection_skips_without_slot(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid = str(uuid.uuid4()) + _insert_memory(conn, mid, "no slot content", slot=None, scope_type="user") + + worker._process_conflicts(mid) + + conflict_rows = conn.execute("SELECT * FROM memory_conflicts").fetchall() + assert len(conflict_rows) == 0 + + conn.close() + writer.stop() + + +def test_conflict_detection_null_scope_id(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid1 = str(uuid.uuid4()) + mid2 = str(uuid.uuid4()) + mid3 = str(uuid.uuid4()) + + # mid1 and mid2: user scope (NULL scope_id), same slot + _insert_memory(conn, mid1, "user A", slot="fact:test.key", scope_type="user", scope_id=None) + _insert_memory(conn, mid2, "user B", slot="fact:test.key", scope_type="user", scope_id=None) + # mid3: project scope with same slot — should NOT conflict with mid1/mid2 + _insert_memory(conn, mid3, "project A", slot="fact:test.key", scope_type="project", scope_id="proj1") + + worker._process_conflicts(mid2) + + members = conn.execute("SELECT memory_id FROM conflict_members").fetchall() + member_ids = {m["memory_id"] for m in members} + assert mid1 in member_ids + assert mid2 in member_ids + assert mid3 not in member_ids + + conn.close() + writer.stop() + + +def test_duplicate_detection_skips_retracted_source(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid1 = str(uuid.uuid4()) + mid2 = str(uuid.uuid4()) + blob = _make_embedding([0.5, 0.5, 0.5, 0.5]) + + _insert_memory(conn, mid1, "content A") + _insert_memory(conn, mid2, "content A") + _insert_embedding(conn, mid1, blob) + _insert_embedding(conn, mid2, blob) + + # Retract the source memory + now = datetime.utcnow().isoformat(timespec="seconds") + conn.execute("UPDATE memories SET retracted_at = ? WHERE id = ?", (now, mid2)) + conn.commit() + + worker._process_duplicates(mid2) + + dup_rows = conn.execute("SELECT * FROM memory_duplicates").fetchall() + assert len(dup_rows) == 0 + + conn.close() + writer.stop() + + +def test_duplicate_detection_skips_superseded_source(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid1 = str(uuid.uuid4()) + mid2 = str(uuid.uuid4()) + blob = _make_embedding([0.5, 0.5, 0.5, 0.5]) + + _insert_memory(conn, mid1, "content A") + _insert_memory(conn, mid2, "content A") + _insert_embedding(conn, mid1, blob) + _insert_embedding(conn, mid2, blob) + + now = datetime.utcnow().isoformat(timespec="seconds") + conn.execute("UPDATE memories SET superseded_by = ?, superseded_at = ? WHERE id = ?", (mid1, now, mid2)) + conn.commit() + + worker._process_duplicates(mid2) + + assert conn.execute("SELECT count(*) FROM memory_duplicates").fetchone()[0] == 0 + + conn.close() + writer.stop() + + +def test_duplicate_detection_skips_expired_source(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid1 = str(uuid.uuid4()) + mid2 = str(uuid.uuid4()) + blob = _make_embedding([0.5, 0.5, 0.5, 0.5]) + + _insert_memory(conn, mid1, "content A") + _insert_memory(conn, mid2, "content A") + _insert_embedding(conn, mid1, blob) + _insert_embedding(conn, mid2, blob) + + past = (datetime.utcnow() - timedelta(hours=1)).isoformat(timespec="seconds") + conn.execute("UPDATE memories SET expires_at = ? WHERE id = ?", (past, mid2)) + conn.commit() + + worker._process_duplicates(mid2) + + assert conn.execute("SELECT count(*) FROM memory_duplicates").fetchone()[0] == 0 + + conn.close() + writer.stop() + + +def test_conflict_detection_skips_retracted_source(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid1 = str(uuid.uuid4()) + mid2 = str(uuid.uuid4()) + + _insert_memory(conn, mid1, "value A", slot="fact:test.key", scope_type="user") + _insert_memory(conn, mid2, "value B", slot="fact:test.key", scope_type="user") + + # Retract the source memory + now = datetime.utcnow().isoformat(timespec="seconds") + conn.execute("UPDATE memories SET retracted_at = ? WHERE id = ?", (now, mid2)) + conn.commit() + + worker._process_conflicts(mid2) + + conflict_rows = conn.execute("SELECT * FROM memory_conflicts").fetchall() + assert len(conflict_rows) == 0 + + conn.close() + writer.stop() + + +def test_conflict_detection_skips_superseded_source(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid1 = str(uuid.uuid4()) + mid2 = str(uuid.uuid4()) + + _insert_memory(conn, mid1, "value A", slot="fact:test.key", scope_type="user") + _insert_memory(conn, mid2, "value B", slot="fact:test.key", scope_type="user") + + now = datetime.utcnow().isoformat(timespec="seconds") + conn.execute("UPDATE memories SET superseded_by = ?, superseded_at = ? WHERE id = ?", (mid1, now, mid2)) + conn.commit() + + worker._process_conflicts(mid2) + + assert conn.execute("SELECT count(*) FROM memory_conflicts").fetchone()[0] == 0 + + conn.close() + writer.stop() + + +def test_conflict_detection_skips_expired_source(tmp_path: Path) -> None: + db_path = tmp_path / "hoard.db" + conn = connect(db_path) + initialize_db(conn) + + writer = WriteCoordinator(db_path=db_path) + worker = Worker(db_path=db_path, config=_detection_config(), writer=writer) + + mid1 = str(uuid.uuid4()) + mid2 = str(uuid.uuid4()) + + _insert_memory(conn, mid1, "value A", slot="fact:test.key", scope_type="user") + _insert_memory(conn, mid2, "value B", slot="fact:test.key", scope_type="user") + + past = (datetime.utcnow() - timedelta(hours=1)).isoformat(timespec="seconds") + conn.execute("UPDATE memories SET expires_at = ? WHERE id = ?", (past, mid2)) + conn.commit() + + worker._process_conflicts(mid2) + + assert conn.execute("SELECT count(*) FROM memory_conflicts").fetchone()[0] == 0 + + conn.close() + writer.stop()