[truealpha-datahub] Accept ephemeral capture persistence - #193
Merged
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
This PR advances the D3 Staging TOPT capture batch from E2→E3 by introducing an E2 persistence slice that lands raw Yahoo responses into the S3-compatible raw store and raw.fetches, then appends normalized market-price rows and persists replayable registry/scope/manifest/evaluation contracts into Postgres.
Changes:
- Advance D3 batch governance metadata to reflect E2 acceptance and target E3.
- Add
staging_topt_captureD3 E2 persistence implementation (persist_e1_execution) for raw + normalized + contract persistence with idempotent identical-byte retries and restatement-on-changed-bytes behavior. - Add an integration test suite validating E2 idempotence, append-only enforcement, retry landing, restatement semantics, and S3 round-trip.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 4 comments.
| File | Description |
|---|---|
| governance/vision-issue-graph.json | Advances the D3 batch target rung to E3 and updates the manifest hash pointer. |
| governance/batches/D3-staging-topt-capture.v1.json | Updates batch revision/acceptance metadata for the E2 slice and pins the base SHA. |
| apps/data-engine/src/data_engine/batches/staging_topt_capture/e2_slice.py | Implements D3 E2 persistence for raw bytes, normalized records, and replayable capture contracts. |
| apps/data-engine/tests/batches/staging_topt_capture/test_e2_slice.py | Adds integration coverage for E2 persistence semantics (idempotence, append-only, restatements, S3 round-trip). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+123
to
+124
| adapter_id="d3:YahooRawHttpAdapter", | ||
| adapter_version=VERSION, |
Comment on lines
+230
to
+236
| if ( | ||
| result.source_id != context.source_entry.source_id | ||
| or result.source_version != context.source_entry.version | ||
| or response.sha256 != result.normalized_bar.raw_response_sha256 | ||
| ): | ||
| raise ValueError("E1 result does not bind the E2 source or raw bytes") | ||
|
|
Comment on lines
+345
to
+349
| evidence = CaptureRecordEvidence( | ||
| source_coverage_entry_id=SOURCE_COVERAGE_ENTRY_ID, | ||
| raw_id=f"raw.object:{record.raw_object_sha256}", | ||
| raw_sha256=record.raw_object_sha256, | ||
| normalized_id=record.normalized_record_id, |
Comment on lines
+428
to
+469
| def persist_e1_execution( | ||
| connection: Connection[Any], | ||
| raw_store: RawObjectStore, | ||
| execution: D3E1TinyExecution, | ||
| context: D3E2CaptureContext, | ||
| *, | ||
| predecessor: NormalizedRecordRef | None = None, | ||
| ) -> D3E2PersistenceResult: | ||
| """Persist one accepted E1 execution through immutable E2 stores.""" | ||
|
|
||
| if not execution.landed_raw_responses: | ||
| raise ValueError("D3 E2 requires retained E1 raw bytes") | ||
| _verify_execution_binding(execution, context) | ||
| raw_fetch_ids: list[int] = [] | ||
| final_landing = execution.landed_raw_responses[-1] | ||
| for landed in execution.landed_raw_responses: | ||
| response = landed.response | ||
| source_record_suffix = "result" if landed is final_landing else f"attempt:{response.attempt_number}" | ||
| fetch_id = insert_fetch( | ||
| connection, | ||
| source=DataSource.YAHOO, | ||
| source_record_id=f"{response.call_plan_id}:{source_record_suffix}", | ||
| body=response.body, | ||
| content_type=response.content_type or "application/octet-stream", | ||
| fetched_at=response.fetched_at, | ||
| metadata={ | ||
| "source_id": response.source_id, | ||
| "source_version": response.source_version, | ||
| "adapter_id": response.adapter_id, | ||
| "adapter_version": response.adapter_version, | ||
| "call_plan_id": response.call_plan_id, | ||
| "configuration_sha256": response.configuration_sha256, | ||
| "attempt_number": response.attempt_number, | ||
| "landing_id": landed.landing_id, | ||
| "e1_interaction_id": execution.result.interaction_id, | ||
| }, | ||
| store=raw_store, | ||
| recorded_at=response.fetched_at + timedelta(seconds=2), | ||
| ) | ||
| if get_payload(connection, fetch_id, store=raw_store) != response.body: | ||
| raise ValueError("persisted raw bytes failed checksum-verified readback") | ||
| raw_fetch_ids.append(fetch_id) |
This was referenced Jul 14, 2026
wangzitian0
changed the title
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
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.
Summary
raw.fetchesbefore writing the generic normalized recordAcceptance evidence
make checkon another fresh database: 709 passed, 4 skipped3b18ac4585c4f5183bac376b5a105f0760df06baClaim ceiling
This accepts ephemeral Local/CI persistence for one frozen NVDA market-price cell. It does not claim the 20-issuer/21-instrument/84-cell denominator, a live Staging run, scheduling, Production access, or release activation.
Advances #171.