fix(memory): keep same-timestamp rows in order and never skip them in upkeep

CHANGELOG.md

@@ -201,6 +201,9 @@ and versions follow [Semantic Versioning](https://semver.org/).
 - A task that repeats the same failing step three times now pauses as stuck and asks what to do, instead of failing.
 
 ### Fixed
+- Messages and memories saved at the same moment now always keep their order, and the daily upkeep of MEMORY.md and
+  USER.md no longer misses a memory saved while it was reading. Summarizing a long chat no longer skips messages
+  saved at the same moment as the last summary.
 - When a cloud model says no, chat and tasks now say why in a plain sentence and what to do, instead of showing the
   provider's raw reply. For example: "OpenRouter's free models have reached today's limit. Add credits on
   openrouter.ai, wait until tomorrow, or pick another model in Settings > Models." This covers rate limits, an

sentient/agent/loop.py

@@ -79,7 +79,7 @@
 from sentient.memory.workspace import Workspace
 from sentient.services import cancel_tasks
 from sentient.skills.loader import SkillLibrary
-from sentient.store.db import Store, new_id
+from sentient.store.db import Store, after_watermark, new_id, next_watermark
 from sentient.tools.base import (
     UNPROMPTED_ORIGINS,
     Risk,
@@ -1368,7 +1368,7 @@ async def _chat_untrusted(self, session_id: str, session: dict | None) -> str:
         if stored is not None:
             return str(stored)
         rows = await self.store.fetchall(
-            "SELECT role, name, content FROM messages WHERE session_id = ? AND role = 'tool' ORDER BY created_at",
+            "SELECT role, name, content FROM messages WHERE session_id = ? AND role = 'tool' ORDER BY created_at, rowid",
             (session_id,),
         )
         source = untrusted_in([dict(r) for r in rows], self.registry)
@@ -1529,10 +1529,11 @@ async def _maybe_compress(self, session_id: str, *, flush: bool = True) -> None:
                 return
             session = await self.store.get_session(session_id)
             upto = (session or {}).get("context_upto") or ""
+            after, params = after_watermark(upto, "created_at", "rowid")  # same-tick messages are not skipped (#236)
             rows = await self.store.fetchall(
-                "SELECT role, content, created_at FROM messages WHERE session_id = ? AND created_at > ?"
-                " AND role IN ('user','assistant') AND content IS NOT NULL ORDER BY created_at",
-                (session_id, upto),
+                f"SELECT rowid, role, content, created_at FROM messages WHERE session_id = ? AND {after}"
+                " AND role IN ('user','assistant') AND content IS NOT NULL ORDER BY created_at, rowid",
+                (session_id, *params),
             )
             older = rows[: max(0, len(rows) - window)]
             if len(older) < 10:
@@ -1550,7 +1551,7 @@ async def _maybe_compress(self, session_id: str, *, flush: bool = True) -> None:
             )
             await self.store.execute(
                 "UPDATE sessions SET context_summary = ?, context_upto = ? WHERE id = ?",
-                (summary.strip(), older[-1]["created_at"], session_id),
+                (summary.strip(), next_watermark(upto, older, "created_at", "rowid"), session_id),
             )
         except Exception as exc:
             log.debug("context compression skipped: %s", exc)

sentient/evolution/service.py

@@ -36,7 +36,7 @@
 from sentient.memory.vectors import cosine
 from sentient.services import Service, cancel_tasks
 from sentient.skills.loader import slugify, valid_name
-from sentient.store.db import now_iso
+from sentient.store.db import after_watermark, next_watermark, now_iso
 
 log = logging.getLogger(__name__)
 
@@ -48,6 +48,8 @@
 FAILED_RUN_STATUSES = {"error", "failed", "completed_with_errors"}
 REPAIR_TRANSCRIPT_BUDGET = 6_000
 MAX_TRACKED_TURNS = 200
+PROFILE_FACTS_MARK = "evolution.profile_facts_mark"  # what profile upkeep read last (store.db.next_watermark)
+PROFILE_SUMMARIES_MARK = "evolution.profile_summaries_mark"
 
 # cheap first pass before one fast-model confirmation; false positives are fine, misses are not
 _CORRECTION_RE = re.compile(
@@ -825,18 +827,30 @@ async def update_profile(self, now: datetime | None = None, *, force: bool = Fal
             return {"updated": False}
         since = (last or now - timedelta(days=30)).isoformat()
         nowiso = now.isoformat()
+        # what upkeep read last time: rows saved in the same clock tick as that read are still picked up (#236)
+        facts_mark = await store.get_meta(PROFILE_FACTS_MARK) or since
+        summaries_mark = await store.get_meta(PROFILE_SUMMARIES_MARK) or since
+        after, params = after_watermark(facts_mark, "updated_at", "id")
         new_facts = await store.fetchall(
-            "SELECT content, topics, source FROM facts WHERE status = 'active' AND memory_type = 'long-term' AND updated_at > ?"
-            " AND (expires_at IS NULL OR expires_at > ?) ORDER BY updated_at",
-            (since, nowiso),
+            "SELECT id, content, topics, source, updated_at FROM facts WHERE status = 'active'"
+            f" AND memory_type = 'long-term' AND {after} AND (expires_at IS NULL OR expires_at > ?)"
+            " ORDER BY updated_at, id",
+            (*params, nowiso),
         )
         # chats that read outside content never feed MEMORY.md (ADR 0021): their facts wait for review and their
         # summaries are left out
+        after, params = after_watermark(summaries_mark, "created_at", "id")
         new_summaries = await store.fetchall(
-            f"SELECT content FROM summaries WHERE created_at > ? AND {CLEAN_SUMMARY} ORDER BY end_at DESC LIMIT 8",
-            (since,),
+            f"SELECT id, content, created_at FROM summaries WHERE {after} AND {CLEAN_SUMMARY}"
+            " ORDER BY created_at, id LIMIT 8",  # in watermark order, so the limit never jumps over a summary
+            params,
         )
         await store.set_meta("evolution.profile_last_run", nowiso)
+        # saved after every read, also an empty one: the next read starts where this one did, never at this run's time
+        await store.set_meta(PROFILE_FACTS_MARK, next_watermark(facts_mark, new_facts, "updated_at", "id"))
+        await store.set_meta(
+            PROFILE_SUMMARIES_MARK, next_watermark(summaries_mark, new_summaries, "created_at", "id")
+        )
         if not new_facts and not new_summaries:
             return {"updated": False}
 
@@ -849,11 +863,11 @@ async def update_profile(self, now: datetime | None = None, *, force: bool = Fal
         budget = cfg.memory.workspace_budget_chars
         all_facts = await store.fetchall(
             "SELECT content FROM facts WHERE status = 'active' AND (expires_at IS NULL OR expires_at > ?)"
-            " ORDER BY updated_at DESC LIMIT 80",
+            " ORDER BY updated_at DESC, id DESC LIMIT 80",
             (nowiso,),
         )
         summaries = await store.fetchall(
-            f"SELECT content FROM summaries WHERE {CLEAN_SUMMARY} ORDER BY end_at DESC LIMIT 8"
+            f"SELECT content FROM summaries WHERE {CLEAN_SUMMARY} ORDER BY end_at DESC, rowid DESC LIMIT 8"
         )
         current = app.workspace.read_full()["memory"]
         memory_chars = 0

sentient/memory/episodic.py

@@ -216,7 +216,7 @@ async def messages_between(self, start: str, end: str, limit: int = 200) -> list
             " LEFT JOIN sessions s ON s.id = m.session_id"
             " WHERE m.created_at >= ? AND m.created_at <= ? AND m.role IN ('user', 'assistant')"
             " AND m.content IS NOT NULL AND m.content != ''"
-            f" AND {NOT_INCOGNITO.format(col='m.session_id')} ORDER BY m.created_at LIMIT ?",
+            f" AND {NOT_INCOGNITO.format(col='m.session_id')} ORDER BY m.created_at, m.rowid LIMIT ?",
             (s, e, limit),
         )
         return [dict(r) for r in rows]

sentient/memory/facts.py

@@ -1154,7 +1154,7 @@ async def graph(self, max_nodes: int = 1500) -> dict:
         """v2 create_memory_graph: nodes plus links where cosine similarity >= threshold."""
         rows = await self.store.fetchall(
             f"SELECT {MEMORY_COLUMNS} FROM facts WHERE {ACTIVE} AND (expires_at IS NULL OR expires_at > ?)"
-            " ORDER BY created_at DESC LIMIT ?",
+            " ORDER BY created_at DESC, id DESC LIMIT ?",
             (now_iso(), max_nodes),
         )
         nodes = []

sentient/memory/usermodel.py

@@ -620,7 +620,7 @@ async def _gather(self, since: str) -> tuple[dict[str, dict], dict[str, str], di
         if cfg.recent_facts:
             rows = await store.fetchall(
                 "SELECT id, content, updated_at FROM facts WHERE status = 'active' AND updated_at > ?"
-                " AND (expires_at IS NULL OR expires_at > ?) ORDER BY updated_at DESC LIMIT ?",
+                " AND (expires_at IS NULL OR expires_at > ?) ORDER BY updated_at DESC, id DESC LIMIT ?",
                 (since, now_iso(), cfg.recent_facts),
             )
             lines = []
@@ -633,7 +633,7 @@ async def _gather(self, since: str) -> tuple[dict[str, dict], dict[str, str], di
             rows = await store.fetchall(
                 "SELECT m.id, m.content, m.end_at, COALESCE(NULLIF(m.untrusted, ''), s.untrusted) AS untrusted"
                 " FROM summaries m LEFT JOIN sessions s"
-                " ON s.id = m.session_id WHERE m.created_at > ? ORDER BY m.created_at DESC LIMIT ?",
+                " ON s.id = m.session_id WHERE m.created_at > ? ORDER BY m.created_at DESC, m.rowid DESC LIMIT ?",
                 (since, cfg.recent_summaries),
             )
             lines = []

sentient/store/db.py

@@ -36,6 +36,48 @@ def new_id() -> str:
     return uuid.uuid4().hex
 
 
+# ---------------------------------------------------------------- watermarks (#236)
+# Rows saved in the same clock tick share a timestamp (Windows' clock moves about every 15.6 ms), so "newer than the
+# last timestamp seen" skips a row saved in that tick after the read. A watermark keeps the newest timestamp read and
+# the ids already read at it, as JSON: {"at": "<iso>", "ids": [...]}. A plain timestamp (from an older build, or a
+# starting point) means "strictly newer than this".
+
+
+def _read_mark(mark: str | None) -> tuple[str, list]:
+    if not mark:
+        return "", []
+    try:
+        data = json.loads(mark)
+    except ValueError:
+        return mark, []
+    if isinstance(data, dict):
+        return str(data.get("at") or ""), list(data.get("ids") or [])
+    return mark, []
+
+
+def after_watermark(mark: str | None, ts_col: str, id_col: str) -> tuple[str, list]:
+    """SQL condition and parameters for rows after ``mark``: newer than its timestamp, or at it and not read yet."""
+    at, ids = _read_mark(mark)
+    if not ids:
+        return f"{ts_col} > ?", [at]
+    marks = ",".join("?" * len(ids))
+    return f"({ts_col} > ? OR ({ts_col} = ? AND {id_col} NOT IN ({marks})))", [at, at, *ids]
+
+
+def next_watermark(mark: str | None, rows: Iterable[Any], ts_key: str, id_key: str) -> str | None:
+    """The watermark after reading ``rows`` (each with ``ts_key`` and ``id_key``): their newest timestamp and the ids
+    read at it, kept together with ``mark``'s ids when the timestamp did not move. ``mark`` itself when no rows."""
+    rows = list(rows)
+    if not rows:
+        return mark
+    at = max(str(r[ts_key]) for r in rows)
+    ids = [r[id_key] for r in rows if str(r[ts_key]) == at]
+    old_at, old_ids = _read_mark(mark)
+    if old_at == at:
+        ids = [*old_ids, *(i for i in ids if i not in old_ids)]
+    return json.dumps({"at": at, "ids": ids})
+
+
 async def _load_vec(db: aiosqlite.Connection) -> bool:
     try:
         import sqlite_vec

sentient/store/schema.sql

@@ -19,7 +19,7 @@ CREATE TABLE IF NOT EXISTS sessions (
     updated_at  TEXT NOT NULL,
     archived    INTEGER NOT NULL DEFAULT 0,
     context_summary TEXT,                         -- running summary of turns older than the history window
-    context_upto    TEXT,                         -- created_at of the last message folded into the summary
+    context_upto    TEXT,                         -- watermark of the messages folded into the summary (store.db.next_watermark)
     untrusted       TEXT,                         -- app whose content this chat read ("Gmail"); sends then ask (ADR 0018)
     visited_hosts   TEXT,                         -- JSON list of web hosts this chat loaded (ADR 0018)
     incognito       INTEGER NOT NULL DEFAULT 0    -- 1: nothing is learned from it and it is deleted when closed (#154)

tests/evolution/test_evolution.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import asyncio
 import json
 from datetime import UTC, datetime, timedelta
 
@@ -180,15 +179,6 @@ async def test_curator_merge_proposal(app):
     assert app.skills.get_pending("email-digest") is not None
 
 
-async def clock_passes(store, key: str) -> None:
-    """Wait until the wall clock is past the time stored under ``key``. Windows' clock can stay on one value for
-    about 16 ms, so a write right after a run could otherwise carry the run's own timestamp."""
-    last = datetime.fromisoformat(await store.get_meta(key))
-    async with asyncio.timeout(5):
-        while datetime.now(UTC) <= last:
-            await asyncio.sleep(0.001)
-
-
 async def test_profile_upkeep_appends_without_clobbering(app):
     ws, mem = app.workspace, app.memory
     ws.write("user", "# About Sarthak\n\nMy own words: I like quiet mornings.\n")
@@ -205,7 +195,7 @@ async def test_profile_upkeep_appends_without_clobbering(app):
     assert ws.memory.read_text(encoding="utf-8").startswith("# Long-term memory")
     # nothing new: no rewrite, no duplicate bullets
     assert (await app.evolution.update_profile(force=True))["updated"] is False
-    await clock_passes(app.store, "evolution.profile_last_run")  # the next fact is learned after that upkeep
+    # the next fact may share that upkeep's clock tick; it is still picked up (#236)
     b = await mem.remember("Sarthak mentors two junior developers", use_llm=False)
     await app.store.execute("UPDATE facts SET topics = ? WHERE id = ?", ('["Work & Learning"]', b["id"]))
     app.fake.text_replies.append("no heading but long enough content for the memory file here")

tests/test_same_timestamp.py

@@ -0,0 +1,120 @@
+"""Rows saved in the same clock tick keep their order and are never skipped (issue #236).
+
+Windows' clock moves about every 15.6 ms, so these tests freeze the clock to make every row share one timestamp."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+
+from sentient.app import SentientApp
+from sentient.store import db as store_db
+from sentient.store.db import after_watermark, next_watermark
+from tests.conftest import FakeProvider
+
+TICK = "2026-10-11T06:00:00.123456+00:00"
+
+
+def freeze(monkeypatch, *modules) -> None:
+    for m in modules:
+        monkeypatch.setattr(m, "now_iso", lambda: TICK)
+
+
+def test_watermark_keeps_the_ids_read_at_its_timestamp():
+    assert after_watermark(None, "t", "id") == ("t > ?", [""])
+    assert after_watermark(TICK, "t", "id") == ("t > ?", [TICK])  # an older build's plain timestamp
+    mark = next_watermark(TICK, [{"t": "2026-10-11T05:00", "id": 1}, {"t": TICK, "id": 2}, {"t": TICK, "id": 3}], "t", "id")
+    assert json.loads(mark) == {"at": TICK, "ids": [2, 3]}
+    where, params = after_watermark(mark, "t", "id")
+    assert where == "(t > ? OR (t = ? AND id NOT IN (?,?)))" and params == [TICK, TICK, 2, 3]
+    # more rows at the same timestamp add to the ids; a newer timestamp starts over
+    assert json.loads(next_watermark(mark, [{"t": TICK, "id": 4}], "t", "id"))["ids"] == [2, 3, 4]
+    assert json.loads(next_watermark(mark, [{"t": "2026-10-12", "id": 5}], "t", "id")) == {"at": "2026-10-12", "ids": [5]}
+    assert next_watermark(mark, [], "t", "id") == mark
+
+
+async def test_same_tick_messages_keep_their_order(config, isolated_home, monkeypatch):
+    freeze(monkeypatch, store_db)
+    s = await SentientApp(config, llm=FakeProvider(), db_path=isolated_home / "order.db", enable_background=False).start()
+    try:
+        sid = await s.store.create_session(channel="cli")
+        for i in range(30):
+            await s.store.add_message(sid, "user" if i % 2 == 0 else "assistant", f"message {i}")
+        recent = await s.store.recent_messages(sid, 30)
+        assert [m["content"] for m in recent] == [f"message {i}" for i in range(30)]
+        between = await s.memory.episodic.messages_between("2026-10-11", "2026-10-11") if s.memory else []
+        assert [m["content"] for m in between] == [f"message {i}" for i in range(30)]
+    finally:
+        await s.stop()
+
+
+async def test_compression_does_not_skip_messages_saved_in_the_same_tick(config, isolated_home, monkeypatch):
+    freeze(monkeypatch, store_db)
+    seen: list[str] = []
+
+    class Recorder(FakeProvider):
+        async def complete_text(self, role, messages, *, model=None):
+            seen.append(messages[-1]["content"])
+            return "Running summary."
+
+    config.chat.history_window = 2
+    config.chat.compress_after_messages = 10
+    s = await SentientApp(config, llm=Recorder(), db_path=isolated_home / "compress.db", enable_background=False).start()
+    try:
+        sid = await s.store.create_session(channel="cli")
+        for i in range(20):
+            await s.store.add_message(sid, "user", f"early {i}")
+        await s.agent._maybe_compress(sid)
+        assert "early 17" in seen[0] and "early 18" not in seen[0]  # the last two stay in the window
+        for i in range(12):
+            await s.store.add_message(sid, "user", f"later {i}")
+        await s.agent._maybe_compress(sid)  # every message shares the first run's timestamp
+        assert len(seen) == 2
+        assert "early 17" not in seen[1] and "early 18" in seen[1] and "later 9" in seen[1]
+    finally:
+        await s.stop()
+
+
+async def test_profile_upkeep_learns_a_fact_saved_in_the_same_tick(config, isolated_home, monkeypatch):
+    from sentient.memory import facts as facts_mod
+
+    freeze(monkeypatch, store_db, facts_mod)
+    config.evolution.user_profile_updates = True
+    llm = FakeProvider()
+    s = await SentientApp(config, llm=llm, db_path=isolated_home / "upkeep.db", enable_background=False).start()
+    try:
+        mem, now = s.memory, datetime.fromisoformat(TICK)
+        assert mem is not None
+        a = await mem.remember("Maya is training for a marathon", use_llm=False)
+        await s.store.execute("UPDATE facts SET topics = ? WHERE id = ?", ('["Health & Wellbeing"]', a["id"]))
+        llm.text_replies.append("# Long-term memory\n\n## Goals\n- Maya is training for a marathon.\n")
+        assert (await s.evolution.update_profile(now, force=True))["learned_appended"] == 1
+        # saved after that run's read, in the same clock tick
+        b = await mem.remember("Maya mentors two junior developers", use_llm=False)
+        await s.store.execute("UPDATE facts SET topics = ? WHERE id = ?", ('["Work & Learning"]', b["id"]))
+        llm.text_replies.append("# Long-term memory\n\n## Work\n- Maya mentors two junior developers.\n")
+        out = await s.evolution.update_profile(now, force=True)
+        assert out["updated"] and out["learned_appended"] == 1
+        user_md = s.workspace.user.read_text(encoding="utf-8")
+        assert user_md.count("marathon") == 1 and "mentors two junior" in user_md
+        assert (await s.evolution.update_profile(now, force=True))["updated"] is False  # nothing new is read twice
+    finally:
+        await s.stop()
+
+
+async def test_profile_upkeep_after_an_empty_read_still_learns_a_same_tick_fact(config, isolated_home, monkeypatch):
+    from sentient.memory import facts as facts_mod
+
+    freeze(monkeypatch, store_db, facts_mod)
+    llm = FakeProvider()
+    s = await SentientApp(config, llm=llm, db_path=isolated_home / "empty.db", enable_background=False).start()
+    try:
+        now = datetime.fromisoformat(TICK)
+        assert (await s.evolution.update_profile(now, force=True))["updated"] is False  # nothing to read yet
+        assert s.memory is not None
+        a = await s.memory.remember("Maya is training for a marathon", use_llm=False)  # same tick as that run
+        await s.store.execute("UPDATE facts SET topics = ? WHERE id = ?", ('["Health & Wellbeing"]', a["id"]))
+        llm.text_replies.append("# Long-term memory\n\n## Goals\n- Maya is training for a marathon.\n")
+        assert (await s.evolution.update_profile(now, force=True))["learned_appended"] == 1
+    finally:
+        await s.stop()