Blob + identity storage API isolation - #76
Closed
Diego Colombo (colombod) wants to merge 5 commits into
Closed
Conversation
…ync Protocol - Add neutral BlobStore Protocol: streaming list()/scan() as AsyncIterator[BlobReference], write()->BlobReference, fenced conditional delete(uri, if_unmodified=ref) - Implement AsyncDiskBlobStore with single shared registry.blob_store instance via DI - Privatize IdentityStore._path, add exists() method; fix main.py boot-time reach-throughs - Update /blobs routes to use shared store instance (no multiple construction sites) - Type services.blob_store as BlobStore|None (was Any) - Add isolation tripwire test: validates no blob-path FS ops leak outside blob_store.py - All operations are async and account for latency (thread-offloaded FS ops) Note: reclaim ENDPOINT rewire intentionally NOT included; it lives in PR #70 and will adopt this storage API when #70 is rebased on top of this. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
…ame-ms event collision The blob key is derived from make_node_id(session_id, event, timestamp), but without a tool_call_id disambiguator. Two parallel events with the same (session_id, event_name, timestamp-to-the-ms) but different tool_call_ids would mint identical blob keys, causing the second write to silently overwrite the first blob via os.replace, while both Event nodes still pointed at their original URIs. Fix: include tool_call_id (when present) in the blob key, matching how handlers/data_layer_1 derives the Event node id. This ensures each parallel event gets a distinct blob, even at the same millisecond. - Backward compatible: blob key format unchanged when tool_call_id is absent - Updated test to reflect 4-arg make_node_id call - Added regression test: parallel same-ms events with distinct tool_call_id do not collide 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
…ehind protocols
**Blob store redesign — package split:**
- Moved BlobStore Protocol, BlobReference, BlobNotFoundError to blob_store/protocol.py
- Renamed AsyncDiskBlobStore → FileSystemBlobStore; moved to blob_store/filesystem.py
- Added blob_store/factory.py with create_blob_store(settings) → BlobStore
- Reads settings.blob_backend to select backend ("filesystem" → FileSystemBlobStore, "azure" → NotImplementedError for now)
- This is the ONLY site where the concrete backend is constructed
- Enables Azure implementation as: one new file + one factory branch; zero consumer changes
- blob_store/__init__.py re-exports the protocol and factory function
**Config-driven backend selection:**
- Added blob_backend field to config.py (defaults to "filesystem")
- Registry now receives BlobStore via create_blob_store(settings), typed to the Protocol
- Registry no longer imports FileSystemBlobStore or touches settings.blob_path
- Consumers (registry and above) are entirely backend-agnostic
**Factory pattern applied uniformly to all storages:**
- queue_manager.py: added create_queue_manager(settings: Settings) → QueueManager
- Registry calls this instead of constructing QueueManager directly
- identity_store.py: added create_identity_store(settings: Settings, kind: str) → IdentityStore
- Selects entra or api_key backend by kind parameter
- main.py bootstrap uses this instead of direct construction
- All storage objects now obtained from factories; never instantiated by consumers
**Protocol-typed DI:**
- registry.blob_store now typed to BlobStore (the Protocol), not the concrete class
- This ensures type safety and makes backend swaps transparent to consumers
**Test isolation tightened:**
- test_blob_isolation_tripwire.py: concrete FileSystemBlobStore + settings.blob_path only inside blob_store package and config.py
- Verified non-vacuous: full suite 1873 passed, 4 skipped
- Rename refactor across all test files (AsyncDiskBlobStore → FileSystemBlobStore)
- conftest.py: added blob_backend to _SettingsProxy
**Design principle honored:**
Storages obtained from factories with configuration, not direct construction, so that all consumers are implementation-agnostic. This enables the Azure backend and any future backend to be added without touching consumer code.
Generated with [Amplifier](https://github.com/microsoft/amplifier)
Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
…n factory - Split context_intelligence_server/identity_store.py into a package: - protocol.py: IdentityStore runtime_checkable Protocol with load/put/delete/seed/get/items/__len__/exists + flat_dict - filesystem.py: FileSystemIdentityStore (renamed from concrete IdentityStore) - factory.py: create_identity_store(settings, kind) -> IdentityStore (single backend-selection seam) - __init__.py: re-exports Protocol + factory - Consumers (main.py, routers/admin.py, tests) now obtain instances via factory and type against Protocol - entra_identities_store_path and api_keys_store_path now read only by factory.py + config.py - Reworded 3 startup diagnostic log messages to reference config settings instead of raw paths - Added tests/test_identity_isolation_tripwire.py (non-vacuous enforcement) - Suite: 1875 passed, 4 skipped 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
Queue store is now obtained via create_queue_manager(settings) factory behind the QueueManager Protocol, matching blob and identity storages. The concrete FileSystemQueueManager is isolated to the queue_manager/ package and constructed only in factory.py, with no filesystem paths leaking to consumers. - Split context_intelligence_server/queue_manager.py into a package: - protocol.py: QueueManager runtime_checkable Protocol + Batch value type - filesystem.py: FileSystemQueueManager (renamed, behavior identical) - factory.py: create_queue_manager(settings) -> QueueManager (only reader of queues_path) - __init__.py: re-exports Protocol, concrete, factory - registry.py now types against QueueManager Protocol and builds via factory - Consumers never see FileSystemQueueManager or settings.queues_path - Non-vacuous tripwire added to catch isolation violations 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
This was referenced Aug 24, 2026
Diego Colombo (colombod)
added a commit
that referenced
this pull request
Aug 26, 2026
…ackend-neutral protocols Transplant PR #76's storage-API isolation onto PR #78's hardened durable queue so the two stack cleanly (78 -> 76). Every store is now reached only through a backend-neutral Protocol + factory; no consumer constructs a concrete backend or touches an on-disk path. queue_manager/ FileSystemQueueManager implements the QueueManager Protocol. Built on #78's authoritative body (the durable per-record cursor model): the class body is byte-identical to #78's queue_manager.py apart from the class rename and its self- references. protocol.py carries #78's Record/Batch verbatim; factory.create_queue_manager is the only queue backend selector. blob_store/ FileSystemBlobStore implements the BlobStore Protocol. write() returns a BlobReference (uri + size + last_modified); the store gains scan()/list() (async BlobReference iterators) and a fenced delete(uri, if_unmodified=ref). Adds settings.blob_backend. identity_store/ FileSystemIdentityStore implements the IdentityStore Protocol; the commit-order and fail-closed-load contract lives in the protocol. The backing path is private -- callers use exists(). Also folds in the blob-key fix: process_event includes tool_call_id in the blob-key node_id, so two same-millisecond parallel events no longer collide on one blob and silently overwrite each other. Consumers (registry, main, blob_processor, pipeline) build stores via the factories and pass URIs/references, never paths. The boot-reclaim dry-run log derives the blob path from the QueueManager's own queues_dir (fixing a latent mismatch when it differs from settings.queues_path); the identity first-boot warnings no longer echo the store path. The sole remaining config-path read outside the storage layer is registry.queues_dir_path -- the resolver the WriterLease boot detector uses precisely because it must not construct a QueueManager. Full non-neo4j suite: 2082 passed. Neo4j subsets (queue durability, blob ingest, identity auth): green. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
Diego Colombo (colombod)
marked this pull request as draft
August 26, 2026 12:17
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Isolates all blob (and oid/identity) disk access behind a backend-neutral async Protocol so the storage backend can later be swapped (local disk -> Azure Blob) without touching a single caller. Standalone work, based off main.
Contents: neutral BlobStore Protocol (BlobReference value object with no path; streaming list()/scan() as AsyncIterator[BlobReference]; write()->BlobReference; fenced conditional delete(uri, if_unmodified=ref) compare-and-delete inside the store; backend-neutral BlobNotFoundError with URI only, no on-disk path); single shared registry.blob_store instance; services typed BlobStore|None; /blobs routes through the shared store; IdentityStore path privatized + exists(); blob-key collision fix (tool_call_id disambiguator matches the graph Event node id, preventing parallel same-ms events from silently overwriting blobs); isolation tripwire test (proven non-vacuous).
Deliberately NOT here: the reclaim GC endpoint (POST /admin/blobs/reclaim) lives in PR #70. The new storage API (scan(), fenced delete(if_unmodified=)) is present so reclaim can be performed through the API; #70's endpoint will adopt this storage API when #70 is rebased on top of this. Keeps the isolation reviewable independently of the large #70.
Tests: full non-Neo4j suite green (1872 passed, 4 skipped). No Neo4j required for this PR.