fix: eliminate silent event loss from queue corruption on shared storage (+ crash-safe recovery)

[Diego Colombo (colombod)]

Durable, corruption-free event-ingestion queue

What this is

The event-ingestion server appends session events to per-session on-disk logs on a mounted file volume (Azure Files in the cloud, a local mount in development), then drains them into Neo4j. This work makes that durable queue corruption-free and self-bounding under concurrent writes, and makes the single-writer guarantee enforced rather than merely observed.

The guarantee rests on one property: a single writer holds a per-session file lock across the whole record, so a record lands as exactly one newline-terminated byte range or not at all — independent of the underlying filesystem's concurrent-append semantics. Storage is reclaimed only by draining and then deleting drained data; the queue is never capped or truncated.

What it delivers

• Corruption-free append under concurrency. Per-session serialization writes each record contiguously. Concurrent appends across many sessions, and to a single session, never tear, merge, or lose a record — including multi-megabyte records.
• Exactly-once idempotent ingest. A per-key lock serializes the check → append → store sequence, so two concurrent requests carrying the same idempotency key append exactly once. The key is recorded only after a durable append, so a failed write is retried rather than wrongly answered "duplicate".
• Self-bounding storage. A live session's already-committed prefix is reclaimed continuously regardless of its undrained tail size; fully-drained logs are reclaimed automatically; dead-letter files expire on a retention window. There is no size cap and no truncation of the queue.
• Crash-safe recovery and in-place upgrade. The committed-offset reader accepts both the current and the prior on-disk offset encodings, so an upgraded instance keeps draining data already on the volume. An unreadable offset never deletes an intact log. A poison batch that carries the terminal record still finalizes its session and reclaims its log.
• Enforced single writer. The writer lease defaults to enforce: a second process pointed at the same queue directory refuses to boot instead of silently corrupting it. A clean shutdown releases the lease so the single replica restarts immediately; an unclean exit is taken over after the staleness window, without crash-looping.
• Supervised drain and observability. Worker death, reap, and cancellation, and lease conflicts, emit structured session-tagged logs. Boot phase and counters, spool footprint, and lease state are exposed on /status.

Components added / modified

• context_intelligence_server/queue_manager.py — per-session write serialization and record framing; dual-format committed-offset reader; continuous committed-prefix compaction (no tail-size cap); boot classification and reclamation; non-truncating partial-write rollback.
• context_intelligence_server/registry.py — supervised drain loop; poison-batch isolation that finalizes when the batch carries the terminal record.
• context_intelligence_server/idempotency.py — idempotency cache plus a per-key async lock registry.
• context_intelligence_server/main.py — ingest endpoint with lease-serialized idempotency; boot reconciliation that automatically reclaims provably-drained logs; lease release on clean shutdown.
• context_intelligence_server/writer_lease.py — single-writer lease with enforce / detect / off modes, heartbeat, stale-takeover, and owner-gated release.
• context_intelligence_server/config.py — settings for compaction, dead-letter retention, and the writer lease (enforce by default).
• tests/ — a new concurrent-append stress suite (multi-session and single-session contention, records over 1 MiB), plus adverse-state coverage for offset back-compat, idempotency serialization, terminal-in-poison-batch finalization, uncapped prefix compaction, boot reclamation, non-destructive rollback, and lease enforce / stale-takeover / release.

Validation

• Full non-Neo4j suite green (2064 passed, 4 skipped), including the concurrent-append stress suite.
• Verified on a running single-replica instance: the dual-format offset reader made previously-unreadable sessions drainable (corrupt-offset count returned to zero), the accumulated spool drained into Neo4j and was reclaimed (multi-gigabyte queue directory reduced to a few hundred bytes, in-queue backlog to zero), and the writer lease was acquired in enforce mode with a clean-shutdown release on restart.

[Salil Das (sadlilas)]

Superseded by #92, which lands the same append-serialization fix in a smaller, reviewable form. Closing so there is a single authoritative PR for this work.