zettelkasten.usage

zettelkasten.usage

Access-recency sidecar for Zettelkasten retrieval reranking.

A tiny, concurrency-safe SQLite database recording when and how often each note was surfaced to the agent, so retrieval can boost recently-used notes as one channel of a blended reranker (see :mod:zettelkasten.reranking).

Design guardrails:

• Recency is a BOOST, never a penalty. A note with no access record is simply ABSENT from :func:recency_scores — it carries no recency signal and ranks purely on the other channels. This is deliberately NOT a decay/forgetting mechanism: nothing is ever pushed down for lacking a record.
• Sessions, not wall-clock. "Recency" is measured in sessions since last use against a session timeline (see :func:session_timeline), so a note used last session ranks ahead of one untouched for several sessions regardless of how many calendar days elapsed. The downstream reranker fuses on RANK (:func:memory.ranking.rrf_fuse), so the exponential-decay magnitude is not used for ordering — the only thing the decay shape controls is the STALENESS CUTOFF: a note whose last use is more than a few half-lives ago drops out of the recency channel entirely (absent, never penalized). half_life is therefore a membership knob (how long a note keeps counting as "recent"), not a ranking-shape knob.
• Universal timestamps. last_accessed_ts is a UTC unix epoch float, so a note accessed while working in a sibling (federated) repo carries a timestamp that maps correctly onto whichever repo's session timeline is queried.

The DB lives beside the embedding .kgl caches under <ANGELO_DIR>/zettel/usage.db (same base :mod:zettelkasten.embeddings resolves). Because ANGELO_DIR sits inside a cloud-synced (OneDrive CloudStorage) repo — where SQLite's WAL -shm/-wal shared-memory sidecars are unreliable and can corrupt the DB — it uses a rollback journal (journal_mode=DELETE): a single plain file the sync layer handles safely, with short transactions and a busy-timeout. The schema is created on the WRITE path only (never on a read), and every query path degrades gracefully (returns empty / no-ops) rather than crashing when the DB is missing or locked.

record_access

record_access(graph: str, note_id: str, session_id: str | None = None, weight: float = 1.0) -> None

Record that note_id in graph was surfaced/used now.

Upserts the (graph, note_id) row: sets last_accessed_ts to the current UTC epoch, records last_session_id, and bumps access_count.

access_count counts DISCRETE accesses and is bumped by exactly 1 per call (it is an integer tally of "how many times surfaced"). weight is accepted for forward-compatibility with a future weighted-recency scheme — the current recency model keys off last_accessed_ts (sessions-since-last- use), not the count — but it does not currently scale the tally. Passing it is harmless.

Never raises: a missing/locked DB is logged at debug level and swallowed, so a write on a hot path can never break note surfacing.

Source code in zettelkasten/usage.py

def record_access(
    graph: str,
    note_id: str,
    session_id: str | None = None,
    weight: float = 1.0,
) -> None:
    """Record that ``note_id`` in ``graph`` was surfaced/used now.

    Upserts the ``(graph, note_id)`` row: sets ``last_accessed_ts`` to the
    current UTC epoch, records ``last_session_id``, and bumps ``access_count``.

    ``access_count`` counts DISCRETE accesses and is bumped by exactly 1 per
    call (it is an integer tally of "how many times surfaced"). ``weight`` is
    accepted for forward-compatibility with a future weighted-recency scheme —
    the current recency model keys off ``last_accessed_ts`` (sessions-since-last-
    use), not the count — but it does not currently scale the tally. Passing it
    is harmless.

    Never raises: a missing/locked DB is logged at debug level and swallowed, so
    a write on a hot path can never break note surfacing.
    """
    if not graph or not note_id:
        return
    now = time.time()
    try:
        conn = _connect(create_schema=True)
        try:
            with conn:  # commits on success, rolls back on exception
                conn.execute(
                    """
                    INSERT INTO usage
                        (graph, note_id, last_accessed_ts, last_session_id, access_count)
                    VALUES (?, ?, ?, ?, 1)
                    ON CONFLICT(graph, note_id) DO UPDATE SET
                        last_accessed_ts = excluded.last_accessed_ts,
                        last_session_id = excluded.last_session_id,
                        access_count = usage.access_count + 1
                    """,
                    (graph, note_id, now, session_id),
                )
        finally:
            conn.close()
    except sqlite3.Error as exc:  # pragma: no cover - defensive (locked/missing)
        logger.debug("usage.record_access(%s/%s) failed: %s", graph, note_id, exc)

recency_scores

recency_scores(graph: str, note_ids: list[str], timeline: list[float], *, half_life_sessions: float = DEFAULT_HALF_LIFE_SESSIONS) -> dict[str, float]

Session-recency score in (0, 1] for each recorded note.

For every id in note_ids that has a recorded last_accessed_ts in graph, map that timestamp onto timeline (an ascending list of session start timestamps, e.g. from :func:session_timeline) to get "how many sessions ago" it was last used, then convert to a score via a session-based exponential decay (half_life_sessions). A score of 1.0 means "used this session".

Notes with NO record are ABSENT from the returned dict — they carry no recency signal and must never be zeroed or penalized. A note whose last use is STALE is likewise OMITTED rather than included with a near-zero score — so a single long-stale access can never make a note the sole rank-1 member of the recency channel. Staleness is bounded two ways: an ABSOLUTE ceiling of :data:_MAX_SESSIONS_AGO sessions (independent of half_life, so a large half_life cannot admit an ancient note) AND the half-life-relative floor :data:_RECENCY_FLOOR (more than :data:_STALENESS_HALF_LIVES half-lives ago). When timeline is empty there is no way to place a timestamp on it, so an empty dict is returned (again: absence, not penalty). Never raises: a missing/locked DB (including a not-yet-created schema) yields an empty dict.

Source code in zettelkasten/usage.py

def recency_scores(
    graph: str,
    note_ids: list[str],
    timeline: list[float],
    *,
    half_life_sessions: float = DEFAULT_HALF_LIFE_SESSIONS,
) -> dict[str, float]:
    """Session-recency score in ``(0, 1]`` for each recorded note.

    For every id in ``note_ids`` that has a recorded ``last_accessed_ts`` in
    ``graph``, map that timestamp onto ``timeline`` (an ascending list of session
    start timestamps, e.g. from :func:`session_timeline`) to get "how many
    sessions ago" it was last used, then convert to a score via a session-based
    exponential decay (``half_life_sessions``). A score of 1.0 means "used this
    session".

    Notes with NO record are ABSENT from the returned dict — they carry no
    recency signal and must never be zeroed or penalized. A note whose last use
    is STALE is likewise OMITTED rather than included with a near-zero score — so
    a single long-stale access can never make a note the sole rank-1 member of the
    recency channel. Staleness is bounded two ways: an ABSOLUTE ceiling of
    :data:`_MAX_SESSIONS_AGO` sessions (independent of ``half_life``, so a large
    ``half_life`` cannot admit an ancient note) AND the half-life-relative floor
    :data:`_RECENCY_FLOOR` (more than :data:`_STALENESS_HALF_LIVES` half-lives
    ago). When ``timeline`` is
    empty there is no way to place a timestamp on it, so an empty dict is
    returned (again: absence, not penalty). Never raises: a missing/locked DB
    (including a not-yet-created schema) yields an empty dict.
    """
    if not graph or not note_ids or not timeline:
        return {}
    # Deduplicate while preserving the ability to build a parameterized IN clause.
    unique_ids = list(dict.fromkeys(note_ids))
    scores: dict[str, float] = {}
    try:
        conn = _connect()
        try:
            # Chunk the IN clause to stay well under SQLite's variable limit
            # (999 by default) for large candidate sets.
            chunk = 400
            for start in range(0, len(unique_ids), chunk):
                batch = unique_ids[start : start + chunk]
                placeholders = ",".join("?" for _ in batch)
                rows = conn.execute(
                    f"SELECT note_id, last_accessed_ts FROM usage "
                    f"WHERE graph = ? AND note_id IN ({placeholders})",
                    (graph, *batch),
                ).fetchall()
                for note_id, ts in rows:
                    if ts is None:
                        continue
                    ago = _sessions_ago(float(ts), timeline)
                    # Absolute session ceiling FIRST, independent of half_life: a
                    # very large half_life keeps the decay above _RECENCY_FLOOR for
                    # dozens of sessions, so the floor cutoff below cannot bound
                    # membership on its own. Drop anything older than the ceiling
                    # so a long-stale access can never leapfrog top cosine.
                    if ago > _MAX_SESSIONS_AGO:
                        continue
                    score = _decay_score(ago, half_life_sessions)
                    # Staleness cutoff: omit genuinely-stale notes from the
                    # channel entirely (absence, never a penalty) so they can't
                    # be a sole rank-1 recency member and leapfrog top cosine.
                    if score < _RECENCY_FLOOR:
                        continue
                    scores[str(note_id)] = score
        finally:
            conn.close()
    except sqlite3.Error as exc:  # pragma: no cover - defensive (locked/missing)
        logger.debug("usage.recency_scores(%s) failed: %s", graph, exc)
        return {}
    return scores

session_timeline

session_timeline(repo_root: Path | str | None = None) -> list[float]

Build the querying repo's session timeline as ascending start timestamps.

Enumerates finalized sessions (.memory/sessions/*.md, start time parsed from the started_at frontmatter field) plus any live-session sidecars (.memory/active-sessions/*.json, started_at field), and returns their start timestamps sorted ascending. The last entry is therefore the most recent (current) session.

Concurrently-open chats each write their own live sidecar, but they are all active "now" — they OVERLAP into a single current session. Counting each as its own session would inflate the session count and over-count "sessions ago" for older notes (unfairly aging their recency). So the ACTIVE live sidecars are COLLAPSED into a single representative start (the earliest, so the whole concurrent window counts as one session). Only sidecars with a fresh heartbeat count as active (see :func:_sidecar_is_stale); a stale/abandoned sidecar left by a crashed chat is ignored so it is not mistaken for a concurrent session.

The collapsed live start is treated as the CURRENT session and is guaranteed to sort LAST: any finalized session whose start falls at/after it began DURING the still-open live window (it ran concurrently), so it is folded into the current session rather than left as a separate, "newer" bucket that would steal current-session status from the live chat. This keeps notes touched during the live chat anchored to "now" instead of aging several sessions stale (the leapfrog the caller's timeline[-1] == current assumption relies on).

Timestamps are UNIVERSAL (UTC epoch) so a note accessed in a sibling repo maps correctly onto this repo's timeline. Robust to a missing directory or unparseable files: those are skipped, never raised.

Source code in zettelkasten/usage.py

def session_timeline(repo_root: Path | str | None = None) -> list[float]:
    """Build the querying repo's session timeline as ascending start timestamps.

    Enumerates finalized sessions (``.memory/sessions/*.md``, start time parsed
    from the ``started_at`` frontmatter field) plus any live-session sidecars
    (``.memory/active-sessions/*.json``, ``started_at`` field), and returns their
    start timestamps sorted ascending. The last entry is therefore the most
    recent (current) session.

    Concurrently-open chats each write their own live sidecar, but they are all
    active "now" — they OVERLAP into a single current session. Counting each as
    its own session would inflate the session count and over-count "sessions
    ago" for older notes (unfairly aging their recency). So the ACTIVE live
    sidecars are COLLAPSED into a single representative start (the earliest, so
    the whole concurrent window counts as one session). Only sidecars with a
    fresh heartbeat count as active (see :func:`_sidecar_is_stale`); a
    stale/abandoned sidecar left by a crashed chat is ignored so it is not
    mistaken for a concurrent session.

    The collapsed live start is treated as the CURRENT session and is guaranteed
    to sort LAST: any finalized session whose start falls at/after it began
    DURING the still-open live window (it ran concurrently), so it is folded into
    the current session rather than left as a separate, "newer" bucket that would
    steal current-session status from the live chat. This keeps notes touched
    during the live chat anchored to "now" instead of aging several sessions
    stale (the leapfrog the caller's ``timeline[-1] == current`` assumption relies
    on).

    Timestamps are UNIVERSAL (UTC epoch) so a note accessed in a sibling repo
    maps correctly onto this repo's timeline. Robust to a missing directory or
    unparseable files: those are skipped, never raised.
    """
    base = _memory_dir(repo_root)
    starts: list[float] = []
    anchor_starts: list[float] = []
    now_ts = time.time()

    sessions_dir = base / "sessions"
    if sessions_dir.is_dir():
        for path in sessions_dir.glob("*.md"):
            try:
                text = path.read_text(encoding="utf-8")
            except OSError:
                continue
            # Only look inside the leading frontmatter block, so a stray
            # ``started_at:`` in prose can't spoof a session start.
            fm_match = re.match(r"^---\n(.*?)\n---", text, re.DOTALL)
            block = fm_match.group(1) if fm_match else text
            m = _STARTED_AT_RE.search(block)
            if not m:
                continue
            ts = _parse_iso_ts(m.group(1))
            if ts is not None:
                starts.append(ts)

    # The oldest known finalized-session start. Used to reject a legacy/malformed
    # live sidecar with an ANCIENT ``started_at`` (older than any finalized
    # session on record, and carrying no freshness proof) from becoming the
    # current-session anchor: such a sidecar is not credibly a currently-open
    # chat, and letting its ancient start pull ``current_start`` into the deep
    # past would drop every finalized session and collapse the whole timeline to
    # one bucket — mapping every note to ``sessions_ago == 0`` and defeating the
    # ``_MAX_SESSIONS_AGO`` ceiling in :func:`recency_scores`.
    oldest_finalized = min(starts) if starts else None

    active_dir = base / "active-sessions"
    if active_dir.is_dir():
        for path in active_dir.glob("*.json"):
            # Skip the server's ``.complete.json`` / ``reopen-request.json``
            # sidecars — only live per-session state carries a session start.
            if path.name.endswith(".complete.json") or path.name == "reopen-request.json":
                continue
            try:
                data = json.loads(path.read_text(encoding="utf-8"))
            except (OSError, json.JSONDecodeError, UnicodeDecodeError):
                continue
            if not isinstance(data, dict):
                continue
            # Ignore stale/abandoned sidecars (dead process or cold heartbeat) so
            # a crashed chat isn't counted as a concurrent live session.
            if _sidecar_is_stale(data, now_ts):
                continue
            ts = _parse_iso_ts(str(data.get("started_at") or ""))
            if ts is None:
                continue
            # Every non-stale live sidecar still COUNTS as a session bucket.
            starts.append(ts)
            # …but only a sidecar with POSITIVE liveness proof (live pid / fresh
            # heartbeat) — or, lacking that, one whose own start is not older than
            # the oldest finalized session — may DEFINE the current-session
            # anchor. A legacy/malformed sidecar with an ancient start is thus
            # folded in as an ordinary bucket but never drags ``current_start``
            # backwards (see ``oldest_finalized`` above).
            if _sidecar_has_freshness_signal(data, now_ts) or (
                oldest_finalized is None or ts >= oldest_finalized
            ):
                anchor_starts.append(ts)

    # Collapse every concurrently-open chat's ACTIVE sidecar into ONE current
    # session (its earliest anchor-eligible start), so N simultaneously-open
    # chats don't inflate the session count or over-count sessions-ago. The
    # collapsed start is the current session, so fold any session that began
    # during the live window (start at/after it) into it: this guarantees the
    # live start sorts LAST, so a session with a newer start can never become
    # ``timeline[-1]`` and shift the current-session anchor off the live chat.
    if anchor_starts:
        current_start = min(anchor_starts)
        starts = [s for s in starts if s < current_start]
        starts.append(current_start)

    starts.sort()
    return starts