Kevin

Provider Intake — Staging & Integrity

Source fetch → collision-proof staging filename → streaming hash + row count → field verification → working/validated/quarantine.

Provider Intake — Staging & Integrity (v2026-08-28T093439Z-747172b178) 2026-08-28 | Source fetch → collision-proof staging filename → streaming hash + row count → field verification → working/validated/quarantine. 1 - Intake Pipeline - Orchestrates the complete intake process IntakeHandler.execute() daemon/src/provider_intake/intake_handler.py:120 daemon/src/provider_intake/intake_handler.py:150 daemon/src/provider_intake/intake_handler.py:180 daemon/src/provider_intake/intake_handler.py:210 StreamingHasher.from_file() daemon/src/provider_intake/storage.py:120 daemon/src/provider_intake/storage.py:150 daemon/src/provider_intake/storage.py:180 daemon/src/provider_intake/storage.py:210 RowCounter.count_rows() daemon/src/provider_intake/storage.py:240 daemon/src/provider_intake/storage.py:270 daemon/src/provider_intake/storage.py:300 daemon/src/provider_intake/storage.py:330 source_path actual_hash, b… 2 - Schema & Validation - Ensures data integrity and schema compliance FieldVerifier.verify_fields() SchemaRegistry daemon/src/provider_intake/schema_registry.py:120 daemon/src/provider_intake/schema_registry.py:150 daemon/src/provider_intake/schema_registry.py:180 daemon/src/provider_intake/schema_registry.py:210 schema 3 - Storage & Completion - Stores validated data and publishes events IntakePath daemon/src/provider_intake/storage.py:360 daemon/src/provider_intake/storage.py:390 daemon/src/provider_intake/storage.py:420 storage.py EventBus.publish() daemon/src/provider_intake/intake_handler.py:240 daemon/src/provider_intake/intake_handler.py:270 daemon/src/provider_intake/intake_handler.py:300 daemon/src/provider_intake/intake_handler.py:330 output_path 1. Check idempotency 2. Download source 3. Compute hash 4. Count rows 5. Verify schema 6. Store validated 7. Register with Hank 8. Publish completion IntakeResult JSON { "intake_id": "intake-12345", "status": "success", "actual_row_count": 1000, "actual_hash": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", "duration_seconds": 10, "output_path": "/path/to/validated/intake-12345.parquet" } intake:completed event payload { "intake_id": "intake-12345", "provider_id": "provider-abc", "row_count": 1000, "output_path": "/path/to/validated/intake-12345.parquet" } The HankAdapter and EventBus are placeholders and need to be implemented via daemon context. Legend actor = initiates work · process = code path · store = state on disk · bus = durable queue · dashed = separate process cyan = request flow · pink = state read/write · dashed green = pull / return path · red = refusal As of 2026-08-28 11:16 UTC, from commit 68470f5

Provenance

This drawing is generated, not drawn. It is rebuilt from the source files below, so when they change the picture changes — a diagram here cannot quietly describe a system that no longer works this way.

Owner
Kevin
Slug
provider-intake
Rendered
2026-08-28T093439Z
From commit
747172b178
Updated
28 Aug 2026, 11:16 UTC
Sources at
68470f5
Watches
5 paths
  • daemon/src/provider_intake/intake_handler.py
  • daemon/src/provider_intake/storage.py
  • daemon/src/provider_intake/schema_registry.py
  • daemon/src/provider_intake/schemas.py
  • daemon/src/provider_intake/errors.py