fix(579): single authority for the first-stage online envelope + identity

[Muizzkolapo]

Summary

The first-stage envelope (payload → content.source + a derived source_guid) was authored four times — batch rows (_add_batch_metadata), batch text (_prepare_text_chunks_batch), online rows (json, inline), online text (inline) — and maintained by convention. The copies had already drifted, which is the root cause behind two observable bugs:

• Online .csv/.tsv/.xlsx shipped flat, unstamped rows (data_chunk = <loader output>). Post-573 a row without source_guid fails loud at write_source, so online tabular workflows were broken at the storage boundary.
• Online .json let a user column literally named source_guid hijack the framework identity (item.get("source_guid") or derive). Since source_guid is not a reserved staging name, a row {"source_guid": "x", ...} adopted "x" as its identity, collapsing distinct rows — the reserved-field collision 577 fixed, which the batch path never had.

Fix (enforcement, not another patched branch): one authority _envelope_row(payload) computes {content: {source: {**payload}}, source_guid: derive_source_guid(payload)} for every first-stage record. Online branches now only produce raw payloads and route through _wrap_online_rows; batch's two stamping sites are folded onto the same _envelope_row (behavior-preserving) and keep decorating it with ancestry. A new online file type can no longer regress to flat/unstamped by construction.

Task 1 decision (per spec) — always-derive, drop the honoring

Dropped the online-json item.get("source_guid") or so identity is always derived over the raw payload, matching batch. Verified no caller depends on the honoring: it was introduced in the 584 commit (not a resume feature), no loader stamps a first-stage row before ingestion, and no test asserts it. This both reconciles the batch↔online drift and closes the 577 collision on the online path.

Deviations / scope notes

• XML deferred to spec 589 (loader-contract root cause): XmlLoader.process returns ET.Element, not rows, so XML has nothing to wrap on either path — a distinct fix needing a fixture + row-shape decision. Online XML is left as-is (still loud-broken) and tracked by 589.
• Batch consolidation touches _add_batch_metadata / _prepare_text_chunks_batch — byte-identical output (same derive_source_guid call), guarded green by the existing golden-guid tests (865e85c1…, cac9d8de…).
• Removed an inaccurate in-code comment on the online csv/tsv branch that claimed tab-delimiter routing. Note (pre-existing, out of scope): TabularLoader uses csv.DictReader with the default comma delimiter, so .tsv currently parses as one column — a separate defect, not envelope-related.
• Docstring on _wrap_online_rows corrected after review (it over-specified downstream storage behavior).

Verification

• RED → GREEN: new test_online_row_envelope_chokepoint.py (7 tests) pinned the flat/unstamped csv+xlsx failures and the json source_guid-hijack collision; all pass after the fix.
• Full suite: 8119 passed; ruff format/check clean; mypy clean.
• Parity + determinism: an online .csv row, the same row via online .json, and via batch _add_batch_metadata produce the same content.source and the same source_guid; re-ingesting a .csv yields identical non-blank uuid5 guids. Reserved-named user columns (node_id, source_guid) survive under content.source and no longer hijack identity.
• Blast radius: all consumers of the three changed functions (10 test files incl. batch golden guids, online envelope shape, collision, tsv dispatch, loader contract) stay green.
• Smoke skipped — risk.sh: LOW, no smoke surface touched (change is input/preprocessing/staging/, not input/loaders/ or storage/). The "online tabular reaches write_source" claim is proven by composition: rows now carry a non-blank uuid5 source_guid, and write_source raises DataValidationError iff the guid is blank. The registry canary is orthogonal (UDF/reprompt discovery, untouched); online smoke-test hangs on the external LLM (environmental).
• Review: one blind fresh-context reviewer (LOW tier) → YES; confirmed batch byte-identity, no legitimate pre-stamped path for the dropped honoring, no missed siblings, non-tautological tests. The one finding (docstring inaccuracy) is fixed.

[Muizzkolapo]

Code review (max effort — 10 finder angles, 1-vote verify, gap sweep)

Ran the full pipeline: 10 parallel review angles (5 correctness + 3 cleanup + altitude + CLAUDE.md conventions), each candidate independently re-verified by a fresh skeptical reviewer, then a gap sweep for anything the first pass missed. 5 candidates were checked and cleanly refuted (dropped source_guid-honoring risk, a task_preparer.py comment, unified.py's independent guid derivation, batch ancestry-decoration duplication, and one dead-code nit) — omitted below. 8 survived verification.

The core fix (single-authority envelope construction, closing the reserved-field collision) is sound and well-tested. The main risk: _envelope_row now routes every online row through json.dumps-based hashing for the first time, and a few realistic inputs weren't accounted for in that path.

1. Online .xlsx + date columns crashes ingestion (new)

initial_pipeline.py#L571

_wrap_online_rows → _envelope_row → IDGenerator.derive_source_guid → bare json.dumps(payload, sort_keys=True) with no fallback encoder. An .xlsx file with a date/datetime column produces pandas.Timestamp values via pd.read_excel(...).to_dict(orient="records"), which crash json.dumps with TypeError: Object of type Timestamp is not JSON serializable (confirmed via live repro against the actual pinned pandas version). A sheet with a bare-numeric column header independently crashes the same call sorting mixed int/str keys. Online .xlsx never called derive_source_guid before this diff, so neither crash mode existed for the online path previously — this PR's own goal of fixing online xlsx ingestion still fails for realistically dated spreadsheets.

2. .tsv online ingestion flips from a loud crash to silent data corruption

initial_pipeline.py#L566

TabularLoader.process (tabular.py:40) always parses via comma-delimited csv.DictReader, ignoring .tsv's extension. A real tab-separated row collapses into one garbage column, e.g. {'id\tbody\tnode_id': '1\thello\tUSER_A'}. Pre-diff, this unstamped row reached write_source and raised a loud DataValidationError (confirmed against the old code path). Post-diff, _envelope_row successfully hashes the garbled dict and write_source persists it cleanly with a valid-looking source_guid — no error, no signal the data is corrupted (confirmed against a live SQLite backend). Neither test_staging_tsv_dispatch.py nor the new chokepoint test file has a .tsv fixture that would catch this. Note: batch mode already has this exact silent-corruption behavior today (pre-existing, unchanged) — this diff makes online match batch rather than inventing a new pattern, but it's still a real loud→silent regression for online mode specifically.

3. Online .csv/.tsv ragged rows crash ingestion (new)

initial_pipeline.py#L566

A data row with more fields than the header makes csv.DictReader's default restkey=None inject a None dict key (e.g. header a,b, row 1,2,3,4 → {'a': '1', 'b': '2', None: ['3', '4']}). json.dumps(payload, sort_keys=True) inside derive_source_guid then raises TypeError: '<' not supported between instances of 'NoneType' and 'str' sorting the keys (confirmed via live repro, full traceback through the actual call chain), crashing ingestion for the entire file. This crash mode is new online-mode exposure — pre-diff, online csv/tsv never called derive_source_guid.

4. _wrap_online_rows's return type lies about non-dict passthrough

initial_pipeline.py#L359

Declared -> list[dict[str, Any]], but a non-dict item in a .json array (e.g. ["just a string", {"a": 1}]) passes through unchanged. It reaches write_source's item.get("source_guid") and raises an unhandled AttributeError: 'str' object has no attribute 'get' rather than the graceful DataValidationError the (false) type signature implies is the only failure mode. This crash-reachability predates the diff for .json, but the diff gives it a newly-misleading type signature (mypy reports no issue) and extends the same untyped helper to the csv/xlsx branches.

5. _prepare_json_batch's single-object branch stays unwrapped (out of scope, informational)

initial_pipeline.py#L413

A batch-mode .json file whose top level is a single object (not a list) still bypasses _envelope_row entirely — no content.source wrap, no source_guid. TaskPreparer._normalize_input then falls back to deriving over the whole item including the per-run-random target_id/batch_id, so re-running the same input yields a different source_guid every run, silently breaking dedup/checkpoint matching (confirmed via live trace). This is explicitly listed as out-of-scope in specs/active/579-first-stage-tabular-source-guid.md line 60 ("the existing deferred item on the 584 row") — not a functional ask of this PR. Only nit: the PR body discloses the XML deferral explicitly but omits this equally-deferred item.

6. Reserved-name collision test only covers .json, not .csv/.xlsx

test_online_row_envelope_chokepoint.py#L233

test_online_json_source_guid_column_does_not_hijack_identity is the only regression guard for the source_guid-column-hijack class this PR fixes. No live bug today (_envelope_row is format-agnostic), but a future change that special-cases one branch would ship silently green for csv/xlsx.

7. _prepare_text_chunks_batch duplicates _add_batch_metadata's loop (cleanup)

initial_pipeline.py#L391

Now structurally identical to _add_batch_metadata's loop except for the payload. Every other batch branch already delegates to _add_batch_metadata; this is the sole holdout. return _add_batch_metadata([{"content": c} for c in chunks], batch_id, node_id) produces identical output and removes ~14 duplicated lines.

8. data_chunk/src_text aliasing repeated 4x (minor cleanup)

initial_pipeline.py#L552

data_chunk = _wrap_online_rows(X); src_text = data_chunk repeats near-identically across all 4 online branches. Could collapse to data_chunk = src_text = _wrap_online_rows(X) per branch with zero behavior change.

---

Automated review — findings 1-4 are correctness bugs worth blocking on before merge; 5-6 are informational/test-coverage; 7-8 are optional cleanup.

[Muizzkolapo]

Review addressed (verify-first on each finding, then fixed at root cause)

Pushed follow-up commits. Full suite 8130 pass, ruff + mypy clean; each fix went RED→GREEN and the identity-function change was re-verified by three independent blind reviewers (the first two caught real collision bugs in my initial attempts — see #4).

#	Finding	Resolution
1a	.xlsx date column crashes json.dumps	Fixed — generate_content_hash uses default=str, so a Timestamp/date cell renders deterministically. Never fires for JSON-native payloads → golden guids unchanged.
1b / 3	numeric header / ragged-csv None restkey crash	Fixed by rejection, not encoding (see #4) — generate_content_hash now raises a clear DataValidationError on any non-string field name instead of an opaque TypeError. Such a name is unusable downstream ({{ source.<name> }}) and has no stable identity.
2	.tsv loud→silent corruption	Fixed at root cause — TabularLoader now selects delimiter="\t" on the .tsv extension, so tab files parse into real columns (fixes batch tsv too; clears the pre-existing defect).
4	_wrap_online_rows return type lies	Fixed — annotated list[Any] (a non-dict passthrough item can remain).
4 (identity)	robust hashing of odd keys	My first two attempts introduced silent dedup collisions (str(k) collapse; then a \x00-prefix namespace overlap) — both caught by blind review. Final design rejects non-string keys, which is collision-free by construction.
5	batch single-object JSON stays unwrapped → nondeterministic guid	Confirmed real, deferred — explicitly out-of-scope in the spec ("the single-object-JSON / non-list batch edge … separate deferred item on the 584 row"). Disclosing here as requested; not silently dropped.
6	collision guard only covered .json	Fixed — added .csv and .xlsx source_guid-column guards.
7	_prepare_text_chunks_batch duplicated the loop	Fixed — it now delegates to _add_batch_metadata (batch's single row authority).
8	data_chunk/src_text aliasing repeated 4×	Fixed — collapsed to data_chunk = src_text = _wrap_online_rows(...).

Note on the final reviewer's out-of-scope observation: default=str makes a datetime value and its string form hash alike. This is new in this PR (not pre-existing), but benign: a given file is parsed by one loader consistently (a csv row's date is already a string), and cross-file dedup is keyed on (relative_path, source_guid), so distinct files can't silently merge.

Smoke: skipped — the smoke project isn't provisioned in this clone (branch-independent), and the touched surfaces (TabularLoader delimiter, generate_content_hash) are covered offline by unit tests + the full green suite; the online agac run is env-blocked on the external LLM.