Blob + identity storage API isolation - #76
Open
Diego Colombo (colombod) wants to merge 5 commits into
Open
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 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.