fix(memhub): retry unknown writes after ten seconds

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

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

@@ -1049,10 +1049,15 @@ def create_app(
                         job.job_id,
                     )
                 await asyncio.to_thread(
-                    memory_write_queue.retry_timed_out_unknown_after_reconcile,
-                    older_than=datetime.now(UTC) - timedelta(minutes=5),
+                    memory_write_queue.retry_first_unknown_after_reconcile,
+                    older_than=datetime.now(UTC) - timedelta(seconds=10),
                     limit=32,
                 )
+                await asyncio.to_thread(
+                    app.state.memory_task_store.retry_first_unknown_after_reconcile,
+                    older_than=datetime.now(UTC) - timedelta(seconds=10),
+                    limit=128,
+                )
 
             async def admit_hy_invocation(
                 source_id: str,

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

@@ -306,7 +306,7 @@ LEGACY_MEMORY_WRITE_JOB_COLUMNS = (
     ("last_error", "TEXT"),
 )
 
-_FIFO_BLOCKING_STATES = (PENDING, RUNNING, WAITING_FOR_PERMISSION, WAITING_FOR_MODEL, UNKNOWN)
+_FIFO_BLOCKING_STATES = (PENDING, RUNNING, WAITING_FOR_PERMISSION, WAITING_FOR_MODEL)
 
 
 @dataclass(frozen=True)
@@ -1492,6 +1492,59 @@ class DurableWriteQueue:
                 )
         return len(job_ids)
 
+    def retry_first_unknown_after_reconcile(
+        self,
+        *,
+        older_than: datetime,
+        limit: int = 32,
+    ) -> int:
+        """Retry one unresolved unknown outcome once, after reconciliation.
+
+        The caller first asks HY and checks linked governed output. Rows still
+        unknown may have lost the response after the SDK call, so one bounded
+        retry is useful. `attempts=1` is the guard: the retry claim increments
+        it to two, and a second unknown is left for explicit user recovery.
+        """
+        if type(limit) is not int or limit < 1:
+            raise ValueError("limit must be positive")
+        now = datetime.now(UTC).isoformat()
+        with (
+            main_database(self.database_path).write(row_factory=sqlite3.Row) as connection,
+            connection,
+        ):
+            rows = connection.execute(
+                "SELECT job_id FROM memory_write_jobs "
+                "WHERE state=? AND attempts=1 AND attempts < max_attempts "
+                "AND updated_at <= ? ORDER BY updated_at, job_id LIMIT ?",
+                (UNKNOWN, older_than.isoformat(), limit),
+            ).fetchall()
+            job_ids = tuple(str(row["job_id"]) for row in rows)
+            if not job_ids:
+                return 0
+            placeholders = ",".join("?" for _ in job_ids)
+            connection.execute(
+                f"UPDATE memory_write_jobs SET state=?, lease_owner=NULL, "
+                f"lease_expires_at=NULL, hy_invoked=0, "
+                f"last_error='model_content_unknown_auto_retry', updated_at=? "
+                f"WHERE state=? AND attempts=1 AND job_id IN ({placeholders})",
+                (PENDING, now, UNKNOWN, *job_ids),
+            )
+            if connection.execute(
+                "SELECT 1 FROM sqlite_master WHERE type='table' AND name='memory_tasks'"
+            ).fetchone():
+                connection.execute(
+                    f"UPDATE memory_tasks SET status=?, runner_token=NULL, "
+                    f"last_error_code='model_content_unknown_auto_retry', updated_at=? "
+                    f"WHERE status=? AND task_id IN ("
+                    f"SELECT inputs.task_id FROM memory_task_inputs AS inputs "
+                    f"JOIN memory_write_job_evidence AS linked ON EXISTS ("
+                    f"SELECT 1 FROM json_each(inputs.refs_json) AS ref "
+                    f"WHERE json_extract(ref.value, '$.evidence_id')=linked.evidence_id) "
+                    f"WHERE linked.job_id IN ({placeholders}))",
+                    (MemoryTaskStatus.PENDING, now, MemoryTaskStatus.UNKNOWN, *job_ids),
+                )
+        return len(job_ids)
+
     def resolve_unknown_from_linked_output(self, job_id: str) -> bool:
         """Close an unknown durable row when local governance output proves success."""
         with main_database(self.database_path).read(row_factory=sqlite3.Row) as connection:
@@ -1896,6 +1949,13 @@ class DurableWriteDrain:
         #: Reaping is a write; it runs at most once per this interval on the
         #: idle path so an empty queue does not issue one write per poll.
         self._reap_interval = reap_interval
+        # Unknown recovery is user-visible and lightweight enough to check
+        # sooner than general database housekeeping. Tests that set the common
+        # interval to zero keep immediate maintenance.
+        self._unknown_reconcile_interval = min(
+            reap_interval,
+            timedelta(seconds=10),
+        )
         #: The maintenance writes below are recoverability/retention work, not
         #: idle housekeeping, so each has its own interval and runs at the start
         #: of every drain tick. Sharing one timestamp would let a busy tick
@@ -2308,7 +2368,8 @@ class DurableWriteDrain:
                     )
                 if self.reconcile_unknown is not None and (
                     self._last_unknown_reconcile is None
-                    or tick_started - self._last_unknown_reconcile >= self._reap_interval
+                    or tick_started - self._last_unknown_reconcile
+                    >= self._unknown_reconcile_interval
                 ):
                     self._last_unknown_reconcile = tick_started
                     try:

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

@@ -2161,6 +2161,43 @@ class MemoryTaskStore:
             )
         return changed.rowcount
 
+    def retry_first_unknown_after_reconcile(
+        self,
+        *,
+        older_than: datetime,
+        limit: int = 128,
+    ) -> int:
+        """Automatically retry a task-only unknown exactly once.
+
+        Reconciliation runs before this transition. Attempts are preserved, so
+        the retry claim moves `attempts=1` to two and another unknown is left
+        for explicit user recovery.
+        """
+        if type(limit) is not int or limit < 1:
+            raise ValueError("limit must be positive")
+        now = datetime.now(UTC).isoformat()
+        with self._connect(writing=True) as connection:
+            changed = connection.execute(
+                """
+                UPDATE memory_tasks
+                SET status=?, runner_token=NULL,
+                    last_error_code='model_content_unknown_auto_retry', updated_at=?
+                WHERE task_id IN (
+                    SELECT task_id FROM memory_tasks
+                    WHERE status=? AND attempts=1 AND updated_at <= ?
+                    ORDER BY updated_at, task_id LIMIT ?
+                )
+                """,
+                (
+                    MemoryTaskStatus.PENDING,
+                    now,
+                    MemoryTaskStatus.UNKNOWN,
+                    older_than.isoformat(),
+                    limit,
+                ),
+            )
+        return changed.rowcount
+
     def recover_running(self) -> int:
         """Return tasks orphaned by a previous process to the pending queue."""
         return self.recover_orphan_running()

memhub/apps/core/tests/test_memory_server_durable_queue.py

@@ -1097,6 +1097,52 @@ def test_reconciled_absent_timeout_is_requeued_once_without_resetting_attempts(
     )
 
 
+def test_unresolved_unknown_is_automatically_retried_only_once(tmp_path) -> None:
+    path = tmp_path / "one-memory.sqlite3"
+    queue = DurableWriteQueue(path)
+    record = queue.submit(_request())
+    with sqlite3.connect(path) as connection, connection:
+        connection.execute(
+            "UPDATE memory_write_jobs SET state='unknown',attempts=1,hy_invoked=1,"
+            "last_error='model_content_outcome_unknown',updated_at=? WHERE job_id=?",
+            ("2026-01-01T00:00:00+00:00", record.job_id),
+        )
+
+    cutoff = datetime(2026, 1, 1, 0, 5, tzinfo=UTC)
+    assert queue.retry_first_unknown_after_reconcile(older_than=cutoff) == 1
+    first_retry = queue.get(record.request_id)
+    assert first_retry.state == PENDING
+    with sqlite3.connect(path) as connection:
+        assert connection.execute(
+            "SELECT attempts FROM memory_write_jobs WHERE job_id=?",
+            (record.job_id,),
+        ).fetchone()[0] == 1
+
+    drain = DurableWriteDrain(queue, execute=_never_execute, owner="retry-owner")
+    assert drain.claim(limit=1) == (record.job_id,)
+    assert queue.mark_invoked(record.job_id, owner="retry-owner")
+    queue.settle(
+        record.job_id,
+        state=UNKNOWN,
+        owner="retry-owner",
+        error_code="model_content_outcome_unknown",
+        hy_invoked=True,
+    )
+
+    with sqlite3.connect(path) as connection:
+        assert connection.execute(
+            "SELECT attempts FROM memory_write_jobs WHERE job_id=?",
+            (record.job_id,),
+        ).fetchone()[0] == 2
+    assert (
+        queue.retry_first_unknown_after_reconcile(
+            older_than=datetime.now(UTC) + timedelta(minutes=1)
+        )
+        == 0
+    )
+    assert queue.get(record.request_id).state == UNKNOWN
+
+
 def test_drain_runs_unknown_reconciliation_during_busy_safe_maintenance(tmp_path) -> None:
     queue = DurableWriteQueue(tmp_path / "one-memory.sqlite3")
     calls = 0
@@ -1116,6 +1162,20 @@ def test_drain_runs_unknown_reconciliation_during_busy_safe_maintenance(tmp_path
     assert calls == 1
 
 
+def test_unknown_reconciliation_runs_on_a_ten_second_ceiling(tmp_path) -> None:
+    queue = DurableWriteQueue(tmp_path / "one-memory.sqlite3")
+
+    production = DurableWriteDrain(queue, execute=_never_execute)
+    immediate = DurableWriteDrain(
+        queue,
+        execute=_never_execute,
+        reap_interval=timedelta(0),
+    )
+
+    assert production._unknown_reconcile_interval == timedelta(seconds=10)
+    assert immediate._unknown_reconcile_interval == timedelta(0)
+
+
 def test_a_reclaimed_uninvoked_job_with_attempts_left_is_retryable(tmp_path) -> None:
     """Recovery of a never-invoked lease is the retry path, not `unknown`.
 
@@ -2063,6 +2123,19 @@ def test_sqlite_claim_keeps_fifo_order_when_later_work_is_live(tmp_path) -> None
     assert drain.claim(limit=1) == (normal.job_id,)
 
 
+def test_unknown_source_head_does_not_block_later_pending_work(tmp_path) -> None:
+    path = tmp_path / "one-memory.sqlite3"
+    queue = DurableWriteQueue(path)
+    uncertain = queue.submit(_request(evidence=("e1",), key="unknown"))
+    later = queue.submit(_request(evidence=("e2",), key="later"))
+    queue.settle(uncertain.job_id, state=UNKNOWN, hy_invoked=True)
+
+    drain = DurableWriteDrain(queue, execute=_never_execute, owner="next-owner")
+
+    assert drain.claim(limit=1) == (later.job_id,)
+    assert queue.get("unknown").state == UNKNOWN
+
+
 def test_sqlite_claim_orders_different_sources_by_enqueue_sequence(tmp_path) -> None:
     path = tmp_path / "one-memory.sqlite3"
     queue = DurableWriteQueue(path)

memhub/apps/core/tests/test_memory_tasks.py

@@ -1070,6 +1070,49 @@ def test_confirmed_retry_requeues_task_only_unknown_without_a_durable_job(tmp_pa
     assert store.retry_unknown_confirmed() == 0
 
 
+def test_task_only_unknown_is_automatically_retried_only_once(tmp_path) -> None:
+    store, profile = _profile_and_tasks(tmp_path)
+    task = store.create(
+        request_id="auto-retry-unknown",
+        user_id="local",
+        kind=MemoryTaskKind.EXTRACT,
+        input_ref="orphan-evidence",
+        profile=profile,
+    )
+    with sqlite3.connect(store.database_path) as connection, connection:
+        connection.execute(
+            "UPDATE memory_tasks SET status=?,attempts=1,last_error_code=?,updated_at=? "
+            "WHERE task_id=?",
+            (
+                MemoryTaskStatus.UNKNOWN,
+                "model_content_outcome_unknown",
+                "2026-01-01T00:00:00+00:00",
+                task.task_id,
+            ),
+        )
+
+    cutoff = datetime(2026, 1, 1, 0, 5, tzinfo=UTC)
+    assert store.retry_first_unknown_after_reconcile(older_than=cutoff) == 1
+    retried = store.require(task.task_id)
+    assert retried.status is MemoryTaskStatus.PENDING
+    assert retried.attempts == 1
+
+    running, token = store.claim(task.task_id)
+    assert running.attempts == 2
+    store.mark_unknown(
+        task.task_id,
+        runner_token=token,
+        error_code="model_content_outcome_unknown",
+    )
+    assert (
+        store.retry_first_unknown_after_reconcile(
+            older_than=datetime.now(UTC) + timedelta(minutes=1)
+        )
+        == 0
+    )
+    assert store.require(task.task_id).status is MemoryTaskStatus.UNKNOWN
+
+
 def test_orphan_recovery_preserves_active_running_tasks(tmp_path) -> None:
     store, profile = _profile_and_tasks(tmp_path)
     active = store.create(