mempalace.pending_queue

Source: mempalace/pending_queue.py

Append-only journal for mine requests that couldn't reach the daemon.

When PALACE_DAEMON_URL is set but the daemon (or its backend) is unreachable, _post_daemon_mine would previously log the failure and drop the request. With daemon-strict mode in effect, hooks don't fall back to a local mine — so a stale daemon means silent write loss for the duration of the outage.

This module captures the dropped requests as JSONL lines under ~/.mempalace/pending/YYYY-MM-DD.jsonl and exposes a replay function that re-issues them when the daemon recovers. Lines that succeed are removed from the file; lines that fail stay for the next attempt. Atomic rewrite via tempfile + os.replace prevents partial file corruption on crash mid-drain.

The request payload is tiny — {dir, wing, mode, ts} — because the transcript file the daemon actually mines lives on disk and survives the outage. We don't archive conversation content here; the file on disk is the durable source.

Classes

class ReplayReport

Outcome of a replay sweep.

is_empty

def is_empty(self) -> bool

Functions

enqueue

def enqueue(request: dict, *, now: datetime | None = None) -> Path

Append a pending mine request to today's queue file.

request must contain at minimum dir, wing, mode. A UTC ts field is added automatically. Writes are flushed and fsynced before return so a crash immediately after enqueue cannot lose the line.

pending_count

def pending_count(directory: Path | None = None) -> int

Return the number of queued (not-yet-replayed) requests.

Useful for the session-start warning. Cheap: counts non-empty lines without parsing JSON.

replay

def replay(post_fn: Callable[[dict], bool], *, directory: Optional[Path] = None, deadline: Optional[float] = None) -> ReplayReport

Drain the queue by re-issuing each request via post_fn.

post_fn(request) -> bool is the caller's hook to actually transmit one request. True means the daemon accepted it and the line should be consumed; False means keep it for next time. Anything that raises is treated as False — replay is a best-effort background sweep and must not abort partway just because one request blew up.

Concurrency model (Gemini PR #104 review): atomic-rewrite via tempfile + os.replace is NOT safe against concurrent enqueue calls: any line appended while we hold the file in memory would be silently lost when we replace. We instead "claim" each file by renaming it to <name>.replay-<pid> before reading. enqueue always writes to YYYY-MM-DD.jsonl, so any concurrent appends land in a fresh file rather than the one we're draining. Failed lines are appended back to the live file (not rewritten) so they merge with any newly-enqueued entries.

deadline is a time.monotonic() timestamp at which the sweep stops processing more lines and returns whatever has been drained so far. Used by hook_session_start to cap total replay cost on the hot path. None (default) means no time limit.