From 243f8746ed11c6bf6eb13e8776b0117517bdd146 Mon Sep 17 00:00:00 2001 From: Varun Pratap Bhardwaj Date: Wed, 28 Oct 2026 10:58:56 +0530 Subject: [PATCH 1/8] fix: a slow explicit recall answers with labelled keyword matches, not 503 (#304) On a 37k-fact store every explicit recall overran its run budget and was told "busy, ask again in 2 seconds" on an idle daemon; every retry overran too. - An overrun now serves the keyword fallback, started beside the run when only its floor is left, labelled degraded_lexical and unjudged. Busy stays the answer only when no slot was free, where asking again does help. - The channel phase never waits past the caller's deadline (less a reserve to fuse, rerank and check); the 8 s guard is unchanged otherwise. - Semantic, hopfield and spreading activation shared nothing: three identical vec0 scans per recall. A search for the same vector already in flight now answers the others (k=500: 113 ms vs three scans of 93-113 ms each). Co-Authored-By: Claude Opus 5.5 --- .../retrieval/answer_check_status.py | 5 + src/superlocalmemory/retrieval/engine.py | 26 ++++- .../retrieval/vector_store.py | 40 ++++++- src/superlocalmemory/server/recall_core.py | 19 ++- .../test_a_slow_channel_is_waited_for.py | 33 ++++++ ...est_vector_search_is_shared_by_channels.py | 108 ++++++++++++++++++ .../test_server/test_explicit_recall_queue.py | 36 +++++- tests/test_views/test_one_run_path.py | 14 +-- 8 files changed, 262 insertions(+), 19 deletions(-) create mode 100644 tests/test_retrieval/test_vector_search_is_shared_by_channels.py diff --git a/src/superlocalmemory/retrieval/answer_check_status.py b/src/superlocalmemory/retrieval/answer_check_status.py index 125e2d382..3ab1fe3e9 100644 --- a/src/superlocalmemory/retrieval/answer_check_status.py +++ b/src/superlocalmemory/retrieval/answer_check_status.py @@ -148,6 +148,11 @@ def caller_deadline(deadline: float | None) -> Iterator[None]: _CALLER_DEADLINE.reset(token) +def caller_deadline_at() -> float | None: + """The monotonic instant this recall's caller stops waiting, or ``None``.""" + return _CALLER_DEADLINE.get() + + @dataclass(frozen=True) class JudgeOutcome: """A verdict (or None) and the status that explains it.""" diff --git a/src/superlocalmemory/retrieval/engine.py b/src/superlocalmemory/retrieval/engine.py index 9bf170dfe..52c65d6df 100644 --- a/src/superlocalmemory/retrieval/engine.py +++ b/src/superlocalmemory/retrieval/engine.py @@ -106,6 +106,25 @@ # measure what it costs in answer quality first and record the number. CHANNEL_HANG_GUARD_SECONDS = 8.0 +# #304: the guard never outlasts the caller. A channel still running when the +# caller is about to stop listening costs the whole answer, not just its own +# candidates, so the wait ends early enough to fuse, rerank and check what +# finished (the reserve), and the dropped channels are named as for any other +# timeout. Explicit recalls have ~16 s, so this binds only near their end. +CHANNEL_RESERVE_S = 1.5 +CHANNEL_WAIT_FLOOR_S = 1.0 + + +def _channel_wait_s() -> float: + """How long the channel phase may wait: the guard, or less near the caller's deadline.""" + from superlocalmemory.retrieval.answer_check_status import caller_deadline_at + + caller = caller_deadline_at() + if caller is None: + return CHANNEL_HANG_GUARD_SECONDS + left = caller - time.monotonic() - CHANNEL_RESERVE_S + return max(CHANNEL_WAIT_FLOOR_S, min(CHANNEL_HANG_GUARD_SECONDS, left)) + # How long a recall waits for a query vector from a model that has NEVER loaded # in this process (a fresh daemon). Only that case: a model that loaded before # is waited for unbounded, exactly as in 4.1.19 (retrieval/query_embedding.py). @@ -1414,9 +1433,8 @@ def _safe_channel(name: str, fn, *args): # One shared limit keeps parallel dispatch genuinely bounded. A # per-future timeout here would serialise the wait and turn five slow # channels into five seconds of UI latency. - done, pending = concurrent.futures.wait( - futures.values(), timeout=CHANNEL_HANG_GUARD_SECONDS, - ) + wait_s = _channel_wait_s() + done, pending = concurrent.futures.wait(futures.values(), timeout=wait_s) for name, fut in futures.items(): if fut in pending: # Not a latency notice: this answer is missing whatever this @@ -1425,7 +1443,7 @@ def _safe_channel(name: str, fn, *args): logger.error( "Channel %s did not finish within %.1fs; this recall is " "answering without it", - name, CHANNEL_HANG_GUARD_SECONDS, + name, wait_s, ) if dropped_channels is not None: dropped_channels.add(name) diff --git a/src/superlocalmemory/retrieval/vector_store.py b/src/superlocalmemory/retrieval/vector_store.py index a057b4d03..a55bb1c34 100644 --- a/src/superlocalmemory/retrieval/vector_store.py +++ b/src/superlocalmemory/retrieval/vector_store.py @@ -19,6 +19,7 @@ import re import sqlite3 import threading +from concurrent.futures import Future from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path @@ -31,6 +32,11 @@ logger = logging.getLogger(__name__) +#: Searches asking for this many or fewer share one scan (#304). Hopfield asks +#: for 500, semantic for 100, spreading activation for 10; on a 37k-vector store +#: k=500 measured 113 ms against 93-113 ms for each of the three separately. +SHARED_SEARCH_K = 512 + class VectorDelete(enum.Enum): """The three honest answers to "delete this fact's vector". @@ -95,6 +101,8 @@ def __init__(self, db_path: Path, config: VectorStoreConfig) -> None: self._db_path = Path(db_path) self._config = config self._lock = threading.Lock() + self._flights_lock = threading.Lock() + self._flights: dict[tuple, Future] = {} self._available = False if not config.enabled: @@ -499,12 +507,38 @@ def search( if len(query_embedding) != self._config.dimension: return [] + if top_k <= 0: + return [] vec_bytes = self._serialize_f32(query_embedding) - + # #304: semantic, hopfield and spreading activation ask with the same + # vector at the same moment, and vec0 scans every vector whatever k is. + # A small k asks for the shared k and cuts (the rows are ranked, so its + # head is exactly what a smaller k returns), and an identical search + # already running answers the others. Only searches in flight are + # shared: one that has finished is never reused, so writes are seen. + k = max(top_k, SHARED_SEARCH_K) + key = (profile_id, vec_bytes, k) + with self._flights_lock: + flight = self._flights.get(key) + leader = flight is None + if leader: + flight = self._flights[key] = Future() + if leader: + try: + flight.set_result(self._search_rows(vec_bytes, k, profile_id)) + except BaseException as exc: # noqa: BLE001 -- the waiters get it too + flight.set_exception(exc) + finally: + with self._flights_lock: + self._flights.pop(key, None) + return flight.result()[:top_k] + + def _search_rows( + self, vec_bytes: bytes, top_k: int, profile_id: str | None, + ) -> list[tuple[str, float]]: + """The ranked nearest ``top_k`` (fact_id, cosine) pairs; ``[]`` on error.""" try: with self._managed_connection() as conn: - if top_k <= 0: - return [] # A withheld fact must not occupy a nearest-neighbour slot. # # Its vector stays in the index — quarantine is reversible and diff --git a/src/superlocalmemory/server/recall_core.py b/src/superlocalmemory/server/recall_core.py index 348b21ab0..0d1fd722e 100644 --- a/src/superlocalmemory/server/recall_core.py +++ b/src/superlocalmemory/server/recall_core.py @@ -451,7 +451,15 @@ async def _run_admitted(engine, call, app_state, runtime, snapshot, future = submit(loop, hold, runtime, app_state, engine, _engine_call, engine, call) hedge = None if explicit: - response = await wait_until(future, run_deadline) + # #304: a run still going when only the fallback's floor is left + # starts the keyword fallback beside it, so an overrun answers with + # labelled keyword matches. It used to say "busy, ask again in 2 s" + # on an idle daemon; on a 37k-fact store every retry overran too. + response = await wait_until(future, run_deadline - RECALL_BUDGET_FLOOR_S) + if response is None: + hedge = submit(loop, hold, runtime, app_state, engine, + _fallback, engine, call, app_state) + response = await wait_until(future, run_deadline) else: # Delayed hedge: a short-clock run still going near its deadline # starts the keyword fallback beside it, so an overrun answers with @@ -465,10 +473,13 @@ async def _run_admitted(engine, call, app_state, runtime, snapshot, _fallback, engine, call, app_state) response = await wait_until(future, worker_deadline) if response is None: + # Busy is only the answer when no slot was free (above): asking + # again helps a full house, never a run that overran. if explicit: - logger.warning("Explicit recall overran its run budget; reporting busy") - return busy(call, snapshot, retry_after_s=_retry_after()) - logger.warning("Recall deadline reached; serving unjudged keyword fallback") + logger.warning("Explicit recall overran its run budget; " + "serving unjudged keyword fallback") + else: + logger.warning("Recall deadline reached; serving unjudged keyword fallback") return await _bounded_fallback( engine, call, app_state, public_deadline, snapshot, hold, runtime, started=hedge) diff --git a/tests/test_retrieval/test_a_slow_channel_is_waited_for.py b/tests/test_retrieval/test_a_slow_channel_is_waited_for.py index 0ce5ee76f..349e76e06 100644 --- a/tests/test_retrieval/test_a_slow_channel_is_waited_for.py +++ b/tests/test_retrieval/test_a_slow_channel_is_waited_for.py @@ -231,3 +231,36 @@ def test_the_sink_is_not_shared_between_concurrent_recalls(self) -> None: "a mutable default would accumulate across every recall in the " "process and report channels that this query never lost" ) + + +class TestTheWaitEndsBeforeTheCallerStopsListening: + """#304: the guard never outlasts the caller. + + On a 37k-fact store three channels ran past the explicit recall's deadline, + so the server gave up and the engine kept waiting for an answer nobody would + read. With the caller's deadline known, the wait ends in time to fuse what + finished, and the dropped channels are named as for any other timeout. + """ + + def test_no_caller_deadline_keeps_the_full_guard(self) -> None: + assert engine_mod._channel_wait_s() == CHANNEL_HANG_GUARD_SECONDS + + def test_a_distant_deadline_keeps_the_full_guard(self) -> None: + from superlocalmemory.retrieval.answer_check_status import caller_deadline + + with caller_deadline(time.monotonic() + 60.0): + assert engine_mod._channel_wait_s() == CHANNEL_HANG_GUARD_SECONDS + + def test_a_near_deadline_leaves_room_to_answer(self) -> None: + from superlocalmemory.retrieval.answer_check_status import caller_deadline + + with caller_deadline(time.monotonic() + 5.0): + wait = engine_mod._channel_wait_s() + assert wait <= 5.0 - engine_mod.CHANNEL_RESERVE_S + 0.05 + assert wait >= engine_mod.CHANNEL_WAIT_FLOOR_S + + def test_a_passed_deadline_still_gives_the_floor(self) -> None: + from superlocalmemory.retrieval.answer_check_status import caller_deadline + + with caller_deadline(time.monotonic() - 1.0): + assert engine_mod._channel_wait_s() == engine_mod.CHANNEL_WAIT_FLOOR_S diff --git a/tests/test_retrieval/test_vector_search_is_shared_by_channels.py b/tests/test_retrieval/test_vector_search_is_shared_by_channels.py new file mode 100644 index 000000000..1515286c5 --- /dev/null +++ b/tests/test_retrieval/test_vector_search_is_shared_by_channels.py @@ -0,0 +1,108 @@ +# Copyright (c) 2026 Varun Pratap Bhardwaj / Qualixar +# Licensed under AGPL-3.0-or-later - see LICENSE file +"""Channels asking the same question at once share one vector scan (#304). + +Semantic, hopfield and spreading activation each run a nearest-neighbour search +with the same query vector, at the same moment, on every recall. vec0 scans +every vector whatever k is, so three scans cost three times one. Thread dumps +of slow recalls on a 37k-fact store found all three channels inside that one +query. A search already running for the same vector now answers the others. +""" + +from __future__ import annotations + +import sqlite3 +import threading +from pathlib import Path + +import numpy as np +import pytest + +from superlocalmemory.retrieval import vector_store as vs_mod +from superlocalmemory.retrieval.vector_store import VectorStore, VectorStoreConfig +from superlocalmemory.storage import schema as real_schema + +DIM = 4 + + +def _store(tmp_path: Path) -> VectorStore: + path = tmp_path / "v.db" + conn = sqlite3.connect(str(path)) + real_schema.create_all_tables(conn) + conn.commit() + conn.close() + store = VectorStore(path, VectorStoreConfig(dimension=DIM, enabled=True)) + if not store.available: + pytest.skip("sqlite-vec not loadable here") + rng = np.random.default_rng(7) + for i in range(120): + v = rng.normal(size=DIM) + store.upsert(f"f-{i:03d}", "default", (v / np.linalg.norm(v)).tolist()) + return store + + +def test_a_small_k_answers_what_it_answered_alone(tmp_path: Path) -> None: + """Asking with the shared k and cutting gives the same nearest facts.""" + store = _store(tmp_path) + q = [0.4, -0.1, 0.8, 0.2] + alone = store._search_rows(store._serialize_f32(q), 10, "default") + assert store.search(q, top_k=10, profile_id="default") == alone + assert store.search(q, top_k=3, profile_id="default") == alone[:3] + + +def test_concurrent_identical_searches_run_one_scan(tmp_path: Path, monkeypatch) -> None: + store = _store(tmp_path) + q = [0.1, 0.9, -0.3, 0.2] + scans: list[int] = [] + entered = threading.Event() + release = threading.Event() + real = store._search_rows + + def slow(vec_bytes, k, profile_id): + scans.append(k) + entered.set() + release.wait(5) + return real(vec_bytes, k, profile_id) + + monkeypatch.setattr(store, "_search_rows", slow) + out: dict[str, list] = {} + + def ask(name, k): + out[name] = store.search(q, top_k=k, profile_id="default") + + first = threading.Thread(target=ask, args=("hopfield", 100)) + first.start() + assert entered.wait(5) + rest = [threading.Thread(target=ask, args=(n, k)) + for n, k in (("semantic", 50), ("spreading_activation", 10))] + for t in rest: + t.start() + release.set() + for t in (first, *rest): + t.join(5) + + assert scans == [vs_mod.SHARED_SEARCH_K] # one scan answered all three + assert out["semantic"] == out["hopfield"][:50] + assert out["spreading_activation"] == out["hopfield"][:10] + + +def test_a_finished_search_is_not_reused(tmp_path: Path, monkeypatch) -> None: + """Sharing is only for searches in flight: a later one sees new writes.""" + store = _store(tmp_path) + q = [0.2, 0.2, 0.2, 0.9] + before = store.search(q, top_k=5, profile_id="default") + store.upsert("f-new", "default", (np.array(q) / np.linalg.norm(q)).tolist()) + after = store.search(q, top_k=5, profile_id="default") + assert after[0][0] == "f-new" and after != before + + +def test_profiles_and_large_k_are_never_shared(tmp_path: Path, monkeypatch) -> None: + store = _store(tmp_path) + seen: list[tuple[int, str | None]] = [] + monkeypatch.setattr(store, "_search_rows", + lambda vb, k, pid: seen.append((k, pid)) or []) + q = [0.5, 0.5, 0.5, 0.5] + store.search(q, top_k=5, profile_id="other") + store.search(q, top_k=vs_mod.SHARED_SEARCH_K + 50, profile_id="default") + assert seen == [(vs_mod.SHARED_SEARCH_K, "other"), + (vs_mod.SHARED_SEARCH_K + 50, "default")] diff --git a/tests/test_server/test_explicit_recall_queue.py b/tests/test_server/test_explicit_recall_queue.py index 116b4c248..97495b345 100644 --- a/tests/test_server/test_explicit_recall_queue.py +++ b/tests/test_server/test_explicit_recall_queue.py @@ -308,6 +308,40 @@ async def run(): return out, held_after_busy, gate.locked() out, held_after_busy, held_at_end = asyncio.run(run()) - assert out["code"] == "RECALL_BUSY" + assert out.get("code") != "RECALL_BUSY" # #304: an overrun is not a full house assert held_after_busy is True # the thread still runs: the slot stays taken assert held_at_end is False # and is given back when it ends + + +def test_an_overrun_answers_with_labelled_keyword_matches_not_503(monkeypatch, runtime): + """#304: a started run that cannot finish is not "busy". + + On a 37k-fact store every explicit recall overran and was told to ask again + in 2 seconds, forever: retrying never helps an overrun, only a full house. + The keyword fallback answers instead, labelled as unchecked keyword matches. + """ + release = threading.Event() + + def stuck(*a): + release.wait(5) + return object() + + monkeypatch.setattr(recall_core, "_engine_call", stuck) + monkeypatch.setattr(recall_core, "effective_budget_s", lambda call: 0.05) + monkeypatch.setattr(recall_core, "EXPLICIT_RUN_GRACE_S", 0.05) + monkeypatch.setattr(recall_core, "EXPLICIT_CEILING_S", 0.6) + monkeypatch.setattr(recall_core, "RECALL_BUDGET_FLOOR_S", 0.3) + async def run(): + start = time.monotonic() + out = await recall_core.run_recall(object(), _call(), app_state=SimpleNamespace()) + return out, time.monotonic() - start # not the loop's wait for the stuck thread + + try: + out, elapsed = asyncio.run(run()) + finally: + release.set() + assert out["answerability"] == "unjudged" + assert out["results"] == [{"content": "keyword guess"}] + assert "keyword" in out["answer_check_note"].lower() + runtime.assert_called_once() + assert elapsed < 1.5 # answered near its own ceiling, not after the stuck run diff --git a/tests/test_views/test_one_run_path.py b/tests/test_views/test_one_run_path.py index 0df16c38d..8240ce0bb 100644 --- a/tests/test_views/test_one_run_path.py +++ b/tests/test_views/test_one_run_path.py @@ -102,9 +102,9 @@ def test_one_view_gives_one_answer_on_every_surface(daemon, capsys) -> None: def test_the_over_budget_answer_is_shared_too(daemon, monkeypatch) -> None: - """Past its budget an explicit question is told to ask again (503 busy), - never handed a keyword guess; a view runs the same function, so it says the - same thing.""" + """Past its budget an explicit question answers with keyword matches labelled + unchecked (#304: a 503 "ask again" on an overrun never succeeded on retry); + a view runs the same function, so it says the same thing.""" import time from superlocalmemory.server import recall_core @@ -119,10 +119,10 @@ def slow(*a, **k): view = daemon.client.get("/api/v3/views/run", params={"name": "Widget"}) direct = daemon.client.get("/recall", params={"q": "widget release", "limit": 4}) for res in (view, direct): - assert res.status_code == 503, res.text - assert int(res.headers["Retry-After"]) >= 1 - assert res.json()["code"] == "RECALL_BUSY" - assert res.json()["results"] == [] + assert res.status_code == 200, res.text + assert "Retry-After" not in res.headers + assert res.json().get("code") != "RECALL_BUSY" + assert res.json()["retrieval_mode"] == "degraded_lexical" def test_the_history_says_which_surface_ran_the_view(daemon, capsys, monkeypatch) -> None: From ab14873344a0ac2a5f9fa6ffa4cf6b68274a213d Mon Sep 17 00:00:00 2001 From: Varun Pratap Bhardwaj Date: Wed, 28 Oct 2026 11:02:41 +0530 Subject: [PATCH 2/8] fix: a note retired while its save is still queryable settles, so purge finishes (#297) Revising or deleting a folder note while its save was still `queryable` archived or erased the save's facts; enrichment then failed three times with "queryable fact profile mismatch or missing fact", and `sources remove --purge` refused until every retry ran out (2 s doubling to 300 s). When none of the save's queryable facts is live any more (archived, gone, or being erased) the operation now settles as complete at once, with no final facts and derivation_state {"superseded": true}, and enriches nothing back. Its queryable ids stay on the row for the erasure scrubs. A fact still live, or in another profile, keeps the old error: those are real faults. Co-Authored-By: Claude Opus 5.5 --- src/superlocalmemory/core/engine_ingestion.py | 36 ++++++++++- .../core/ingestion_command.py | 18 +++++- .../core/materialization_control.py | 9 +++ ...est_canonical_ingestion_materialization.py | 59 +++++++++++++++++++ 4 files changed, 120 insertions(+), 2 deletions(-) diff --git a/src/superlocalmemory/core/engine_ingestion.py b/src/superlocalmemory/core/engine_ingestion.py index cb0de8d03..050ccfad4 100644 --- a/src/superlocalmemory/core/engine_ingestion.py +++ b/src/superlocalmemory/core/engine_ingestion.py @@ -13,7 +13,7 @@ import hashlib import logging import uuid -from typing import TYPE_CHECKING, Protocol +from typing import TYPE_CHECKING, Any, Protocol from superlocalmemory.core.derivation_lineage import checkpoint_with_lineage from superlocalmemory.core.ingestion_command import ( @@ -37,6 +37,35 @@ _DERIVATION_VERSION = "v3.7-ingestion-1" +def _nothing_left_live(db: Any, operation: IngestionOperation) -> bool: + """True when every queryable fact of the save is archived, erased or being erased. + + That is a save retired before it was enriched (a folder note revised or + deleted while it was still queryable, #297), not a broken one. A fact that + is still live in this profile, or that sits in another profile, keeps the + "mismatch or missing fact" error: those are real faults. + """ + ids = list(operation.queryable_fact_ids) + if not ids: + return False + marks = ",".join("?" for _ in ids) + rows = db.execute( + f"SELECT fact_id, profile_id, COALESCE(archive_status, 'live') AS status " + f"FROM atomic_facts WHERE fact_id IN ({marks})", tuple(ids)) + try: + erasing = {str(dict(r)["fact_id"]) for r in db.execute( + f"SELECT fact_id FROM projection_tombstones WHERE profile_id=? " + f"AND fact_id IN ({marks})", (operation.profile_id, *ids))} + except Exception: # noqa: BLE001 -- a schema without tombstones erases nothing + erasing = set() + for row in map(dict, rows): + if row["profile_id"] != operation.profile_id: + return False + if row["status"] == "live" and str(row["fact_id"]) not in erasing: + return False + return True + + def _require_known_profile(engine: "MemoryEngine", profile_id: str) -> None: """Fail closed on a write routed to a profile that does not exist. @@ -828,6 +857,11 @@ def materialize(operation: IngestionOperation) -> MaterializationResult: operation.profile_id, ) if len(facts) != len(operation.queryable_fact_ids): + if _nothing_left_live(engine._db, operation): + from superlocalmemory.core.materialization_control import ( + MaterializationSuperseded, + ) + raise MaterializationSuperseded(operation.operation_id) raise ValueError("queryable fact profile mismatch or missing fact") memory_ids = {fact.memory_id for fact in facts} if len(memory_ids) != 1: diff --git a/src/superlocalmemory/core/ingestion_command.py b/src/superlocalmemory/core/ingestion_command.py index 0311c23b0..c9c3eb33c 100644 --- a/src/superlocalmemory/core/ingestion_command.py +++ b/src/superlocalmemory/core/ingestion_command.py @@ -23,7 +23,10 @@ from typing import Any, Callable from superlocalmemory.core.ingestion_lease_owner import owner_token, release_dead_owner_leases -from superlocalmemory.core.materialization_control import MaterializationDeferred +from superlocalmemory.core.materialization_control import ( + MaterializationDeferred, + MaterializationSuperseded, +) from superlocalmemory.storage.database import DatabaseManager from superlocalmemory.storage.idempotency_identity import same_principal @@ -1007,6 +1010,19 @@ def _materialize_locked( operation_id, owner=self._owner, ) + except MaterializationSuperseded: + # #297: the save was retired before it was enriched. Done, with + # nothing enriched back; its queryable ids stay on the row for the + # erasure scrubs, and "superseded" says why it has no final facts. + logger.info("ingestion %s was retired before enrichment; settled", operation_id) + return self.repository.finish_enriching( + operation_id, + owner=self._owner, + target=IngestionState.COMPLETE, + final_fact_ids=(), + derivation_version=self._derivation_version, + derivation_state={"superseded": True}, + ) except Exception as exc: return self.repository.finish_enriching( operation_id, diff --git a/src/superlocalmemory/core/materialization_control.py b/src/superlocalmemory/core/materialization_control.py index 32ab37273..2ece3e4ce 100644 --- a/src/superlocalmemory/core/materialization_control.py +++ b/src/superlocalmemory/core/materialization_control.py @@ -18,3 +18,12 @@ class MaterializationDeferred(RuntimeError): The queryable projection is already durable. This signal must be handled by :class:`IngestionCommand` as a requeue, never as a failed attempt. """ + + +class MaterializationSuperseded(RuntimeError): + """Nothing of a queryable save is live any more: it was retired before enrichment. + + A folder note revised or deleted while its save was still queryable has its + facts archived or erased (#297). Enriching it would put retired content back, + so :class:`IngestionCommand` settles the operation instead of failing it. + """ diff --git a/tests/test_core/test_canonical_ingestion_materialization.py b/tests/test_core/test_canonical_ingestion_materialization.py index 790cb9d76..5c7bb8d53 100644 --- a/tests/test_core/test_canonical_ingestion_materialization.py +++ b/tests/test_core/test_canonical_ingestion_materialization.py @@ -887,6 +887,65 @@ def test_materializer_rejects_queryable_facts_from_another_profile( assert "profile" in failed.last_error.lower() +def _queryable_save(engine, key: str): + _install_m018(engine) + command = build_engine_ingestion_command(engine) + receipt = command.submit(IngestionRequest( + content="Fictional EmberRail-731 uses CopperLake version 3 for 11 days.", + profile_id=engine._profile_id, + source_type="folder", + idempotency_key=key, + trusted_actor_id="daemon-capability:owned-instance", + )) + assert receipt.state is IngestionState.QUERYABLE + return command, receipt + + +@pytest.mark.parametrize("retire", ["archived", "erased"]) +def test_a_save_retired_before_enrichment_settles_instead_of_failing( + engine_with_mock_deps, retire, +) -> None: + """#297 follow-up: a note revised or deleted while its save was still + queryable had its facts hidden or erased; enrichment then failed three + times with "queryable fact profile mismatch or missing fact", and the purge + could not finish until every retry ran out. Nothing is left to enrich: the + operation settles at once, says why, and enriches nothing back.""" + engine = engine_with_mock_deps + command, receipt = _queryable_save(engine, f"folder:retired-{retire}") + fact_id = receipt.queryable_fact_ids[0] + if retire == "archived": + engine._db.execute( + "UPDATE atomic_facts SET archive_status='archived' WHERE fact_id=?", (fact_id,)) + else: + engine._db.execute("DELETE FROM atomic_facts WHERE fact_id=?", (fact_id,)) + + with patch.object(engine._fact_extractor, "extract_facts") as extract: + settled = command.materialize(receipt.operation_id) + + assert settled.state is IngestionState.COMPLETE + assert settled.attempt_count == 1 and settled.last_error == "" + assert settled.derivation_state == {"superseded": True} + assert settled.final_fact_ids == () + assert settled.queryable_fact_ids == (fact_id,) # erasure scrubs still find it + extract.assert_not_called() + live = engine._db.execute( + "SELECT COUNT(*) AS n FROM atomic_facts WHERE profile_id=? " + "AND COALESCE(archive_status, 'live')='live'", (engine._profile_id,)) + assert dict(live[0])["n"] == 0 # nothing came back + + +def test_a_save_with_a_live_fact_left_is_still_an_error(engine_with_mock_deps) -> None: + """Superseded only when nothing of the save is live: a fact moved to another + profile is a real mismatch and still fails loudly.""" + engine = engine_with_mock_deps + command, receipt = _queryable_save(engine, "folder:moved") + engine._db.execute("INSERT INTO profiles(profile_id, name) VALUES ('foreign', 'Foreign')") + engine._db.execute("UPDATE atomic_facts SET profile_id='foreign' WHERE fact_id=?", + (receipt.queryable_fact_ids[0],)) + failed = command.materialize(receipt.operation_id) + assert failed.state is IngestionState.FAILED + + @pytest.mark.parametrize( ("component", "stage", "error_text"), [ From d83428ad3b7c6177c0856388b6afe054035af59b Mon Sep 17 00:00:00 2001 From: Varun Pratap Bhardwaj Date: Wed, 28 Oct 2026 11:04:37 +0530 Subject: [PATCH 3/8] fix(gateway): the connection picker tells identical entries apart and can prune old ones (#301) After local reinstalls the GitHub sign-in page listed several identical "default (chatgpt)" connections. Each option now shows a short connection ID and its credential expiry, newest first. An opt-in, never prechecked checkbox revokes the other connections with the same profile and app on this GitHub account, through the same steps as /owner/revoke, fenced to the signed-in owner and re-read before each revoke. On top of the reported patch: cleanup runs only after the consent is approved (a refused consent must not leave older connections revoked with no new one), a cleanup failure never blocks the chosen connection, and a device the kept connection still uses is never revoked. Patch by @DannyGi82 in #301. Co-Authored-By: Claude Opus 5.5 --- .../remote-gateway/src/owner-selection.ts | 24 +++++++ .../src/prune-owner-connections.ts | 45 +++++++++++++ .../remote-gateway/src/worker-auth.ts | 18 +++++- .../tests/owner-profile-cleanup.test.mjs | 64 +++++++++++++++++++ .../tests/owner-selection.test.mjs | 23 +++++++ 5 files changed, 172 insertions(+), 2 deletions(-) create mode 100644 integrations/remote-gateway/src/owner-selection.ts create mode 100644 integrations/remote-gateway/src/prune-owner-connections.ts create mode 100644 integrations/remote-gateway/tests/owner-profile-cleanup.test.mjs create mode 100644 integrations/remote-gateway/tests/owner-selection.test.mjs diff --git a/integrations/remote-gateway/src/owner-selection.ts b/integrations/remote-gateway/src/owner-selection.ts new file mode 100644 index 000000000..74d1546be --- /dev/null +++ b/integrations/remote-gateway/src/owner-selection.ts @@ -0,0 +1,24 @@ +/** GitHub-account-scoped SLM connection picker, with explicit owner cleanup. */ +type Connection = { + connectionId:string; profileId:string; host:string; deviceExpiresAtMs:number; +}; +const htmlEscape=(value:string):string=>value.replaceAll('&','&').replaceAll('<','<') + .replaceAll('>','>').replaceAll('"','"').replaceAll("'",'''); + +export function renderConnectionSelector( + connections:readonly Connection[],handle:string,scopes:readonly string[] +):string { + const ordered=[...connections].sort((a,b)=>b.deviceExpiresAtMs-a.deviceExpiresAtMs); + const options=ordered.map(c=>{ + const expiry=Number.isFinite(c.deviceExpiresAtMs)? + new Date(c.deviceExpiresAtMs).toISOString().slice(0,16).replace('T',' '):'unknown'; + const label=c.profileId+' ('+c.host+') — ID '+c.connectionId.slice(0,8)+' — expires '+expiry+' UTC'; + return ''; + }).join(''); + const cleanup=ordered.length>1? + '':''; + return '

Choose your SLM connection

GitHub sign-in is complete. Select the local profile this application may use. The ID distinguishes otherwise identical connections; verify the current one in your local SLM dashboard.

'+ + '
'+ + cleanup+'
'+ + '
Approved permissions

'+scopes.map(htmlEscape).join(', ')+'

'; +} diff --git a/integrations/remote-gateway/src/prune-owner-connections.ts b/integrations/remote-gateway/src/prune-owner-connections.ts new file mode 100644 index 000000000..33142ef44 --- /dev/null +++ b/integrations/remote-gateway/src/prune-owner-connections.ts @@ -0,0 +1,45 @@ +/** + * Account-owner cleanup initiated by explicit consent on the GitHub-authenticated + * OAuth selection page. This helper is never exposed as an unauthenticated API. + */ +import type {OwnedConnection,OwnerIndexDO} from './owner-index-do.ts'; + +interface OwnerCleanupEnv { + REGISTRIES: {getByName(id:string):{revokeConnection(ownerId:string,generation:number):Promise}}; + DEVICES: {getByName(id:string):{revoke(ownerId:string):Promise}}; + RELAYS: {getByName(id:string):{revoke():Promise}}; +} +type OwnerOps=Pick; + +/** + * Preserve selected current connection and any other host/profile pair. The + * caller is responsible for verifying OAuth owner identity, CSRF and explicit + * opt-in before entering this function. + */ +export async function pruneOtherOwnerConnections({owner,ownerId,selectedId,env}:{ + owner:OwnerOps;ownerId:string;selectedId:string;env:OwnerCleanupEnv; +}):Promise{ + if(!/^[a-f0-9]{32}$/.test(selectedId))throw new Error('invalid_selected_connection'); + const selected=await owner.getConnection(ownerId,selectedId); + if(!selected||selected.revokedAt!==null)throw new Error('selected_connection_unavailable'); + const all=await owner.list(ownerId); + const obsolete=all.filter(row=>row.connectionId!==selectedId&&row.revokedAt===null&& + row.profileId===selected.profileId&&row.host===selected.host); + const revoked:string[]=[]; + for(const row of obsolete){ + // The owner Directory is authoritative: changes between listing and + // revocation are fenced via its current generation. + const live=await owner.getConnection(ownerId,row.connectionId); + if(!live||live.revokedAt!==null||live.profileId!==selected.profileId|| + live.host!==selected.host)continue; + await owner.cancelConnection(ownerId,live.connectionId); + const generation=await owner.revoke(ownerId,live.connectionId,live.generation); + await env.REGISTRIES.getByName(live.connectionId).revokeConnection(ownerId,1); + // A device the kept connection still uses is never revoked with the old one. + if(live.deviceDigest!==selected.deviceDigest)await env.DEVICES.getByName(live.deviceDigest).revoke(ownerId); + await env.RELAYS.getByName(live.connectionId).revoke(); + await owner.markCleanup(ownerId,live.connectionId,generation); + revoked.push(live.connectionId); + } + return revoked; +} diff --git a/integrations/remote-gateway/src/worker-auth.ts b/integrations/remote-gateway/src/worker-auth.ts index e1fcd85a6..d4fd496a0 100644 --- a/integrations/remote-gateway/src/worker-auth.ts +++ b/integrations/remote-gateway/src/worker-auth.ts @@ -14,6 +14,8 @@ import {tokenHash} from './device-proof.ts'; import type {BootstrapBinding} from './bootstrap-do.ts'; import type {AuthProps,Scope} from './contracts.ts'; import {toolsForScopes} from './request-policy.ts'; +import {renderConnectionSelector} from './owner-selection.ts'; +import {pruneOtherOwnerConnections} from './prune-owner-connections.ts'; export interface AuthWorkerEnv extends AuthorizationEnv,IssuedTokenEnv,ConnectEnv,SetupAdmissionEnv {GITHUB_CLIENT_ID:string;GITHUB_CLIENT_SECRET:string;DEVICE_WRAP_KEY:string;} interface ConsentContext {request:AuthRequest;bootstrapId?:string;ownerId?:string;} const LOOPBACK_HOSTS=['127.0.0.1','localhost','[::1]']; @@ -109,8 +111,7 @@ async function githubCallback(request:Request,env:AuthWorkerEnv):PromiseLink your computer first

No active computer connection is available for this GitHub account. Open your local SLM dashboard, enable a web connection, then reconnect this application using the same GitHub account.

'),resumed.headers,resumed.request.redirectUri); const consent=await api.beginConsent(resumed.request); await env.OAUTH_KV.put('slm-consent:'+consent.handle,JSON.stringify({request:resumed.request,ownerId}),{expirationTtl:900}); - const options=connections.map(c=>'').join(''); - return html(renderAuthPage('Choose your SLM connection','

Choose your SLM connection

GitHub sign-in is complete. Select the local profile this application may use.

Approved permissions

'+resumed.request.scope.map(escapeHtml).join(', ')+'

'),consent.headers,resumed.request.redirectUri); + return html(renderAuthPage('Choose your SLM connection',renderConnectionSelector(connections,consent.handle,resumed.request.scope)),consent.headers,resumed.request.redirectUri); } async function selectConnection(request:Request,env:AuthWorkerEnv):Promise{ const fields=await form(request);const handle=fields.get('handle')??'';const saved=await consentContext(handle,env);if(!saved?.ownerId)return missingConsent(request,handle,env); @@ -123,6 +124,19 @@ async function selectConnection(request:Request,env:AuthWorkerEnv):Promise { + const ownerId='1234'; + const entries=[ + {connectionId:'a'.repeat(32),ownerId,profileId:'default',host:'chatgpt',revokedAt:null,generation:2,deviceDigest:'a'.repeat(64)}, + {connectionId:'b'.repeat(32),ownerId,profileId:'default',host:'chatgpt',revokedAt:null,generation:1,deviceDigest:'b'.repeat(64)}, + {connectionId:'c'.repeat(32),ownerId,profileId:'default',host:'chatgpt',revokedAt:null,generation:1,deviceDigest:'c'.repeat(64)}, + {connectionId:'d'.repeat(32),ownerId,profileId:'default',host:'claude_web',revokedAt:null,generation:1,deviceDigest:'d'.repeat(64)}, + {connectionId:'e'.repeat(32),ownerId,profileId:'bughunting',host:'chatgpt',revokedAt:null,generation:1,deviceDigest:'e'.repeat(64)}, + {connectionId:'f'.repeat(32),ownerId,profileId:'default',host:'chatgpt',revokedAt:'2026-10-23',generation:2,deviceDigest:'f'.repeat(64)} + ]; + const steps=[]; + const owner={ + async list(id){assert.equal(id,ownerId);return structuredClone(entries);}, + async getConnection(id,key){assert.equal(id,ownerId);return structuredClone(entries.find(x=>x.connectionId===key)??null);}, + async cancelConnection(id,key){steps.push(['cancel',key]);}, + async revoke(id,key,generation){steps.push(['revoke',key,generation]);const row=entries.find(x=>x.connectionId===key);if(row)row.revokedAt='now';return generation+1;}, + async markCleanup(id,key,generation){steps.push(['cleanup',key,generation]);return true;} + }; + const env={ + REGISTRIES:{getByName(id){return {async revokeConnection(){steps.push(['registry',id]);}}}}, + DEVICES:{getByName(id){return {async revoke(){steps.push(['device',id]);}}}}, + RELAYS:{getByName(id){return {async revoke(){steps.push(['relay',id]);}}}} + }; + return {owner,ownerId,env,steps,entries}; +}; +test('explicit owner-scoped cleanup revokes only other same-profile same-host connections',async()=>{ + const f=fixture();const ids=await pruneOtherOwnerConnections({owner:f.owner,ownerId:f.ownerId,selectedId:'a'.repeat(32),env:f.env}); + assert.deepEqual(ids,['b'.repeat(32),'c'.repeat(32)]); + assert.equal(f.steps.filter(x=>x[0]==='revoke').length,2); + assert.ok(!f.steps.some(x=>['a','d','e','f'].includes(x[1]?.[0]))); + assert.equal(f.entries.filter(x=>x.revokedAt===null&&x.profileId==='default'&&x.host==='chatgpt').length,1); +}); +test('rejects stale selected connection before any revocations',async()=>{ + const f=fixture();f.entries[0].revokedAt='old'; + await assert.rejects(()=>pruneOtherOwnerConnections({owner:f.owner,ownerId:f.ownerId,selectedId:'a'.repeat(32),env:f.env}),/selected_connection_unavailable/); + assert.equal(f.steps.length,0); +}); +test('rejects unknown selection before any revocations',async()=>{ + const f=fixture(); + await assert.rejects(()=>pruneOtherOwnerConnections({owner:f.owner,ownerId:f.ownerId,selectedId:'9'.repeat(32),env:f.env}),/selected_connection_unavailable/); + assert.equal(f.steps.length,0); +}); +test('does not mark cleanup complete if remote revocation fails',async()=>{ + const f=fixture(); + f.env.REGISTRIES.getByName=()=>({async revokeConnection(){throw Error('relay outage')}}); + await assert.rejects(()=>pruneOtherOwnerConnections({owner:f.owner,ownerId:f.ownerId,selectedId:'a'.repeat(32),env:f.env}),/relay outage/); + assert.equal(f.steps.some(x=>x[0]==='cleanup'),false); +}); +test('validates selected identifier to prevent cross-account lookup',async()=>{ + const f=fixture(); + await assert.rejects(()=>pruneOtherOwnerConnections({owner:f.owner,ownerId:f.ownerId,selectedId:'../../not-a-uuid',env:f.env}),/invalid_selected_connection/); + assert.equal(f.steps.length,0); +}); +test('never revokes the device the kept connection still uses',async()=>{ + const f=fixture();f.entries[1].deviceDigest=f.entries[0].deviceDigest; + await pruneOtherOwnerConnections({owner:f.owner,ownerId:f.ownerId,selectedId:'a'.repeat(32),env:f.env}); + const devices=f.steps.filter(x=>x[0]==='device').map(x=>x[1]); + assert.deepEqual(devices,['c'.repeat(64)]); + assert.equal(f.steps.filter(x=>x[0]==='revoke').length,2); +}); diff --git a/integrations/remote-gateway/tests/owner-selection.test.mjs b/integrations/remote-gateway/tests/owner-selection.test.mjs new file mode 100644 index 000000000..4e9f4c259 --- /dev/null +++ b/integrations/remote-gateway/tests/owner-selection.test.mjs @@ -0,0 +1,23 @@ +import {test} from 'node:test'; +import assert from 'node:assert/strict'; +import {renderConnectionSelector} from '../src/owner-selection.ts'; +const c=(id,expires,profile='default',host='chatgpt')=>({connectionId:id,deviceExpiresAtMs:expires,profileId:profile,host,revokedAt:null}); +test('distinguishes identical profile labels with nonsecret short IDs and expiry, recent first',()=>{ + const page=renderConnectionSelector([c('a'.repeat(32),20000),c('b'.repeat(32),50000),c('c'.repeat(32),40000)],'safe-handle',['slm:read']); + assert.equal((page.match(/default \(chatgpt\)/g)||[]).length,3); + assert.ok(page.indexOf('bbbbbbbb'){ + const page=renderConnectionSelector([c('b'.repeat(32),50000),c('a'.repeat(32),40000)],'safe-handle',['slm:read']); + assert.match(page,/name="remove_duplicates" value="yes"/); + assert.doesNotMatch(page,/\bchecked\b/); + assert.match(page,/other connections with the same local profile and app/i); +}); +test('escapes names and consent handle from external data',()=>{ + const page=renderConnectionSelector([c('a'.repeat(32),30000,'','chatgpt')],'',['slm:read']); + assert.doesNotMatch(page,/