From 7c29a22cf3d970ba588daa4d249401f2d31b95c8 Mon Sep 17 00:00:00 2001 From: r266-tech Date: Sun, 2 Aug 2026 10:25:21 +0800 Subject: [PATCH 1/2] fix(search): bound temporal query analysis --- .../hindsight_api/engine/query_analyzer.py | 11 + .../hindsight_api/engine/search/retrieval.py | 78 ++++- .../tests/test_query_analyzer.py | 32 +++ .../tests/test_temporal_recall_selection.py | 272 ++++++++++++++++++ 4 files changed, 392 insertions(+), 1 deletion(-) diff --git a/hindsight-api-slim/hindsight_api/engine/query_analyzer.py b/hindsight-api-slim/hindsight_api/engine/query_analyzer.py index d89fd0573a..673c7ddc53 100644 --- a/hindsight-api-slim/hindsight_api/engine/query_analyzer.py +++ b/hindsight-api-slim/hindsight_api/engine/query_analyzer.py @@ -20,6 +20,14 @@ logger = logging.getLogger(__name__) +# Temporal parsing is synchronous regex-heavy work. Recall normally receives +# query-sized inputs, but internal callers such as consolidation can pass a +# full memory as the query. For the repetition reported in issue #3134, 512 +# characters parsed in under a second on reference hardware while 4096 took +# roughly seven seconds. Temporal filtering is optional, so longer inputs fail +# open to non-temporal retrieval; accepted inputs are also parsed off-loop. +_MAX_TEMPORAL_ANALYSIS_CHARS = 512 + # dateparser.search_dates over-matches: short common words that happen to be # weekday/month abbreviations in *some* language ("we"/"me"/"did" -> a weekday, # "do" -> Sunday) come back as bogus dates. When such a false positive appears @@ -221,6 +229,9 @@ def analyze(self, query: str, reference_date: datetime | None = None) -> QueryAn Returns: QueryAnalysis with temporal_constraint if found """ + if len(query) > _MAX_TEMPORAL_ANALYSIS_CHARS: + return QueryAnalysis(temporal_constraint=None) + if reference_date is None: reference_date = datetime.now() diff --git a/hindsight-api-slim/hindsight_api/engine/search/retrieval.py b/hindsight-api-slim/hindsight_api/engine/search/retrieval.py index 59935ad4aa..bfde82ec04 100644 --- a/hindsight-api-slim/hindsight_api/engine/search/retrieval.py +++ b/hindsight-api-slim/hindsight_api/engine/search/retrieval.py @@ -11,8 +11,11 @@ import asyncio import logging import re +from concurrent.futures import Future, ThreadPoolExecutor from dataclasses import dataclass, field from datetime import UTC, datetime +from functools import partial +from threading import BoundedSemaphore, Event from typing import TYPE_CHECKING, Any, Optional from ...config import DEFAULT_BM25_MAX_QUERY_TERMS, DEFAULT_TEMPORAL_SEMANTIC_MIN_SIMILARITY, get_config @@ -29,6 +32,26 @@ logger = logging.getLogger(__name__) +_TEMPORAL_ANALYSIS_WORKERS = 2 +_TEMPORAL_ANALYSIS_CAPACITY = 4 +_TEMPORAL_ANALYSIS_TIMEOUT_SECONDS = 1.0 +_TEMPORAL_ANALYSIS_EXECUTOR = ThreadPoolExecutor( + max_workers=_TEMPORAL_ANALYSIS_WORKERS, + thread_name_prefix="hindsight-temporal", +) +_TEMPORAL_ANALYSIS_SLOTS = BoundedSemaphore(_TEMPORAL_ANALYSIS_CAPACITY) +_TEMPORAL_ANALYSIS_DISABLED = Event() + + +def _release_temporal_analysis_slot(_future: Future[Any], *, slots: BoundedSemaphore) -> None: + """Release capacity only when parser work actually stops. + + Cancelling the awaiting recall cannot stop a running thread. Releasing from + the concurrent future's callback prevents cancelled work from admitting a + replacement and silently creating an unbounded executor queue. + """ + slots.release() + def tokenize_query(query_text: str) -> list[str]: """Normalize query text and split into BM25 tokens. @@ -881,9 +904,62 @@ async def retrieve_all_fact_types_parallel( # Step 1: Extract temporal constraint first (CPU work, no DB) # Do this before DB queries so we know if we need temporal retrieval temporal_extraction_start = time.time() + from ..query_analyzer import _MAX_TEMPORAL_ANALYSIS_CHARS, DateparserQueryAnalyzer from .temporal_extraction import extract_temporal_constraint - temporal_constraint = extract_temporal_constraint(query_text, reference_date=question_date, analyzer=query_analyzer) + # dateparser performs synchronous regex-heavy work. Keep it in an isolated + # executor with a two-item bounded queue. Normal short bursts can wait for + # a worker, while overload or slow parsing fails open after a short deadline + # instead of starving unrelated executor users or the event loop. + analysis_slots = _TEMPORAL_ANALYSIS_SLOTS + uses_default_dateparser = query_analyzer is None or type(query_analyzer) is DateparserQueryAnalyzer + if not uses_default_dateparser: + # Custom analyzers predate the bounded dateparser path. Preserve their + # synchronous calling and threading semantics because the interface + # does not require implementations to be thread-safe. + temporal_constraint = extract_temporal_constraint( + query_text, + reference_date=question_date, + analyzer=query_analyzer, + ) + elif ( + len(query_text) > _MAX_TEMPORAL_ANALYSIS_CHARS + or _TEMPORAL_ANALYSIS_DISABLED.is_set() + or not analysis_slots.acquire(blocking=False) + ): + temporal_constraint = None + else: + try: + temporal_future = _TEMPORAL_ANALYSIS_EXECUTOR.submit( + extract_temporal_constraint, + query_text, + reference_date=question_date, + analyzer=query_analyzer, + ) + except RuntimeError: + # submit() may have enqueued before worker startup failed. Keep the + # slot reserved and trip a circuit breaker rather than admitting + # replacement work into an ambiguous queue. + _TEMPORAL_ANALYSIS_DISABLED.set() + logger.error("Temporal analysis executor unavailable; disabling temporal analysis") + temporal_constraint = None + else: + temporal_future.add_done_callback(partial(_release_temporal_analysis_slot, slots=analysis_slots)) + try: + temporal_constraint = await asyncio.wait_for( + asyncio.shield(asyncio.wrap_future(temporal_future)), + timeout=_TEMPORAL_ANALYSIS_TIMEOUT_SECONDS, + ) + except TimeoutError: + # Shielding keeps queued or running work tied to its capacity + # slot until the concurrent-future callback sees it really end. + temporal_constraint = None + except Exception as exc: + logger.warning( + "Temporal analysis failed with %s; continuing without a temporal filter", + type(exc).__name__, + ) + temporal_constraint = None temporal_extraction_time = time.time() - temporal_extraction_start timings["temporal_extraction"] = temporal_extraction_time diff --git a/hindsight-api-slim/tests/test_query_analyzer.py b/hindsight-api-slim/tests/test_query_analyzer.py index e401687260..961993e789 100644 --- a/hindsight-api-slim/tests/test_query_analyzer.py +++ b/hindsight-api-slim/tests/test_query_analyzer.py @@ -91,6 +91,38 @@ def test_query_analyzer_no_temporal(query_analyzer): assert analysis.temporal_constraint is None, "Should not extract temporal constraint" +def test_query_analyzer_skips_oversized_queries_before_parsing(query_analyzer, monkeypatch): + from hindsight_api.engine import query_analyzer as query_analyzer_module + + def fail_extract_period(query, reference_date): + pytest.fail("oversized query reached temporal period parsing") + + monkeypatch.setattr(query_analyzer_module, "extract_period", fail_extract_period) + monkeypatch.setattr( + query_analyzer, + "load", + lambda: pytest.fail("oversized query reached dateparser"), + ) + + query = "x" * (query_analyzer_module._MAX_TEMPORAL_ANALYSIS_CHARS + 1) + analysis = query_analyzer.analyze(query, datetime(2025, 1, 15, 12, 0, 0)) + + assert analysis.temporal_constraint is None + + +@pytest.mark.timeout(5) +def test_query_analyzer_pathological_query_at_limit_is_bounded(query_analyzer): + from hindsight_api.engine import query_analyzer as query_analyzer_module + + repeated_word = "mandatory " + repetitions = query_analyzer_module._MAX_TEMPORAL_ANALYSIS_CHARS // len(repeated_word) + 1 + query = (repeated_word * repetitions)[: query_analyzer_module._MAX_TEMPORAL_ANALYSIS_CHARS] + + analysis = query_analyzer.analyze(query, datetime(2025, 1, 15, 12, 0, 0)) + + assert analysis.temporal_constraint is None + + def test_query_analyzer_activities_june_2024(query_analyzer): reference_date = datetime(2025, 1, 15, 12, 0, 0) diff --git a/hindsight-api-slim/tests/test_temporal_recall_selection.py b/hindsight-api-slim/tests/test_temporal_recall_selection.py index 50678b1d30..efd73601cd 100644 --- a/hindsight-api-slim/tests/test_temporal_recall_selection.py +++ b/hindsight-api-slim/tests/test_temporal_recall_selection.py @@ -10,12 +10,15 @@ These are pure mechanics (no LLM), so they assert directly. """ +import asyncio +import threading from contextlib import asynccontextmanager from datetime import UTC, datetime, timedelta from types import SimpleNamespace import pytest +import hindsight_api.engine.query_analyzer as query_analyzer_module import hindsight_api.engine.search.retrieval as retrieval_module from hindsight_api.engine.search.retrieval import _select_with_temporal_coverage, retrieve_temporal_combined from hindsight_api.engine.task_backend import fq_table @@ -78,6 +81,275 @@ def test_coverage_degenerate_dates_fall_back_to_similarity(): assert [r["similarity"] for r in selected] == [0.95, 0.9] +@pytest.mark.asyncio +async def test_temporal_analysis_is_bounded_and_cancellation_safe(monkeypatch): + """Saturation must fail open without releasing cancelled parser work.""" + analysis_started = threading.Event() + queued_analysis_started = threading.Event() + release_analysis = threading.Event() + analysis_calls = 0 + + def blocking_extract(*_args, **_kwargs): + nonlocal analysis_calls + analysis_calls += 1 + if analysis_calls == 1: + analysis_started.set() + elif analysis_calls == 2: + queued_analysis_started.set() + release_analysis.wait(timeout=2) + return None + + @asynccontextmanager + async def fake_acquire_with_retry(pool): + yield object() + + async def fake_semantic_bm25_combined(*args, **kwargs): + return {"world": retrieval_module.SemanticBm25Result(semantic=[], bm25=[], graph_seeds=None)} + + async def fake_temporal_combined(*args, **kwargs): + return {"world": []} + + class FakeGraphRetriever: + async def retrieve(self, **kwargs): + return [], None + + monkeypatch.setattr(retrieval_module, "acquire_with_retry", fake_acquire_with_retry) + monkeypatch.setattr(retrieval_module, "retrieve_semantic_bm25_combined", fake_semantic_bm25_combined) + monkeypatch.setattr(retrieval_module, "retrieve_temporal_combined", fake_temporal_combined) + monkeypatch.setattr( + retrieval_module, + "get_config", + lambda: SimpleNamespace( + graph_seed_min_similarity=0.3, + temporal_semantic_min_similarity=0.24, + ), + ) + monkeypatch.setattr( + "hindsight_api.engine.search.temporal_extraction.extract_temporal_constraint", + blocking_extract, + ) + + class CountingExecutor: + def __init__(self, executor): + self.executor = executor + self.calls = 0 + + def submit(self, *args, **kwargs): + self.calls += 1 + return self.executor.submit(*args, **kwargs) + + executor = retrieval_module.ThreadPoolExecutor(max_workers=1) + counting_executor = CountingExecutor(executor) + analysis_disabled = threading.Event() + monkeypatch.setattr(retrieval_module, "_TEMPORAL_ANALYSIS_EXECUTOR", counting_executor) + monkeypatch.setattr(retrieval_module, "_TEMPORAL_ANALYSIS_SLOTS", threading.BoundedSemaphore(2)) + monkeypatch.setattr(retrieval_module, "_TEMPORAL_ANALYSIS_DISABLED", analysis_disabled) + monkeypatch.setattr(retrieval_module, "_TEMPORAL_ANALYSIS_TIMEOUT_SECONDS", 1.0) + + oversized_result = await asyncio.wait_for( + retrieval_module.retrieve_all_fact_types_parallel( + object(), + query_text="x" * (query_analyzer_module._MAX_TEMPORAL_ANALYSIS_CHARS + 1), + query_embedding_str=_QUERY, + bank_id="test_temporal_oversized", + fact_types=["world"], + thinking_budget=10, + graph_retriever=FakeGraphRetriever(), + ), + timeout=0.5, + ) + assert oversized_result.results_by_fact_type["world"].temporal_constraint is None + assert analysis_calls == 0 + assert counting_executor.calls == 0 + + custom_calls = [] + + def custom_extract(query, *_args, **_kwargs): + custom_calls.append((query, threading.get_ident())) + return None + + monkeypatch.setattr( + "hindsight_api.engine.search.temporal_extraction.extract_temporal_constraint", + custom_extract, + ) + custom_query = "x" * (query_analyzer_module._MAX_TEMPORAL_ANALYSIS_CHARS + 1) + custom_result = await asyncio.wait_for( + retrieval_module.retrieve_all_fact_types_parallel( + object(), + query_text=custom_query, + query_embedding_str=_QUERY, + bank_id="test_custom_temporal_analyzer", + fact_types=["world"], + thinking_budget=10, + query_analyzer=object(), + graph_retriever=FakeGraphRetriever(), + ), + timeout=0.5, + ) + assert custom_result.results_by_fact_type["world"].temporal_constraint is None + assert custom_calls == [(custom_query, threading.get_ident())] + assert counting_executor.calls == 0 + + valid_start = datetime(2025, 1, 1, tzinfo=UTC) + valid_end = datetime(2025, 1, 31, 23, 59, 59, tzinfo=UTC) + monkeypatch.setattr( + "hindsight_api.engine.search.temporal_extraction.extract_temporal_constraint", + lambda *_args, **_kwargs: (valid_start, valid_end), + ) + boundary_result = await asyncio.wait_for( + retrieval_module.retrieve_all_fact_types_parallel( + object(), + query_text="x" * query_analyzer_module._MAX_TEMPORAL_ANALYSIS_CHARS, + query_embedding_str=_QUERY, + bank_id="test_temporal_boundary", + fact_types=["world"], + thinking_budget=10, + graph_retriever=FakeGraphRetriever(), + ), + timeout=0.5, + ) + assert boundary_result.results_by_fact_type["world"].temporal_constraint == (valid_start, valid_end) + assert counting_executor.calls == 1 + + monkeypatch.setattr( + "hindsight_api.engine.search.temporal_extraction.extract_temporal_constraint", + blocking_extract, + ) + + loop = asyncio.get_running_loop() + started_at = loop.time() + blocked_retrieval = asyncio.create_task( + retrieval_module.retrieve_all_fact_types_parallel( + object(), + query_text="what happened?", + query_embedding_str=_QUERY, + bank_id="test_temporal_off_loop", + fact_types=["world"], + thinking_budget=10, + graph_retriever=FakeGraphRetriever(), + ) + ) + + parser_started = await asyncio.wait_for(asyncio.to_thread(analysis_started.wait, 1), timeout=1.5) + event_loop_delay = loop.time() - started_at + assert parser_started + assert event_loop_delay < 0.5 + assert await asyncio.wait_for(asyncio.to_thread(lambda: "responsive"), timeout=0.5) == "responsive" + + blocked_retrieval.cancel() + with pytest.raises(asyncio.CancelledError): + await blocked_retrieval + + monkeypatch.setattr(retrieval_module, "_TEMPORAL_ANALYSIS_TIMEOUT_SECONDS", 0.05) + saturated_result = await asyncio.wait_for( + retrieval_module.retrieve_all_fact_types_parallel( + object(), + query_text="another query", + query_embedding_str=_QUERY, + bank_id="test_temporal_saturated", + fact_types=["world"], + thinking_budget=10, + graph_retriever=FakeGraphRetriever(), + ), + timeout=0.5, + ) + assert saturated_result.results_by_fact_type["world"].temporal_constraint is None + assert analysis_calls == 1 + assert counting_executor.calls == 3 + + cancelled_result = await asyncio.wait_for( + retrieval_module.retrieve_all_fact_types_parallel( + object(), + query_text="query after cancellation", + query_embedding_str=_QUERY, + bank_id="test_temporal_cancelled", + fact_types=["world"], + thinking_budget=10, + graph_retriever=FakeGraphRetriever(), + ), + timeout=0.5, + ) + assert cancelled_result.results_by_fact_type["world"].temporal_constraint is None + assert analysis_calls == 1 + assert counting_executor.calls == 3 + + release_analysis.set() + queued_parser_started = await asyncio.wait_for( + asyncio.to_thread(queued_analysis_started.wait, 1), + timeout=1.5, + ) + assert queued_parser_started + await asyncio.wait_for(asyncio.wrap_future(executor.submit(lambda: None)), timeout=1.0) + + recovered_result = await asyncio.wait_for( + retrieval_module.retrieve_all_fact_types_parallel( + object(), + query_text="query after recovery", + query_embedding_str=_QUERY, + bank_id="test_temporal_recovered", + fact_types=["world"], + thinking_budget=10, + graph_retriever=FakeGraphRetriever(), + ), + timeout=0.5, + ) + assert recovered_result.results_by_fact_type["world"].temporal_constraint is None + assert analysis_calls == 3 + assert counting_executor.calls == 4 + + def failed_extract(*_args, **_kwargs): + raise ValueError("parser failure") + + monkeypatch.setattr( + "hindsight_api.engine.search.temporal_extraction.extract_temporal_constraint", + failed_extract, + ) + failed_result = await asyncio.wait_for( + retrieval_module.retrieve_all_fact_types_parallel( + object(), + query_text="query that triggers a parser failure", + query_embedding_str=_QUERY, + bank_id="test_temporal_failure", + fact_types=["world"], + thinking_budget=10, + graph_retriever=FakeGraphRetriever(), + ), + timeout=0.5, + ) + assert failed_result.results_by_fact_type["world"].temporal_constraint is None + assert counting_executor.calls == 5 + executor.shutdown(wait=True) + + class RejectingExecutor: + calls = 0 + + def submit(self, *_args, **_kwargs): + self.calls += 1 + raise RuntimeError("executor unavailable") + + rejecting_executor = RejectingExecutor() + monkeypatch.setattr(retrieval_module, "_TEMPORAL_ANALYSIS_EXECUTOR", rejecting_executor) + monkeypatch.setattr(retrieval_module, "_TEMPORAL_ANALYSIS_SLOTS", threading.BoundedSemaphore(2)) + + for bank_id in ("test_temporal_submit_failure", "test_temporal_circuit_open"): + submit_failure_result = await asyncio.wait_for( + retrieval_module.retrieve_all_fact_types_parallel( + object(), + query_text="query after executor failure", + query_embedding_str=_QUERY, + bank_id=bank_id, + fact_types=["world"], + thinking_budget=10, + graph_retriever=FakeGraphRetriever(), + ), + timeout=0.5, + ) + assert submit_failure_result.results_by_fact_type["world"].temporal_constraint is None + + assert analysis_disabled.is_set() + assert rejecting_executor.calls == 1 + + # --------------------------------------------------------------------------- # DB-backed: similarity gating + window filter + coverage # --------------------------------------------------------------------------- From 7cab0399087b9a938f85bb83ed0460e2bf559867 Mon Sep 17 00:00:00 2001 From: r266-tech Date: Sun, 2 Aug 2026 11:56:11 +0800 Subject: [PATCH 2/2] chore(docs): sync generated OpenAPI skill --- skills/hindsight-docs/references/openapi.json | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/skills/hindsight-docs/references/openapi.json b/skills/hindsight-docs/references/openapi.json index 4a3cf6aef4..28b83303cb 100644 --- a/skills/hindsight-docs/references/openapi.json +++ b/skills/hindsight-docs/references/openapi.json @@ -3026,7 +3026,7 @@ "Documents" ], "summary": "List documents", - "description": "List documents with pagination and optional search. Documents are the source content from which memory units are extracted.", + "description": "List documents with pagination and optional search, most recently written first (`updated_at` descending). Documents are the source content from which memory units are extracted.", "operationId": "list_documents", "parameters": [ { @@ -6938,7 +6938,20 @@ "type": "null" } ], - "title": "Last Document At" + "title": "Last Document At", + "description": "When a document was last *ingested* into this bank. Appending to an existing document does not move this \u2014 use `last_write_at` for write activity." + }, + "last_write_at": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Last Write At", + "description": "When anything was last written to this bank: a document retained (including appends to an existing document) or a fact stored. Null if the bank is empty." } }, "type": "object", @@ -6977,6 +6990,7 @@ }, "fact_count": 156, "last_document_at": "2024-01-16T14:20:00Z", + "last_write_at": "2024-01-17T09:05:00Z", "mission": "I am a software engineer helping my team ship quality code", "name": "Alice", "updated_at": "2024-01-16T14:20:00Z"