feat(delphi): storage v2 P5 — explicit --job-id threading through all pipeline entry points#2601
Open
jucor wants to merge 1 commit into
Open
feat(delphi): storage v2 P5 — explicit --job-id threading through all pipeline entry points#2601jucor wants to merge 1 commit into
jucor wants to merge 1 commit into
Conversation
Collaborator
Author
|
Post-push self-review (Claude subagent): no behavior regressions found across the six stage commands, the poller extraction, or the entry-point changes (math body untouched — golden invariance holds). One test-hermeticity fix applied and pushed: run_delphi.main()'s transition export writes DELPHI_JOB_ID to the real os.environ, which monkeypatch can't undo — the test module now has an autouse fixture that snapshots/restores the variable, keeping the session hermetic. (The reviewer also noted the pre-existing broken |
jucor
force-pushed
the
jc/delphi-storage-v2-p5
branch
from
July 6, 2026 15:47
35fed19 to
95099f9
Compare
jucor
added a commit
that referenced
this pull request
Jul 6, 2026
… + purge tool
Implements the capture half of Stack 2 / P6 (docs/STORAGE_V2_DESIGN.md §4.2
entity 2, §4.4): reproducibility begins with recording EXACTLY what the
pipeline read. The read seam (--input-source, P6b) follows.
- delphi_storage/inputs.py — capture_run_inputs(store, job_id, zid, rid):
writes the six snapshot kinds as run_inputs items (codec envelopes,
zstd-compressed votes) and returns per-kind fingerprints (sha256 of the
canonical uncompressed payload, row_count, max vote created — the P7
manifest inputs). Key fidelity decisions, pinned by tests:
- votes = the RAW stream stage 1 consumes: ORDER BY created, raw PG signs
(the production math path never flips — fetch_votes()'s flip is dead
code), superseded votes included, order preserved (KMeans init depends
on vote-encounter order, INVESTIGATION_K_DIVERGENCE.md).
- The vote SQL is shared BY CONSTRUCTION: run_math_pipeline.py now imports
VOTES_COUNT_SQL / VOTES_BATCH_SQL from delphi_storage.inputs (strings
unchanged; golden suite + the e2e SQL pins stay green).
- comments/participants captured in FULL, including moderated-out rows
(mod = -1) and banned participants — filtering is pipeline behavior,
not snapshot behavior.
- clojure_math_main captured for ALL math_env rows (501 reads env-less
latest, 801 filters MATH_ENV with a different default — snapshotting
every env keeps both replayable).
- derive_votes_latest_unique() reproduces PG's RULE-maintained
votes_latest_unique from the stream (last vote per (pid, tid) in stream
order) — tested against a seeded PG table mimicking the RULE.
- delphi_storage/purge.py — purge_job(): deletes every generic-entity
partition for a job (the §9 GDPR path for snapshot copies).
- run_delphi.py — --snapshot-inputs flag (or DELPHI_SNAPSHOT_INPUTS=1):
captures before any stage runs, aborts the run on capture failure
(a run without recorded inputs defeats the point when enabled).
Off by default until P7's DELPHI_WRITE_MODE subsumes it.
Tests (TDD; tests/test_input_snapshots.py, RED first): 16 tests — raw
stream order/signs, compression, fingerprints, derivation-vs-PG-RULE,
full-row comments/participants, all-env math_main, SQL-parity-by-import,
purge, and the run_delphi hook (flag, env, off-by-default). PG-backed tests
use a seeded scratch database (skip-locally / fail-in-CI convention).
Full suite: 429 passed, 41 skipped, 58 xfailed (5 pre-existing
DynamoDB-unavailable errors unchanged); golden snapshots ran and passed —
math outputs untouched.
Stack: P1 #2597, P2 #2598, P3 #2599, P4 #2600, P5 #2601, P6a this commit.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
jucor
force-pushed
the
jc/delphi-storage-v2-p4
branch
from
July 6, 2026 18:48
5a5e0d4 to
5f9fbd5
Compare
jucor
added a commit
that referenced
this pull request
Jul 6, 2026
… + purge tool
Implements the capture half of Stack 2 / P6 (docs/STORAGE_V2_DESIGN.md §4.2
entity 2, §4.4): reproducibility begins with recording EXACTLY what the
pipeline read. The read seam (--input-source, P6b) follows.
- delphi_storage/inputs.py — capture_run_inputs(store, job_id, zid, rid):
writes the six snapshot kinds as run_inputs items (codec envelopes,
zstd-compressed votes) and returns per-kind fingerprints (sha256 of the
canonical uncompressed payload, row_count, max vote created — the P7
manifest inputs). Key fidelity decisions, pinned by tests:
- votes = the RAW stream stage 1 consumes: ORDER BY created, raw PG signs
(the production math path never flips — fetch_votes()'s flip is dead
code), superseded votes included, order preserved (KMeans init depends
on vote-encounter order, INVESTIGATION_K_DIVERGENCE.md).
- The vote SQL is shared BY CONSTRUCTION: run_math_pipeline.py now imports
VOTES_COUNT_SQL / VOTES_BATCH_SQL from delphi_storage.inputs (strings
unchanged; golden suite + the e2e SQL pins stay green).
- comments/participants captured in FULL, including moderated-out rows
(mod = -1) and banned participants — filtering is pipeline behavior,
not snapshot behavior.
- clojure_math_main captured for ALL math_env rows (501 reads env-less
latest, 801 filters MATH_ENV with a different default — snapshotting
every env keeps both replayable).
- derive_votes_latest_unique() reproduces PG's RULE-maintained
votes_latest_unique from the stream (last vote per (pid, tid) in stream
order) — tested against a seeded PG table mimicking the RULE.
- delphi_storage/purge.py — purge_job(): deletes every generic-entity
partition for a job (the §9 GDPR path for snapshot copies).
- run_delphi.py — --snapshot-inputs flag (or DELPHI_SNAPSHOT_INPUTS=1):
captures before any stage runs, aborts the run on capture failure
(a run without recorded inputs defeats the point when enabled).
Off by default until P7's DELPHI_WRITE_MODE subsumes it.
Tests (TDD; tests/test_input_snapshots.py, RED first): 16 tests — raw
stream order/signs, compression, fingerprints, derivation-vs-PG-RULE,
full-row comments/participants, all-env math_main, SQL-parity-by-import,
purge, and the run_delphi hook (flag, env, off-by-default). PG-backed tests
use a seeded scratch database (skip-locally / fail-in-CI convention).
Full suite: 429 passed, 41 skipped, 58 xfailed (5 pre-existing
DynamoDB-unavailable errors unchanged); golden snapshots ran and passed —
math outputs untouched.
Stack: P1 #2597, P2 #2598, P3 #2599, P4 #2600, P5 #2601, P6a this commit.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
jucor
force-pushed
the
jc/delphi-storage-v2-p5
branch
from
July 6, 2026 18:48
95099f9 to
abac670
Compare
jucor
force-pushed
the
jc/delphi-storage-v2-p4
branch
from
July 6, 2026 22:52
5f9fbd5 to
efc507b
Compare
jucor
force-pushed
the
jc/delphi-storage-v2-p5
branch
from
July 6, 2026 22:52
abac670 to
df5baee
Compare
jucor
added a commit
that referenced
this pull request
Jul 6, 2026
… + purge tool
Implements the capture half of Stack 2 / P6 (docs/STORAGE_V2_DESIGN.md §4.2
entity 2, §4.4): reproducibility begins with recording EXACTLY what the
pipeline read. The read seam (--input-source, P6b) follows.
- delphi_storage/inputs.py — capture_run_inputs(store, job_id, zid, rid):
writes the six snapshot kinds as run_inputs items (codec envelopes,
zstd-compressed votes) and returns per-kind fingerprints (sha256 of the
canonical uncompressed payload, row_count, max vote created — the P7
manifest inputs). Key fidelity decisions, pinned by tests:
- votes = the RAW stream stage 1 consumes: ORDER BY created, raw PG signs
(the production math path never flips — fetch_votes()'s flip is dead
code), superseded votes included, order preserved (KMeans init depends
on vote-encounter order, INVESTIGATION_K_DIVERGENCE.md).
- The vote SQL is shared BY CONSTRUCTION: run_math_pipeline.py now imports
VOTES_COUNT_SQL / VOTES_BATCH_SQL from delphi_storage.inputs (strings
unchanged; golden suite + the e2e SQL pins stay green).
- comments/participants captured in FULL, including moderated-out rows
(mod = -1) and banned participants — filtering is pipeline behavior,
not snapshot behavior.
- clojure_math_main captured for ALL math_env rows (501 reads env-less
latest, 801 filters MATH_ENV with a different default — snapshotting
every env keeps both replayable).
- derive_votes_latest_unique() reproduces PG's RULE-maintained
votes_latest_unique from the stream (last vote per (pid, tid) in stream
order) — tested against a seeded PG table mimicking the RULE.
- delphi_storage/purge.py — purge_job(): deletes every generic-entity
partition for a job (the §9 GDPR path for snapshot copies).
- run_delphi.py — --snapshot-inputs flag (or DELPHI_SNAPSHOT_INPUTS=1):
captures before any stage runs, aborts the run on capture failure
(a run without recorded inputs defeats the point when enabled).
Off by default until P7's DELPHI_WRITE_MODE subsumes it.
Tests (TDD; tests/test_input_snapshots.py, RED first): 16 tests — raw
stream order/signs, compression, fingerprints, derivation-vs-PG-RULE,
full-row comments/participants, all-env math_main, SQL-parity-by-import,
purge, and the run_delphi hook (flag, env, off-by-default). PG-backed tests
use a seeded scratch database (skip-locally / fail-in-CI convention).
Full suite: 429 passed, 41 skipped, 58 xfailed (5 pre-existing
DynamoDB-unavailable errors unchanged); golden snapshots ran and passed —
math outputs untouched.
Stack: P1 #2597, P2 #2598, P3 #2599, P4 #2600, P5 #2601, P6a this commit.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Pull request overview
This PR advances Delphi Storage V2 provenance plumbing by ensuring a single pipeline job_id is explicitly threaded via --job-id across all pipeline entry points, while keeping DELPHI_JOB_ID as a transitional fallback. It introduces a shared resolver for consistent precedence (CLI > env > auto local-<uuid4>), updates orchestrators/pollers to pass the id on the command line, and adds focused unit tests to lock in the contract.
Changes:
- Add
delphi_storage.job_id.resolve_job_id()and adopt it across pipeline entry points andrun_delphi.py. - Thread
--job-idthrough the poller and all pipeline stage CLIs; exportDELPHI_JOB_IDinrun_delphi.pyfor legacy env readers. - Add
tests/test_job_id_threading.pyto validate precedence, subprocess command propagation, and stage--helpcoverage.
Reviewed changes
Copilot reviewed 11 out of 11 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| delphi/delphi_storage/job_id.py | Adds canonical job-id resolution logic and precedence. |
| delphi/run_delphi.py | Accepts/exports a resolved job id and passes it to all stage subprocesses. |
| delphi/scripts/job_poller.py | Centralizes per-job command construction and adds --job-id propagation. |
| delphi/polismath/run_math_pipeline.py | Adds --job-id plumbing and logs the resolved id. |
| delphi/umap_narrative/run_pipeline.py | Adds --job-id argument and resolves job id for the pipeline stage. |
| delphi/umap_narrative/reset_conversation.py | Introduces cli() and adds --job-id parsing/logging for the reset entry point. |
| delphi/umap_narrative/501_calculate_comment_extremity.py | Adds --job-id parsing/logging for stage 501. |
| delphi/umap_narrative/502_calculate_priorities.py | Adds --job-id parsing/logging for stage 502. |
| delphi/umap_narrative/700_datamapplot_for_layer.py | Threads job id into S3 key construction (with env/unknown fallback). |
| delphi/umap_narrative/801_narrative_report_batch.py | Accepts explicit --job-id with env fallback (no auto-gen) for narrative batch. |
| delphi/tests/test_job_id_threading.py | Adds unit tests covering job-id precedence and CLI threading invariants. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
36
to
+40
| parser.add_argument('--include_moderation', type=bool, default=False, help='Whether or not to include moderated comments in reports. If false, moderated comments will appear.') | ||
| parser.add_argument('--exclude_comment_selections', type=bool, default=True, help='Whether to exclude comments with selection=-1 in report_comment_selections table.') | ||
| parser.add_argument('--region', type=str, default='us-east-1', help='AWS region') | ||
| parser.add_argument('--job-id', dest='job_id', default=None, | ||
| help='Pipeline job id (auto local-<uuid4> when omitted; ' |
Comment on lines
1516
to
+1529
| parser.add_argument( | ||
| "--include_moderation", | ||
| type=bool, | ||
| default=False, | ||
| help="Whether or not to include moderated comments in reports. If false, moderated comments will appear.", | ||
| ) | ||
| parser.add_argument( | ||
| "--exclude_comment_selections", | ||
| type=bool, | ||
| default=True, | ||
| help="Whether to exclude comments with selection=-1 in report_comment_selections table.", | ||
| ) | ||
| parser.add_argument( | ||
| "--job-id", |
Comment on lines
191
to
+195
| parser.add_argument('--verbose', action='store_true', help='Show detailed output') | ||
| parser.add_argument('--include_moderation', type=bool, default=False, help='Whether or not to include moderated comments in reports. If false, moderated comments will appear.') | ||
| parser.add_argument('--exclude_comment_selections', type=bool, default=True, help='Whether to exclude comments with selection=-1 in report_comment_selections table.') | ||
| parser.add_argument('--job-id', dest='job_id', default=None, | ||
| help='Pipeline job id (Storage V2 provenance, design §4.4); defaults to DELPHI_JOB_ID env, else auto local-<uuid4>') |
Comment on lines
1526
to
1531
| help='Specific layer numbers to process (e.g., --layers 0 1 2). If not specified, all layers will be processed.') | ||
| parser.add_argument('--include_moderation', type=bool, default=False, help='Whether or not to include moderated comments in reports. If false, moderated comments will appear.') | ||
| parser.add_argument('--exclude_comment_selections', type=bool, default=True, help='Whether to exclude comments with selection=-1 in report_comment_selections table.') | ||
| parser.add_argument('--job-id', dest='job_id', default=None, | ||
| help='Pipeline job id (Storage V2 provenance, design §4.4); defaults to DELPHI_JOB_ID env; NO auto-generation here — narrative section keys embed it, so a missing id must keep failing loudly downstream') | ||
| args = parser.parse_args() |
jucor
force-pushed
the
jc/delphi-storage-v2-p4
branch
from
July 6, 2026 23:58
efc507b to
68aa10e
Compare
jucor
force-pushed
the
jc/delphi-storage-v2-p5
branch
from
July 6, 2026 23:58
df5baee to
90dc74d
Compare
… pipeline entry points Implements Stack 2 / P5 of docs/STORAGE_V2_DESIGN.md (§4.4): the pipeline job id travels ON THE COMMAND LINE instead of the env-var-only propagation that reached 2 of 18 tables (§3.2). - delphi_storage/job_id.py — resolve_job_id: explicit CLI value > DELPHI_JOB_ID env (transition fallback, removed in a later phase) > auto local-<uuid4> for dev runs. - run_delphi.py — gains --job-id, resolves once, exports DELPHI_JOB_ID for un-migrated env readers (same id as the command lines), and appends --job-id=<id> to all six stage subprocesses (reset, math, umap 500s, 501, 502, 700 per layer). - scripts/job_poller.py — command construction extracted into module-level build_job_command() (now unit-testable); FULL_PIPELINE and CREATE_NARRATIVE_BATCH commands carry --job-id=<queue job_id>; the DELPHI_JOB_ID env export stays for the transition window. 803's existing --job-id (the batch-tracking id, a distinct concept) is unchanged. - Stage entry points all accept --job-id: reset_conversation (CLI moved into cli(), main() stays the worker), run_math_pipeline (arg + log only — no math-logic change), run_pipeline (threaded into process_conversation, replacing the buried env read; auto prefix changes pipeline_run_<uuid> → local-<uuid> for bare dev runs), 501, 502 (arg + log, used from P7), 700 (threaded down to the S3-key builder; 'unknown' fallback preserved), 801 (arg > env, NO auto-generation — narrative section keys embed the id, so a missing one must keep failing loudly downstream). Tests (TDD; tests/test_job_id_threading.py, RED first): - resolve_job_id precedence; run_delphi threads one shared id to all six stage commands (explicit, auto, and env-fallback variants); - every stage entry point's --help advertises --job-id; - process_conversation exposes the job_id parameter; - poller commands for all three job types (FULL_PIPELINE + 801 carry the queue job id; 803 keeps batch_job_id semantics). Full suite: 424 passed, 30 skipped, 58 xfailed (5 pre-existing DynamoDB-unavailable errors unchanged); golden snapshots ran and passed — math outputs untouched. Stack: P1 #2597, P2 #2598, P3 #2599, P4 #2600, P5 this commit. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
jucor
force-pushed
the
jc/delphi-storage-v2-p5
branch
from
July 10, 2026 11:40
90dc74d to
cf3898c
Compare
jucor
force-pushed
the
jc/delphi-storage-v2-p4
branch
from
July 10, 2026 11:40
68aa10e to
926e37b
Compare
Delphi Coverage Report
|
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.
Stack 2 / P5 of the Storage V2 effort (design in #2597, §4.4): the pipeline job id now travels on the command line through every entry point, replacing the env-var-only propagation that reached 2 of 18 tables (§3.2). This is the provenance foundation P6 (input snapshots) and P7 (manifests + dual-write) build on.
Stacked on #2600 (
jc/delphi-storage-v2-p4).What this changes
delphi_storage/job_id.py—resolve_job_id: explicit CLI value >DELPHI_JOB_IDenv (kept as a transition fallback for one phase, per §4.4) > autolocal-<uuid4>for dev runs.run_delphi.py— gains--job-id, resolves once, exportsDELPHI_JOB_IDso un-migrated env readers inherit the same id as the command lines, and appends--job-id=<id>to all six stage subprocesses (reset, math, umap 500s, 501, 502, 700 per layer).scripts/job_poller.py— command construction extracted into a module-levelbuild_job_command()(now unit-testable);FULL_PIPELINEandCREATE_NARRATIVE_BATCHcommands carry--job-id=<queue job_id>; the env export stays during the transition. 803's existing--job-idis untouched — it's the batch-tracking id (batch_job_id), a distinct concept, documented as such.--job-id:reset_conversation.py— CLI moved from the__main__block intocli()(main()stays the worker function; behavior unchanged);polismath/run_math_pipeline.py— arg + log only, no math-logic change (golden suite ran and passed);run_pipeline.py— threaded intoprocess_conversation(job_id=…), replacing the env read buried at the call depth; the auto-id prefix for bare dev runs changespipeline_run_<uuid>→local-<uuid>;501/502— arg + log (consumed from P7 on);700_datamapplot_for_layer.py— threaded down to the S3-key builder ('unknown'fallback preserved for bare dev runs);801_narrative_report_batch.py— arg > env, deliberately no auto-generation: narrative section keys embed the id, so a missing one must keep failing loudly downstream.Testing (TDD — tests first, 1 collection error RED)
tests/test_job_id_threading.py:resolve_job_idprecedence (CLI / env / auto-distinct);run_delphi.pypasses one shared id to all six stage commands (explicit, auto-generated, and env-fallback variants), and exports it;--helpadvertises--job-id(real argparse, incl. the async 801 main);process_conversationexposes thejob_idparameter;build_job_commandfor all three job types (FULL_PIPELINE/801 carry the queue job id; 803 keepsbatch_job_idsemantics).Full suite: 424 passed, 30 skipped, 58 xfailed (5 pre-existing DynamoDB-unavailable errors unchanged). Golden snapshots ran and passed — math outputs untouched.
Stack
--input-sourceseam; golden-invariant), P7 manifests + dual-write, P8 LLM recorder + EVōC seeding🤖 Generated with Claude Code