feat(payload)!: switch pipeline wire format from TSV to JSON, v0.3.0 (AGG-116)#6
Closed
anthrax63 wants to merge 1 commit into
Closed
Conversation
The lookup pipeline's `publish_to_session` was `type: file` on a TSV
Arrow file, which DDM does not support — `task_executors/io_operation/
publish_to_session.rs` errors "Publish file in session is not supported".
Only `type: json` is supported, so the pipeline (see aggregion/kz-scoring
AGG-116 PR) now emits its result as a JSON array of row objects and
extracts it via `type: json / extract single_value: true` from a single-
row `payload_json` cell produced by the decrypt task.
This PR replaces the API-side parser to match:
- `src/kz_scoring_api/payload.py` new — `parse_pipeline_result(str) →
list[dict]`. Empty payload / JSON `null` / `[]` all fold to `[]`
(which LookupService._shape then maps to `null` per the v0.2.1
not-found contract, so /single and /multi shapes are unchanged).
- `src/kz_scoring_api/tsv.py` removed. `LookupService._run_one` now
calls `parse_pipeline_result` instead of `parse_tsv`.
- Defensive `_maybe_double_decode`: DDM's `Extract single_value` on a
Utf8 cell serializes as a JSON string, and depending on how vaultee-
pipelines relays DDM's `session.result_json` field, the payload may
arrive wrapped in one extra JSON-string layer. Being defensive on
both is cheaper than pinning the interop precisely — a mis-shape
here would surface as a 500 to the caller, not silent data
corruption.
Tests
- `tests/test_payload.py` replaces `tests/test_tsv.py`; covers empty /
whitespace / `null` / `[]` / single object / array / bare object /
double-encoded string / UTF-8 BOM / invalid JSON / unexpected
top-level type / non-object row.
- `tests/test_lookup.py` and `tests/test_endpoints.py` — every TSV
payload replaced with `json.dumps(...)`. New coverage:
- `test_lookup_iin_only_empty_array_returns_none` — pipeline
fast-path on 0-row CH-lookup writes "[]", API surfaces as null.
- `test_lookup_double_encoded_payload_is_unwrapped` — defensive
layer 2 decoding.
Wire contract change (breaking on the API↔pipeline boundary — outer
`/single` and `/multi` JSON contract is unchanged, callers not
affected). Bump: 0.2.1 → 0.3.0.
Not in scope
- The pipeline templates + decrypt script rewrite lives in the
aggregion/kz-scoring PR (AGG-116). This PR only touches the client.
- DDM executor fix — AGG-115. e2e cannot be run until the executor
starts claiming tasks again; unit + integration-mock tests all
pass here (59/59).
Refs: AGG-116, AGG-113 (parent), AGG-115 (blocker for e2e).
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Co-authored-by: multica-agent <github@multica.ai>
Member
Author
|
Closing — overengineered. Реальная причина UnicodeDecodeError в decrypt_row — отсутствие extract task перед python-таской в 4 lookup-шаблонах (тот же паттерн что мы уже дважды применяли — AGG-95, AGG-99). Wire format менять не нужно; достаточно перед каждым |
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.
Summary
DDM's
publish_to_sessiononly supportstype: json(seeddm/crates/task_executors/io_operation/src/publish_to_session.rs), so the lookup pipeline inaggregion/kz-scoringPR (#27) now emits its result as a JSON array of row objects extracted viasingle_value: truefrom apayload_jsoncell.This PR replaces the API-side parser to match. Outer
/single//multiJSON contract is unchanged — callers not affected.src/kz_scoring_api/payload.pynew —parse_pipeline_result(str) → list[dict]. Empty payload /null/[]all fold to[], whichLookupService._shapethen maps tonullper the v0.2.1 not-found contract.src/kz_scoring_api/tsv.pyremoved.LookupService._run_onenow callsparse_pipeline_resultinstead ofparse_tsv._maybe_double_decode— DDM'sExtract single_valueon a Utf8 cell serializes as a JSON string, and depending on how vaultee-pipelines relays DDM'ssession.result_json, the payload may arrive wrapped in one extra JSON-string layer. Being defensive on both is cheaper than pinning the interop precisely.test_payload.pyreplacestest_tsv.py(empty / whitespace /null/[]/ single object / array / bare object / double-encoded string / UTF-8 BOM / invalid JSON / unexpected top-level / non-object row);test_lookup.pyandtest_endpoints.pyupdated withjson.dumpspayloads; new coverage for 0-row fast-path ("[]"→null) and defensive layer-2 decoding.0.2.1 → 0.3.0(wire format change on the API↔pipeline boundary).Refs: AGG-116, AGG-113 (parent, provides v0.2.1 null-contract this builds on), AGG-115 (blocker for e2e).
Test plan
pytest -q→ 59/59 passed (was 50/50 in PR feat(lookup)!: return null for not-found, v0.2.1 (AGG-113) #5; net +9 tests covering the new JSON contract).ruff check src/ tests/— clean.aggregion/kz-scoringPR #27.v0.3.0вkartel-deploy/argo/overlays/vaultee/kz-scoring-api.yaml, coordinated with the kz-scoring template rollout.🤖 Generated with Claude Code