fix: half-open as-of interval for KG supersession

Fixes #1913.\n\nVerified locally on Windows with focused knowledge graph/MCP KG tests plus ruff check and ruff format --check.

Author: ggettert

CHANGELOG.md

@@ -8,6 +8,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 
 ## [Unreleased]
 
+### Features
+
+- **`supersede()` / `mempalace_kg_supersede` — atomic fact replacement.** Closes an open fact and opens its successor at a single shared instant, so a point-in-time query at the boundary returns only the new value. This is the primitive for a single-valued fact change (model, employer, address) instead of hand-rolling `kg_invalidate` + `kg_add`, which left the two facts sharing the transition day. `at` defaults to the current UTC instant. (#1913)
+
+### Bug Fixes
+
+- **`kg query --as-of` no longer returns a superseded fact and its successor at the shared boundary.** `_temporal_filter_sql` now treats validity as half-open `[valid_from, valid_to)` (strict upper bound), so a fact whose `valid_to` equals the query instant has ended and only the successor matches. Standalone date-only facts still stay valid through the end of their final day (whole-day expansion retained). (#1913)
+
 ---
 
 ## [3.5.0] — 2026-06-22

mempalace/knowledge_graph.py

@@ -39,7 +39,7 @@
 import os
 import sqlite3
 import threading
-from datetime import date, datetime
+from datetime import date, datetime, timezone
 from pathlib import Path
 from typing import Optional
 from .config import sanitize_iso_temporal
@@ -113,6 +113,14 @@ def _temporal_filter_sql(as_of: str) -> tuple[str, list[str]]:
 
     This keeps legacy date-only facts working when callers query with
     canonical UTC datetimes such as '2026-05-06T15:00:00Z'.
+
+    The upper bound is *strict* (``valid_to > as_of``): a fact whose
+    ``valid_to`` equals the query instant has already ended at that instant,
+    so the interval is treated as half-open ``[valid_from, valid_to)``. This
+    is what lets a fact and its successor share a boundary instant without an
+    as-of query returning both. Date-only ``valid_to`` still expands to the
+    end of that day (``T23:59:59Z``), so a standalone date-only fact stays
+    valid through its whole final day exactly as before.
     """
 
     as_of_key = _temporal_start_key(as_of)
@@ -121,7 +129,7 @@ def _temporal_filter_sql(as_of: str) -> tuple[str, list[str]]:
 
     return (
         f" AND (t.valid_from IS NULL OR {valid_from_expr} <= ?) "
-        f"AND (t.valid_to IS NULL OR {valid_to_expr} >= ?)",
+        f"AND (t.valid_to IS NULL OR {valid_to_expr} > ?)",
         [as_of_key, as_of_key],
     )
 
@@ -359,6 +367,125 @@ def invalidate(self, subject: str, predicate: str, obj: str, ended: str = None):
                     (ended, sub_id, pred, obj_id),
                 )
 
+    def supersede(
+        self,
+        subject: str,
+        predicate: str,
+        old_obj: str,
+        new_obj: str,
+        at: str = None,
+        confidence: float = 1.0,
+        source_closet: str = None,
+        source_file: str = None,
+        source_drawer_id: str = None,
+        adapter_name: str = None,
+    ):
+        """Atomically replace one fact with another at a single shared boundary.
+
+        Closes the currently-open ``(subject, predicate, old_obj)`` triple with
+        ``valid_to = at`` and opens ``(subject, predicate, new_obj)`` with
+        ``valid_from = at`` in one transaction, at a single shared instant.
+        Paired with the half-open upper bound in ``_temporal_filter_sql``, an
+        as-of query at that instant returns only the successor.
+
+        This is the primitive for a value change. Hand-rolling a handover as
+        ``invalidate(ended=D)`` + ``add_triple(valid_from=D)`` with date-only
+        ``D`` leaves two facts sharing the whole day ``D`` (``valid_to`` expands
+        to ``T23:59:59Z`` while ``valid_from`` expands to ``T00:00:00Z``), so an
+        as-of query on ``D`` returns both. ``supersede`` avoids this by writing
+        one identical precise instant to both sides.
+
+        ``at`` defaults to the current UTC instant. A date-only ``at`` is
+        normalized to ``<date>T00:00:00Z`` so both sides carry the same precise
+        value rather than the asymmetric whole-day expansion.
+
+        Returns the new triple's id. If no open ``old_obj`` triple exists the
+        successor is still opened, so ``supersede`` degrades to ``add_triple``.
+        """
+        if at is None:
+            boundary = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+        elif _is_date_only_temporal(at):
+            boundary = f"{at}T00:00:00Z"
+        else:
+            boundary = at
+        boundary = sanitize_iso_temporal(boundary, "at")
+
+        sub_id = self._entity_id(subject)
+        old_id = self._entity_id(old_obj)
+        new_id = self._entity_id(new_obj)
+        pred = predicate.lower().replace(" ", "_")
+
+        with self._lock:
+            conn = self._conn()
+            with conn:
+                # Only create entities we actually open a fact for. old_obj is
+                # matched by id in the UPDATE below whether or not its row
+                # exists, so inserting it would just orphan an entity when no
+                # open old fact is present (the degrade-to-add path).
+                for name, eid in ((subject, sub_id), (new_obj, new_id)):
+                    conn.execute(
+                        "INSERT OR IGNORE INTO entities (id, name) VALUES (?, ?)",
+                        (eid, name),
+                    )
+
+                # Reject a boundary that precedes the old fact's start — an
+                # inverted interval would be invisible to every KG query.
+                rows = conn.execute(
+                    "SELECT valid_from FROM triples "
+                    "WHERE subject=? AND predicate=? AND object=? AND valid_to IS NULL",
+                    (sub_id, pred, old_id),
+                ).fetchall()
+                for row in rows:
+                    valid_from = row["valid_from"]
+                    if valid_from is not None and _temporal_end_key(boundary) < _temporal_start_key(
+                        valid_from
+                    ):
+                        raise ValueError(
+                            f"at={boundary!r} is before valid_from={valid_from!r}; "
+                            "an inverted interval would be invisible to every KG query"
+                        )
+
+                # Close the open old fact at the shared boundary.
+                conn.execute(
+                    "UPDATE triples SET valid_to=? "
+                    "WHERE subject=? AND predicate=? AND object=? AND valid_to IS NULL",
+                    (boundary, sub_id, pred, old_id),
+                )
+
+                # Open the successor at the same instant (idempotent if already open).
+                existing = conn.execute(
+                    "SELECT id FROM triples "
+                    "WHERE subject=? AND predicate=? AND object=? AND valid_to IS NULL",
+                    (sub_id, pred, new_id),
+                ).fetchone()
+                if existing:
+                    return existing["id"]
+
+                triple_id = make_triple_id(
+                    sub_id, pred, new_id, boundary, datetime.now().isoformat()
+                )
+                conn.execute(
+                    """INSERT INTO triples (
+                        id, subject, predicate, object, valid_from, valid_to,
+                        confidence, source_closet, source_file,
+                        source_drawer_id, adapter_name
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                    (
+                        triple_id,
+                        sub_id,
+                        pred,
+                        new_id,
+                        boundary,
+                        None,
+                        confidence,
+                        source_closet,
+                        source_file,
+                        source_drawer_id,
+                        adapter_name,
+                    ),
+                )
+                return triple_id
+
     # ── Query operations ──────────────────────────────────────────────────
 
     def query_entity(self, name: str, as_of: str = None, direction: str = "outgoing"):

mempalace/mcp_server.py

@@ -381,6 +381,7 @@ def _parse_args():
     {
         "mempalace_kg_add",
         "mempalace_kg_invalidate",
+        "mempalace_kg_supersede",
         "mempalace_create_tunnel",
         "mempalace_delete_tunnel",
         "mempalace_delete_hallway",
@@ -1797,7 +1798,7 @@ def tool_status():
 2. BEFORE RESPONDING about any person, project, or past event: call mempalace_kg_query or mempalace_search FIRST. Never guess — verify.
 3. IF UNSURE about a fact (name, gender, age, relationship): say "let me check" and query the palace. Wrong is worse than slow.
 4. AFTER EACH SESSION: call mempalace_diary_write to record what happened, what you learned, what matters.
-5. WHEN FACTS CHANGE: call mempalace_kg_invalidate on the old fact, mempalace_kg_add for the new one.
+5. WHEN A SINGLE-VALUED FACT CHANGES (model, employer, address): call mempalace_kg_supersede(subject, predicate, old, new) to replace it atomically at one boundary — do NOT hand-roll invalidate + add, which leaves the old and new values overlapping at the boundary. Use mempalace_kg_invalidate for a fact that simply ended, and mempalace_kg_add to add an independent (possibly concurrent) fact.
 
 This protocol ensures the AI KNOWS before it speaks. Storage is not memory — but storage + this protocol = memory."""
 
@@ -3357,6 +3358,57 @@ def tool_kg_invalidate(subject: str, predicate: str, object: str, ended: str = N
     }
 
 
+def tool_kg_supersede(
+    subject: str,
+    predicate: str,
+    old_object: str,
+    new_object: str,
+    at: str = None,
+):
+    """Atomically replace one fact with another at a single shared boundary.
+
+    Closes ``(subject, predicate, old_object)`` and opens
+    ``(subject, predicate, new_object)`` at one shared instant, so a
+    point-in-time query at the boundary returns only the new value. Use this
+    instead of a separate ``kg_invalidate`` + ``kg_add`` when a single-valued
+    fact changes (e.g. a model, employer, or address changes).
+
+    ``at`` accepts ``YYYY-MM-DD`` or a canonical UTC datetime
+    (``YYYY-MM-DDTHH:MM:SSZ``) and defaults to the current UTC instant.
+    """
+    try:
+        subject = sanitize_kg_value(subject, "subject")
+        predicate = sanitize_name(predicate, "predicate")
+        old_object = sanitize_kg_value(old_object, "old_object")
+        new_object = sanitize_kg_value(new_object, "new_object")
+        at = sanitize_iso_temporal(at, "at")
+    except ValueError as e:
+        return {"success": False, "error": str(e)}
+
+    _wal_log(
+        "kg_supersede",
+        {
+            "subject": subject,
+            "predicate": predicate,
+            "old_object": old_object,
+            "new_object": new_object,
+            "at": at,
+        },
+    )
+
+    # Domain ValueErrors from kg.supersede (e.g. inverted boundary) are left to
+    # bubble to the dispatcher, matching tool_kg_add / tool_kg_invalidate: the
+    # -32000 response carries error_class + message in error.data. Only input
+    # sanitization above returns the {success: False} envelope.
+    triple_id = _call_kg(lambda kg: kg.supersede(subject, predicate, old_object, new_object, at=at))
+    return {
+        "success": True,
+        "triple_id": triple_id,
+        "fact": f"{subject} → {predicate} → {new_object}",
+        "superseded": old_object,
+    }
+
+
 def tool_kg_timeline(entity: str = None):
     """Get chronological timeline of facts, optionally for one entity."""
     if entity is not None:
@@ -3976,6 +4028,27 @@ def tool_checkpoint(items, diary=None, dedup_threshold=0.9):
         },
         "handler": tool_kg_invalidate,
     },
+    "mempalace_kg_supersede": {
+        "description": "Atomically replace a fact with its successor at a shared boundary. Use when a single-valued fact changes (model, employer, address) instead of separate kg_invalidate + kg_add — a point-in-time query at the boundary then returns only the new value.",
+        "input_schema": {
+            "type": "object",
+            "properties": {
+                "subject": {"type": "string", "description": "The entity whose fact is changing"},
+                "predicate": {
+                    "type": "string",
+                    "description": "The relationship type (e.g. 'uses_model', 'works_at')",
+                },
+                "old_object": {"type": "string", "description": "The value being replaced"},
+                "new_object": {"type": "string", "description": "The new value"},
+                "at": {
+                    "type": "string",
+                    "description": "Boundary instant (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SSZ, optional; defaults to now UTC)",
+                },
+            },
+            "required": ["subject", "predicate", "old_object", "new_object"],
+        },
+        "handler": tool_kg_supersede,
+    },
     "mempalace_kg_timeline": {
         "description": "Chronological timeline of facts. Shows the story of an entity (or everything) in order.",
         "input_schema": {

tests/test_knowledge_graph.py

@@ -322,3 +322,71 @@ def test_context_manager_closes_connection(self, tmp_path):
         assert kg._connection is None
         with pytest.raises(sqlite3.ProgrammingError):
             conn.execute("SELECT 1")
+
+
+class TestSupersessionBoundary:
+    """Regression coverage for the as-of boundary double-count (issue #1913):
+    an as-of query at the instant one fact ends and its successor begins must
+    return only the successor for a single-valued predicate."""
+
+    def _models(self, kg, as_of):
+        return sorted(
+            f["object"]
+            for f in kg.query_entity("Bot", as_of=as_of, direction="outgoing")
+            if f["predicate"] == "uses_model"
+        )
+
+    def test_exact_datetime_boundary_returns_only_successor(self, kg):
+        # Two facts sharing a precise instant: half-open upper bound (strict >)
+        # means the fact ending at T no longer matches at T.
+        kg.add_triple(
+            "Bot",
+            "uses_model",
+            "A",
+            valid_from="2026-05-01T00:00:00Z",
+            valid_to="2026-06-02T12:00:00Z",
+        )
+        kg.add_triple("Bot", "uses_model", "B", valid_from="2026-06-02T12:00:00Z")
+
+        assert self._models(kg, "2026-06-02T11:59:59Z") == ["A"]
+        assert self._models(kg, "2026-06-02T12:00:00Z") == ["B"]
+        assert self._models(kg, "2026-06-02T12:00:01Z") == ["B"]
+
+    def test_supersede_date_only_resolves_to_successor(self, kg):
+        kg.add_triple("Bot", "uses_model", "claude-opus-4-7", valid_from="2026-05-01")
+        kg.supersede("Bot", "uses_model", "claude-opus-4-7", "claude-opus-4-8", at="2026-06-02")
+
+        assert self._models(kg, "2026-06-01") == ["claude-opus-4-7"]
+        assert self._models(kg, "2026-06-02") == ["claude-opus-4-8"]
+        assert self._models(kg, "2026-06-03") == ["claude-opus-4-8"]
+
+    def test_supersede_datetime_boundary_resolves_to_successor(self, kg):
+        kg.add_triple("Bot", "uses_model", "A", valid_from="2026-05-01T00:00:00Z")
+        kg.supersede("Bot", "uses_model", "A", "B", at="2026-06-02T12:00:00Z")
+
+        assert self._models(kg, "2026-06-02T11:59:59Z") == ["A"]
+        assert self._models(kg, "2026-06-02T12:00:00Z") == ["B"]
+
+    def test_supersede_default_now_closes_old_and_opens_new(self, kg):
+        kg.add_triple("Bot", "uses_model", "A", valid_from="2026-05-01")
+        kg.supersede("Bot", "uses_model", "A", "B")
+        # A far-future as-of sees only the successor; the old fact was closed.
+        assert self._models(kg, "2099-01-01") == ["B"]
+
+    def test_supersede_degrades_to_add_when_no_open_old(self, kg):
+        tid = kg.supersede("Bot", "uses_model", "missing", "B", at="2026-01-01")
+        assert tid.startswith("t_bot_uses_model_b_")
+        assert self._models(kg, "2099-01-01") == ["B"]
+
+    def test_supersede_rejects_boundary_before_valid_from(self, kg):
+        kg.add_triple("Bot", "uses_model", "A", valid_from="2026-06-01")
+        with pytest.raises(ValueError, match="before valid_from"):
+            kg.supersede("Bot", "uses_model", "A", "B", at="2026-05-01")
+
+    def test_standalone_date_only_end_stays_valid_all_day(self, kg):
+        # Half-open change must NOT shrink a standalone date-only fact: it stays
+        # valid through the end of its final day (whole-day expansion retained).
+        kg.add_triple("Bot", "uses_model", "A", valid_from="2026-05-01", valid_to="2026-06-02")
+        assert self._models(kg, "2026-06-02") == ["A"]
+        assert self._models(kg, "2026-06-02T23:00:00Z") == ["A"]
+        assert self._models(kg, "2026-06-03") == []

tests/test_mcp_server.py

@@ -2860,6 +2860,27 @@ def test_kg_invalidate(self, monkeypatch, config, palace_path, seeded_kg):
         # not silently drop it and return the literal string "today".
         assert result["ended"] == "2026-03-01"
 
+    def test_kg_supersede(self, monkeypatch, config, palace_path, kg):
+        _patch_mcp_server(monkeypatch, config, kg)
+        from mempalace.mcp_server import tool_kg_supersede
+
+        kg.add_triple("Bot", "uses_model", "old", valid_from="2026-05-01")
+        result = tool_kg_supersede(
+            subject="Bot",
+            predicate="uses_model",
+            old_object="old",
+            new_object="new",
+            at="2026-06-02",
+        )
+        assert result["success"] is True
+        assert result["superseded"] == "old"
+        models = [
+            f["object"]
+            for f in kg.query_entity("Bot", as_of="2026-06-02", direction="outgoing")
+            if f["predicate"] == "uses_model"
+        ]
+        assert models == ["new"]
+
     def test_kg_add_forwards_valid_to(self, monkeypatch, config, palace_path, kg):
         """Regression #1314 case 1: valid_to must round-trip through kg_add."""
         _patch_mcp_server(monkeypatch, config, kg)

website/reference/mcp-tools.md

@@ -263,6 +263,22 @@ Mark a fact as no longer true.
 
 ---
 
+### `mempalace_kg_supersede`
+
+Atomically replace a fact with its successor at a single shared boundary. Use when a single-valued fact changes (model, employer, address) instead of a separate `mempalace_kg_invalidate` + `mempalace_kg_add` — a point-in-time query at the boundary then returns only the new value.
+
+| Parameter | Type | Required | Description |
+|-----------|------|----------|-------------|
+| `subject` | string | **Yes** | Entity whose fact is changing |
+| `predicate` | string | **Yes** | Relationship (e.g. `uses_model`, `works_at`) |
+| `old_object` | string | **Yes** | Value being replaced |
+| `new_object` | string | **Yes** | New value |
+| `at` | string | No | Boundary instant (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SSZ; default: now UTC) |
+
+**Returns:** `{ success, triple_id, fact, superseded }`
+
+---
+
 ### `mempalace_kg_timeline`
 
 Chronological timeline of facts.