fix(core): park writes until a model is active

Co-authored-by: Cursor <cursoragent@cursor.com>

memhub/apps/core/src/one_memory/api.py

@@ -2793,6 +2793,12 @@ def create_app(
         # D75: work queued for the previous model continues with this one.
         # Only tasks that never reached the model are re-bound.
         await asyncio.to_thread(app.state.memory_task_store.rebind_unstarted_profile, profile)
+        # A missing profile is an event-driven wait, not a provider outage.
+        # Release those durable rows only now that activation has changed the
+        # condition; the ordinary drain poll will claim them without a restart.
+        await asyncio.to_thread(
+            app.state.memory_write_queue.readmit_model_profile_waiters
+        )
         # Not awaited: the catch-up reads every conversation ever imported, and
         # the answer to "which profile is active" does not depend on it.
         _start_backfill(app)
@@ -5062,10 +5068,11 @@ def _source_writes_on(app: FastAPI, source_id: str) -> bool:
 
 
 def _sedimentation_claim_enabled(app: FastAPI, source_id: str) -> bool:
-    """Claim queued model work only while both global and source gates allow it."""
+    """Claim queued model work only when policy, source and model all admit it."""
 
     return (
         _config_store(app).sedimentation_enabled()
+        and _model_profile_store(app).active_identity() is not None
         and _history_import_gate_released(app)
         and _source_writes_on(app, source_id)
     )

memhub/apps/core/src/one_memory/memory_server/durable_queue.py

@@ -1404,11 +1404,12 @@ class DurableWriteQueue:
     ) -> int:
         """Re-admit `waiting_for_model` jobs whose wait has elapsed.
 
-        A model outage clears on its own, so this state is retried on a timer
+        A model outage clears on its own, so transient waiters retry on a timer
         rather than waiting for a grant event like `waiting_for_permission`.
-        Keyed on `updated_at` so repeated calls while a job is still parked do
-        not reset its own clock; the attempt consumed by the parking claim is
-        returned, the same way `readmit_parked` does.
+        `model_profile_not_active` is excluded: only explicit activation can
+        change that condition. Keyed on `updated_at` so repeated calls while a
+        job is still parked do not reset its own clock; the attempt consumed by
+        the parking claim is returned, the same way `readmit_parked` does.
         """
         now = now or datetime.now(UTC)
         threshold = (now - after).isoformat()
@@ -1417,8 +1418,33 @@ class DurableWriteQueue:
                 "UPDATE memory_write_jobs SET state=?, lease_owner=NULL, "
                 "lease_expires_at=NULL, attempts=CASE WHEN attempts>0 "
                 "THEN attempts-1 ELSE 0 END, last_error=NULL, updated_at=? "
-                "WHERE state=? AND updated_at <= ? AND hy_invoked=0",
-                (PENDING, now.isoformat(), WAITING_FOR_MODEL, threshold),
+                "WHERE state=? AND updated_at <= ? AND hy_invoked=0 "
+                "AND (last_error IS NULL OR last_error<>?)",
+                (
+                    PENDING,
+                    now.isoformat(),
+                    WAITING_FOR_MODEL,
+                    threshold,
+                    "model_profile_not_active",
+                ),
+            )
+        return changed.rowcount
+
+    def readmit_model_profile_waiters(self) -> int:
+        """Re-admit no-profile work only after model activation changes the condition."""
+        now = datetime.now(UTC).isoformat()
+        with main_database(self.database_path).write() as connection, connection:
+            changed = connection.execute(
+                "UPDATE memory_write_jobs SET state=?, lease_owner=NULL, "
+                "lease_expires_at=NULL, attempts=CASE WHEN attempts>0 "
+                "THEN attempts-1 ELSE 0 END, last_error=NULL, updated_at=? "
+                "WHERE state=? AND last_error=? AND hy_invoked=0",
+                (
+                    PENDING,
+                    now,
+                    WAITING_FOR_MODEL,
+                    "model_profile_not_active",
+                ),
             )
         return changed.rowcount
 
@@ -2071,7 +2097,12 @@ async def execute_job(
         # Never `logger.exception` here: it prints the exception's message,
         # which may be a provider response or the user's own text. Log the
         # already-sanitised code instead, matching `hy-failure-classification`.
-        logger.error("durable_write_job_failed job_id=%s error_code=%s", job_id, code)
+        if code == "model_profile_not_active":
+            logger.debug("durable_write_job_parked job_id=%s error_code=%s", job_id, code)
+        elif state in PARKED_STATES:
+            logger.info("durable_write_job_parked job_id=%s error_code=%s", job_id, code)
+        else:
+            logger.error("durable_write_job_failed job_id=%s error_code=%s", job_id, code)
         return
     await in_the_background(queue.settle, job_id, state=COMPLETED, owner=owner)
 

memhub/apps/core/tests/test_memory_server_durable_queue.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import asyncio
 import contextlib
 import json
+import logging
 import sqlite3
 import threading
 import time
@@ -1369,24 +1370,30 @@ def test_a_machine_error_code_is_preserved(tmp_path) -> None:
     assert safe_error_code(ValueError("some sentence with spaces")) == "ValueError"
 
 
-def test_an_inactive_profile_parks_for_model_before_any_send(tmp_path) -> None:
+def test_an_inactive_profile_parks_without_error_logging(tmp_path, caplog) -> None:
     """A pre-bridge model-not-ready code parks and keeps the intent."""
     path = tmp_path / "one-memory.sqlite3"
     queue = DurableWriteQueue(path)
     record = queue.submit(_request())
     _claim_through_drain(queue, record.job_id)
 
-    asyncio.run(
-        execute_job(
-            record.job_id,
-            queue=queue,
-            supervisor=_Recorder(error=RuntimeError("model_profile_not_active")),
-            evidence=_EvidenceStore({"e1"}),
+    with caplog.at_level(
+        logging.DEBUG,
+        logger="one_memory.memory_server.durable_queue",
+    ):
+        asyncio.run(
+            execute_job(
+                record.job_id,
+                queue=queue,
+                supervisor=_Recorder(error=RuntimeError("model_profile_not_active")),
+                evidence=_EvidenceStore({"e1"}),
+            )
         )
-    )
 
     assert _state(path, record.job_id) == WAITING_FOR_MODEL
     assert _rows(path)[0]["hy_invoked"] == 0
+    assert "durable_write_job_failed" not in caplog.text
+    assert "durable_write_job_parked" in caplog.text
 
 
 def test_an_authorization_stop_cancels_instead_of_failing(tmp_path) -> None:
@@ -1546,7 +1553,7 @@ def test_parking_does_not_burn_the_retry_budget(tmp_path) -> None:
     assert queue.get("idem-1").state == PENDING
 
 
-def test_a_model_waiter_is_readmitted_once_its_wait_elapses(tmp_path) -> None:
+def test_a_transient_model_waiter_is_readmitted_once_its_wait_elapses(tmp_path) -> None:
     """A model outage clears on its own; nothing re-admits it otherwise.
 
     `waiting_for_permission` needs a grant event, but `waiting_for_model` is
@@ -1564,7 +1571,7 @@ def test_a_model_waiter_is_readmitted_once_its_wait_elapses(tmp_path) -> None:
     queue.settle(
         record.job_id,
         state=WAITING_FOR_MODEL,
-        error_code="model_profile_not_active",
+        error_code="model_secret_unavailable",
         hy_invoked=False,
     )
 
@@ -1581,8 +1588,8 @@ def test_a_model_waiter_is_readmitted_once_its_wait_elapses(tmp_path) -> None:
     assert _rows(path)[0]["attempts"] == 0
 
 
-def test_readmitting_a_model_waiter_clears_its_stale_detail(tmp_path) -> None:
-    """The timer readmission path must clear the old blocking reason too."""
+def test_no_profile_waiter_is_readmitted_only_after_activation(tmp_path) -> None:
+    """No timer can create a model profile; activation owns this transition."""
     path = tmp_path / "one-memory.sqlite3"
     queue = DurableWriteQueue(path)
     record = queue.submit(_request())
@@ -1594,13 +1601,50 @@ def test_readmitting_a_model_waiter_clears_its_stale_detail(tmp_path) -> None:
     )
     now = datetime.now(UTC)
 
-    assert queue.readmit_model_waiters(now=now + timedelta(minutes=2)) == 1
+    assert queue.readmit_model_waiters(now=now + timedelta(minutes=2)) == 0
+    parked = queue.get("idem-1")
+    assert parked.state == WAITING_FOR_MODEL
+    assert parked.detail == "model_profile_not_active"
 
+    assert queue.readmit_model_profile_waiters() == 1
     readmitted = queue.get("idem-1")
     assert readmitted.state == PENDING
     assert readmitted.detail == ""
 
 
+def test_no_profile_backlog_stays_parked_across_maintenance_cycles(tmp_path) -> None:
+    path = tmp_path / "one-memory.sqlite3"
+    queue = DurableWriteQueue(path)
+    for index in range(100):
+        record = queue.submit(_request(key=f"no-profile-{index}"))
+        queue.settle(
+            record.job_id,
+            state=WAITING_FOR_MODEL,
+            error_code="model_profile_not_active",
+            hy_invoked=False,
+        )
+    with sqlite3.connect(path) as connection, connection:
+        connection.execute(
+            "UPDATE memory_write_jobs SET attempts=1,"
+            "updated_at='2000-01-01T00:00:00+00:00'"
+        )
+
+    first = datetime.now(UTC)
+    assert queue.readmit_model_waiters(now=first, after=timedelta(0)) == 0
+    assert (
+        queue.readmit_model_waiters(
+            now=first + timedelta(minutes=2),
+            after=timedelta(0),
+        )
+        == 0
+    )
+    with sqlite3.connect(path) as connection:
+        assert connection.execute(
+            "SELECT state,attempts,COUNT(*) FROM memory_write_jobs "
+            "GROUP BY state,attempts"
+        ).fetchall() == [(WAITING_FOR_MODEL, 1, 100)]
+
+
 def test_readmit_parked_cannot_resurrect_a_terminal_row(tmp_path) -> None:
     """Only parked states may be re-admitted, never a finished write.
 

memhub/apps/core/tests/test_model_profile_api.py

@@ -259,12 +259,17 @@ def test_activation_survives_a_corpus_already_imported(tmp_path) -> None:
                 ),
                 secrets=secrets,
             )
+            readmissions: list[bool] = []
+            app.state.memory_write_queue.readmit_model_profile_waiters = (
+                lambda: readmissions.append(True) or 0
+            )
 
             activated = await client.post(f"/api/v1/model-profiles/{profile_id}/activate")
 
             assert activated.status_code == 200
             assert activated.json()["is_active"] is True
             assert API_KEY not in activated.text
+            assert readmissions == [True]
             # Started, not waited for: the walk reads every conversation.
             assert not app.state.memory_backfill.done()
 

memhub/apps/core/tests/test_policy_api_wiring.py

@@ -111,6 +111,9 @@ def test_the_manual_source_is_admitted_by_grant_without_a_toggle(tmp_path) -> No
     app = SimpleNamespace(
         state=SimpleNamespace(
             config_store=config,
+            model_profile_store=SimpleNamespace(
+                active_identity=lambda: ("profile", 1)
+            ),
             source_grant_store=SimpleNamespace(
                 receipt=lambda **_kwargs: SimpleNamespace(revoked_at=None)
             ),
@@ -125,6 +128,21 @@ def test_the_manual_source_is_admitted_by_grant_without_a_toggle(tmp_path) -> No
     assert _sedimentation_claim_enabled(app, "one-memory-manual") is False
 
 
+def test_claim_gate_waits_without_polling_when_no_model_is_active(tmp_path) -> None:
+    config = ConfigStore(tmp_path)
+    app = SimpleNamespace(
+        state=SimpleNamespace(
+            config_store=config,
+            model_profile_store=SimpleNamespace(active_identity=lambda: None),
+            source_grant_store=SimpleNamespace(
+                receipt=lambda **_kwargs: SimpleNamespace(revoked_at=None)
+            ),
+        )
+    )
+
+    assert _sedimentation_claim_enabled(app, "one-memory-manual") is False
+
+
 def test_first_import_gate_waits_for_every_selected_history_job() -> None:
     state = SimpleNamespace(
         status="importing",