fix(DEVS-145): resolve three post-merge Codex findings on PR #78

AGENTS.md

@@ -45,8 +45,8 @@ Semantic search over agent session transcripts. Independent project, originally
 - **EmbeddingGemma padding mask (SESF-57)** — mlx-embeddings 0.0.5 hands the 0/1 tokenizer mask to SDPA as an *additive* float mask, so padding leaked into shorter docs in a batch. `rag_engine._embed_gemma` runs the gemma forward itself with a boolean key-padding mask (`_key_padding_mask`). Batched now matches single to cos ≥ 0.99999, and a batch of one (every query) is unchanged. `tests/test_embed_padding.py` pins this, so don't revert to calling `model(input_ids, attention_mask=...)`. The 512-token sliding window is still not applied on local layers (SESF-62, measure first). Stored vectors were rewritten in place by `reembed.py` (`cleanup.py reembed {plan|run|drive|status|verify}`; chunked under the MLX crash limit; old vectors backed up to `~/.sessionflow/audit/reembed-<run>.f16`).
 - **Memory store is Postgres (DEVS-145)** — memories commit to Postgres first; the Milvus `memories` collection and `~/.sessionflow/memories_fts.db` are derived indexes. The DSN comes from `SESSIONFLOW_MEMORY_DSN` only: SessionFlow calls no secrets tool and reads no DSN file, so a launchd install sets it in the LaunchAgent environment. Memory starts after transcript startup on its own thread; a missing DSN or a down database leaves `/health` `memory.ready` false with the reason and transcript search untouched. Blocking store calls go through `memory_store.run_db` (the `memory-db` executor), never the event loop or `mlx-embed`.
 - **Never write a memory index directly (DEVS-145)** — every write goes through `MemoryService` to Postgres, which resets that memory's rows in `memory_index_acks` in the same transaction. The sync loop (`SESSIONFLOW_MEMORY_SYNC_INTERVAL`, plus a full reconcile at start and every 6 hours) applies what each consumer (`milvus`, `fts:<host>`) lacks. A Milvus write re-reads the row under `pg_advisory_xact_lock` and upserts only if the embedded hash still matches; the FTS file keeps a per-record watermark so an older write cannot revive a deleted record. `change_seq` orders work but does not follow commit order, so pending work is decided by acknowledgements, not by a cursor.
-- **Memory embeds share the MLX thread (DEVS-145)** — `memory_index.embed_for_index` waits on `embedding_control.EmbedThrottle` (200 ms floor) and runs on `mlx-embed`; it never uses the transcript budget, which drops batches. `memory_cli` never embeds: `rebuild` and `reindex` only reset acknowledgements, and `restore-check` asks the running server's loopback `/memory/restore-check` route to index into a temporary `memories_restore` collection.
-- **Memory embedding generation (DEVS-145)** — `memory_index_meta` names the memory model and generation. `cleanup.py reembed` starts a new generation on a fresh run and resets the Milvus acknowledgements; the server re-embeds from Postgres text. A server whose local model differs refuses memory vector writes and reports it in `/health` (`memory.index.milvus_blocked`).
+- **Memory embeds share the MLX thread (DEVS-145)** — `memory_index.embed_for_index` waits on `embedding_control.EmbedThrottle` (200 ms floor) and runs on `mlx-embed`, where it waits again and stamps the throttle; it never uses the transcript budget, which drops batches. `EmbeddingBudget.before_batch` honors the same throttle, so neither consumer starts inside the other's cooldown. `memory_cli` never embeds: `rebuild` and `reindex` only reset acknowledgements, and `restore-check` asks the running server's loopback `/memory/restore-check` route to index into a temporary `memories_restore` collection.
+- **Memory embedding generation (DEVS-145)** — `memory_index_meta` names the memory model and generation. `cleanup.py reembed` starts a new generation on a fresh run and resets the Milvus acknowledgements; the server re-embeds from Postgres text. A server whose local model differs refuses memory vector writes and vector searches (`MemoryModelMismatch`; search answers from FTS) and reports it in `/health` (`memory.index.milvus_blocked`).
 - **Memory secret check (DEVS-145)** — `add_memory` rejects text with any `secret_redaction.scan_spans` hit and names the rule. Paths and code identifiers can trip it; the fix is a narrow pattern in the operator allowlist (`SESSIONFLOW_REDACT_ALLOWLIST`), never a looser rule.
 - **Transcript cleanup leaves memory alone (DEVS-145)** — age expiry, `cleanup_sessions`, `sanitize --drop`, `reset` and `migrate-schema` act on `sessions` and `fts.db` only; `reset` and `migrate-schema` say so.
 - **Memory tests (DEVS-145)** — Postgres tests need `SESSIONFLOW_TEST_PG_DSN` (one fresh schema per test through the `memory_pg` fixture) and skip with that reason when it is unset; restore-check tests also need `SESSIONFLOW_TEST_PG_RESTORE_DSN` and `pg_dump`/`pg_restore` 17. An autouse fixture clears `SESSIONFLOW_MEMORY_DSN`. `stub_rag_engine` evicts project modules first imported under the stub, so a later test imports them against the real `rag_engine`. pytest prints fixture DSNs in setup errors, so mask `password=` in any shared output.

CLAUDE.md

@@ -46,8 +46,8 @@ Semantic search over Claude Code session transcripts. Independent project, origi
 - **EmbeddingGemma padding mask (SESF-57)** — mlx-embeddings 0.0.5 hands the 0/1 tokenizer mask to SDPA as an *additive* float mask, so padding leaked into shorter docs in a batch. `rag_engine._embed_gemma` runs the gemma forward itself with a boolean key-padding mask (`_key_padding_mask`). Batched now matches single to cos ≥ 0.99999, and a batch of one (every query) is unchanged. `tests/test_embed_padding.py` pins this, so don't revert to calling `model(input_ids, attention_mask=...)`. The 512-token sliding window is still not applied on local layers (SESF-62, measure first). Stored vectors were rewritten in place by `reembed.py` (`cleanup.py reembed {plan|run|drive|status|verify}`; chunked under the MLX crash limit; old vectors backed up to `~/.sessionflow/audit/reembed-<run>.f16`).
 - **Memory store is Postgres (DEVS-145)** — memories commit to Postgres first; the Milvus `memories` collection and `~/.sessionflow/memories_fts.db` are derived indexes. The DSN comes from `SESSIONFLOW_MEMORY_DSN` only: SessionFlow calls no secrets tool and reads no DSN file, so a launchd install sets it in the LaunchAgent environment. Memory starts after transcript startup on its own thread; a missing DSN or a down database leaves `/health` `memory.ready` false with the reason and transcript search untouched. Blocking store calls go through `memory_store.run_db` (the `memory-db` executor), never the event loop or `mlx-embed`.
 - **Never write a memory index directly (DEVS-145)** — every write goes through `MemoryService` to Postgres, which resets that memory's rows in `memory_index_acks` in the same transaction. The sync loop (`SESSIONFLOW_MEMORY_SYNC_INTERVAL`, plus a full reconcile at start and every 6 hours) applies what each consumer (`milvus`, `fts:<host>`) lacks. A Milvus write re-reads the row under `pg_advisory_xact_lock` and upserts only if the embedded hash still matches; the FTS file keeps a per-record watermark so an older write cannot revive a deleted record. `change_seq` orders work but does not follow commit order, so pending work is decided by acknowledgements, not by a cursor.
-- **Memory embeds share the MLX thread (DEVS-145)** — `memory_index.embed_for_index` waits on `embedding_control.EmbedThrottle` (200 ms floor) and runs on `mlx-embed`; it never uses the transcript budget, which drops batches. `memory_cli` never embeds: `rebuild` and `reindex` only reset acknowledgements, and `restore-check` asks the running server's loopback `/memory/restore-check` route to index into a temporary `memories_restore` collection.
-- **Memory embedding generation (DEVS-145)** — `memory_index_meta` names the memory model and generation. `cleanup.py reembed` starts a new generation on a fresh run and resets the Milvus acknowledgements; the server re-embeds from Postgres text. A server whose local model differs refuses memory vector writes and reports it in `/health` (`memory.index.milvus_blocked`).
+- **Memory embeds share the MLX thread (DEVS-145)** — `memory_index.embed_for_index` waits on `embedding_control.EmbedThrottle` (200 ms floor) and runs on `mlx-embed`, where it waits again and stamps the throttle; it never uses the transcript budget, which drops batches. `EmbeddingBudget.before_batch` honors the same throttle, so neither consumer starts inside the other's cooldown. `memory_cli` never embeds: `rebuild` and `reindex` only reset acknowledgements, and `restore-check` asks the running server's loopback `/memory/restore-check` route to index into a temporary `memories_restore` collection.
+- **Memory embedding generation (DEVS-145)** — `memory_index_meta` names the memory model and generation. `cleanup.py reembed` starts a new generation on a fresh run and resets the Milvus acknowledgements; the server re-embeds from Postgres text. A server whose local model differs refuses memory vector writes and vector searches (`MemoryModelMismatch`; search answers from FTS) and reports it in `/health` (`memory.index.milvus_blocked`).
 - **Memory secret check (DEVS-145)** — `add_memory` rejects text with any `secret_redaction.scan_spans` hit and names the rule. Paths and code identifiers can trip it; the fix is a narrow pattern in the operator allowlist (`SESSIONFLOW_REDACT_ALLOWLIST`), never a looser rule.
 - **Transcript cleanup leaves memory alone (DEVS-145)** — age expiry, `cleanup_sessions`, `sanitize --drop`, `reset` and `migrate-schema` act on `sessions` and `fts.db` only; `reset` and `migrate-schema` say so.
 - **Memory tests (DEVS-145)** — Postgres tests need `SESSIONFLOW_TEST_PG_DSN` (one fresh schema per test through the `memory_pg` fixture) and skip with that reason when it is unset; restore-check tests also need `SESSIONFLOW_TEST_PG_RESTORE_DSN` and `pg_dump`/`pg_restore` 17. An autouse fixture clears `SESSIONFLOW_MEMORY_DSN`. `stub_rag_engine` evicts project modules first imported under the stub, so a later test imports them against the real `rag_engine`. pytest prints fixture DSNs in setup errors, so mask `password=` in any shared output.

README.md

@@ -112,7 +112,7 @@ SessionFlow also stores memories: notes an agent writes on purpose, such as sess
 `setup.sh` copies `hooks/memory-session-start.py` into the Claude Code and Codex hook folders. At session start it shows the three newest session summaries and recent turns, and falls back to a local cache when the server cannot answer. Setup registers the hook for Codex and prints the Claude Code `~/.claude/settings.json` SessionStart entry for you to add by hand.
 
 ```bash
-python memory_cli.py audit                        # Likely duplicates, aged open items, counts
+python memory_cli.py audit                        # Likely duplicates, aged open items, counts, unknown projects
 python memory_cli.py rebuild --yes                # Re-index every memory from Postgres
 python memory_cli.py import export.json --report report.json
 python memory_cli.py restore-check memory.dump --target-dsn "$EMPTY_DSN"

embedding_control.py

@@ -117,32 +117,35 @@ class EmbedThrottle:
     """Cooldown between embed batches that waits instead of denying work.
 
     Memory index embeds use it in place of the transcript budget, which drops
-    batches. Transcript batches stamp it too, so a memory embed never starts
-    inside a transcript batch's cooldown. The cooldown never goes below 200 ms.
+    batches. Both consumers stamp it and honor it, so neither starts inside the
+    other's cooldown. The cooldown never goes below 200 ms.
     """
 
     def __init__(
         self,
         cooldown_ms: int = MIN_COOLDOWN_MS,
         clock: Callable[[], float] = time.monotonic,
         sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
+        blocking_sleep: Callable[[float], None] = time.sleep,
     ) -> None:
-        """Set the cooldown; ``clock`` and ``sleep`` are injectable for tests."""
+        """Set the cooldown; ``clock`` and both sleeps are injectable for tests."""
         self.cooldown_ms = max(int(cooldown_ms), MIN_COOLDOWN_MS)
         self._clock = clock
         self._sleep = sleep
+        self._blocking_sleep = blocking_sleep
         self._last_done = 0.0
         # An asyncio.Lock binds to one event loop, so each loop gets its own.
         self._locks: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Lock]" = (
             weakref.WeakKeyDictionary()
         )
 
     @asynccontextmanager
-    async def batch(self) -> AsyncIterator[None]:
+    async def batch(self, stamp: bool = True) -> AsyncIterator[None]:
         """Hold the throttle for one batch: wait out the cooldown, run, then stamp it.
 
         Concurrent batches on the same event loop run one at a time, so each
-        starts a full cooldown after the previous one finished.
+        starts a full cooldown after the previous one finished. Pass
+        ``stamp=False`` when the batch stamps the throttle itself.
         """
         loop = asyncio.get_running_loop()
         lock = self._locks.get(loop)
@@ -153,16 +156,31 @@ async def batch(self) -> AsyncIterator[None]:
             try:
                 yield
             finally:
-                self.done()
+                if stamp:
+                    self.done()
+
+    def remaining(self) -> float:
+        """Return the seconds left in the cooldown after the last finished batch."""
+        if not self._last_done:
+            return 0.0
+        return max(0.0, self.cooldown_ms / 1000.0 - (self._clock() - self._last_done))
 
     async def wait(self) -> None:
         """Sleep until the cooldown after the last finished batch has passed."""
-        if not self._last_done:
-            return
-        remaining = self.cooldown_ms / 1000.0 - (self._clock() - self._last_done)
+        remaining = self.remaining()
         if remaining > 0:
             await self._sleep(remaining)
 
+    def wait_blocking(self) -> None:
+        """Block the calling thread until the cooldown has passed.
+
+        Runs on the ``mlx-embed`` thread just before an embed, so work queued
+        behind another consumer's batch still starts a full cooldown after it.
+        """
+        remaining = self.remaining()
+        if remaining > 0:
+            self._blocking_sleep(remaining)
+
     def done(self) -> None:
         """Record that a batch finished now."""
         self._last_done = self._clock()
@@ -227,13 +245,15 @@ def before_batch(self, batch_size: int, estimated_chars: int = 0) -> EmbeddingDe
                 f"Batch size {batch_size} exceeds configured limit {self.batch_size}",
             )
 
-        elapsed = time.monotonic() - self.last_batch_finished_at
-        cooldown = self.cooldown_ms / 1000.0
-        if self.last_batch_finished_at and elapsed < cooldown:
+        remaining = get_embed_throttle().remaining()
+        if self.last_batch_finished_at:
+            elapsed = time.monotonic() - self.last_batch_finished_at
+            remaining = max(remaining, self.cooldown_ms / 1000.0 - elapsed)
+        if remaining > 0:
             return EmbeddingDecision(
                 False,
                 "Embedding cooldown active",
-                retry_after_seconds=max(0.0, cooldown - elapsed),
+                retry_after_seconds=remaining,
             )
 
         return EmbeddingDecision(True)

memory_cli.py

@@ -115,6 +115,7 @@ async def no_query_embedding(text: str) -> list:
     return memory_service.MemoryService(
         store=_open_store(), index=_DeferredIndex(), embed_query=no_query_embedding,
         allowlist=rag_engine.load_allowlist(allowlist_path),
+        project_inventory=memory_service.project_inventory(),
     )
 
 
@@ -159,6 +160,10 @@ def cmd_audit(args: argparse.Namespace) -> int:
     if index is None or index.unavailable:
         _err("Similar-memory clusters are unavailable: Milvus could not be read. "
              "Stale items and counts are still reported.")
+    if report.get("inventory_checked") is False:
+        _err("Unknown-project check skipped: no known project root has a readable "
+             ".claude/project.json. Open one session in each project so its "
+             "SessionStart hook registers the root, then run the audit again.")
     if args.format == "json":
         print(_json(report))
         return 0

memory_index.py

@@ -38,6 +38,10 @@
                 "type", "status", "issue_ids", "source", "category", "created_at")
 
 
+class MemoryModelMismatch(RuntimeError):
+    """This machine's embedding model is not the memory model, so its vectors cannot be compared."""
+
+
 def _iso(value: Any) -> str:
     if isinstance(value, datetime):
         return value.astimezone(timezone.utc).isoformat()
@@ -62,11 +66,23 @@ async def embed_for_index(texts: List[str]) -> List[List[float]]:
     throttle = embedding_control.get_embed_throttle()
     batch = getattr(throttle, "batch", None)
     if batch is not None:
-        async with batch():
+        # _embed_after_cooldown stamps on mlx-embed; a second stamp here would
+        # push back a transcript batch already waiting out the first one.
+        async with batch(stamp=False):
             return await _embed_on_mlx_thread(texts)
     await throttle.wait()
+    return await _embed_on_mlx_thread(texts)
+
+
+def _embed_after_cooldown(texts: List[str]) -> List[List[float]]:
+    # Runs on mlx-embed. A transcript batch can finish between the async wait
+    # and this call, and the next one checks the stamp on this thread.
+    throttle = embedding_control.get_embed_throttle()
+    wait_blocking = getattr(throttle, "wait_blocking", None)
+    if wait_blocking is not None:
+        wait_blocking()
     try:
-        return await _embed_on_mlx_thread(texts)
+        return rag_engine.embed_texts(texts)
     finally:
         throttle.done()
 
@@ -75,8 +91,8 @@ async def _embed_on_mlx_thread(texts: List[str]) -> List[List[float]]:
     loop = asyncio.get_running_loop()
     executor = rag_engine._embed_executor
     if executor is not None:
-        return await loop.run_in_executor(executor, rag_engine.embed_texts, texts)
-    return await asyncio.to_thread(rag_engine.embed_texts, texts)
+        return await loop.run_in_executor(executor, _embed_after_cooldown, texts)
+    return await asyncio.to_thread(_embed_after_cooldown, texts)
 
 
 class MemoryFTS:
@@ -463,7 +479,14 @@ def _vector_search_sync(self, vector: List[float], filters: Dict[str, Any],
 
     async def vector_search(self, vector: List[float], *, filters: Dict[str, Any],
                             limit: int) -> List[Dict[str, Any]]:
-        """Return Milvus hits (``memory_id``, ``version``, ``generation``, ``score``), best first."""
+        """Return Milvus hits (``memory_id``, ``version``, ``generation``, ``score``), best first.
+
+        Raises ``MemoryModelMismatch`` when the stored memory model differs from
+        this machine's: same-size vectors from another model would still match.
+        """
+        await self._refresh_model_block()
+        if self.milvus_blocked:
+            raise MemoryModelMismatch(self.milvus_blocked)
         return await run_db(self._vector_search_sync, vector, filters, limit)
 
     async def fts_search(self, query: str, *, filters: Dict[str, Any],