From 9d27224c017080549da39ea96e5230ff4e2c68f3 Mon Sep 17 00:00:00 2001 From: AStaroverov Date: Fri, 12 Jun 2026 11:45:49 +0200 Subject: [PATCH 1/3] Namespace exported columns by content hash instead of blockId Stamp pl7.app/contentHash (base32 of the source table's CanonicalID) on the exported property columns instead of the per-block blockId, and drop the blockId-keyed trace step id. Identical results across blocks/projects now produce content-identical columns that dedupe downstream, while different results stay distinct. Mirrors the humanization-score approach. The Python step and Parquet import already deduped; this extends it to the export. Co-Authored-By: Claude Fable 5 --- .changeset/content-hash-namespacing.md | 5 +++ test/src/wf.test.ts | 4 +- workflow/src/columns.lib.tengo | 8 ++-- workflow/src/main.tpl.tengo | 55 ++++++++++++++++---------- 4 files changed, 46 insertions(+), 26 deletions(-) create mode 100644 .changeset/content-hash-namespacing.md diff --git a/.changeset/content-hash-namespacing.md b/.changeset/content-hash-namespacing.md new file mode 100644 index 0000000..da3fdc1 --- /dev/null +++ b/.changeset/content-hash-namespacing.md @@ -0,0 +1,5 @@ +--- +"@platforma-open/milaboratories.sequence-properties.workflow": patch +--- + +Namespace exported property columns by a content hash of the source table (pl7.app/contentHash, derived from the backend CanonicalID) instead of the per-block blockId, and drop the blockId-keyed trace step id. Identical results across blocks/projects now produce content-identical columns that dedupe downstream instead of being made unique per block; different results stay distinct. The heavy Python step and Parquet import already deduped; this extends dedup to the exported pframe identity. diff --git a/test/src/wf.test.ts b/test/src/wf.test.ts index 4719d14..6706372 100644 --- a/test/src/wf.test.ts +++ b/test/src/wf.test.ts @@ -127,7 +127,7 @@ describe('edge cases', () => { describe('downstream consumption', () => { it.todo('score columns are discoverable by a downstream block via the result pool'); it.todo('output PColumns carry pl7.app/trace stamped with this block\'s label'); - it.todo('export PFrame stamps blockId on score column domains'); + it.todo('export PFrame stamps pl7.app/contentHash on score column domains'); }); // --------------------------------------------------------------------------- @@ -160,6 +160,6 @@ describe('dedup', () => { describe('cross-block composition', () => { it.todo('canary: real MiXCR fastq → sequence-properties detects sc IG and emits scores'); it.todo('end-to-end: peptide-extraction → sequence-properties → lead-selection'); - it.todo('two sequence-properties instances in one project disambiguate via blockId'); + it.todo('two sequence-properties instances on identical input share columns via pl7.app/contentHash'); it.todo('VDJ chain: full-coverage MiXCR → seq-properties → lead-selection'); }); diff --git a/workflow/src/columns.lib.tengo b/workflow/src/columns.lib.tengo index ae56e05..7f985f6 100644 --- a/workflow/src/columns.lib.tengo +++ b/workflow/src/columns.lib.tengo @@ -6,13 +6,13 @@ // Returns `[{column, id, naRegex, // allowNA, spec}, ...]` with feature- // specific domains and annotations. -// No blockId, no isOutput, no trace +// No content tag, no isOutput, no trace // — those are stamped on by the caller // at pframe-build time. // - `aaFractionColumn()` — 2-axis AA-fraction column descriptor // (peptide mode only). // - `cloneSpec(spec, dExtras, aExtras)` — spec-cloning helper used by the -// caller to layer blockId / isOutput / +// caller to layer contentHash / isOutput / // any other per-consumer overrides. // // `args` shape: @@ -70,7 +70,7 @@ makeCol := func(tsvCol, valName, valueType, label, domain, annotations) { } // --------------------------------------------------------------------------- -// Canonical column lists per scope. No `toPlot`, no `blockId` — the public +// Canonical column lists per scope. No `toPlot`, no content tag — the public // getters layer those on per consumer. // --------------------------------------------------------------------------- @@ -386,7 +386,7 @@ buildColumns := func(args) { // --------------------------------------------------------------------------- // Spec-cloning helper. Builds a fresh spec dict with optional domain and // annotation extras. Callers use this to layer per-consumer overrides -// (e.g. `pl7.app/blockId` in domain, `pl7.app/isOutput` in annotations). +// (e.g. `pl7.app/contentHash` in domain, `pl7.app/isOutput` in annotations). // --------------------------------------------------------------------------- cloneSpec := func(spec, domainExtras, annotationExtras) { diff --git a/workflow/src/main.tpl.tengo b/workflow/src/main.tpl.tengo index 603fcf7..c27fea8 100644 --- a/workflow/src/main.tpl.tengo +++ b/workflow/src/main.tpl.tengo @@ -31,6 +31,21 @@ canonicalJsonResource := func(value) { return smart.createValueResource(constants.RTYPE_JSON, canonical.encode(value)) } +// Content tag for exported columns, replacing the per-block blockId. We use the +// backend's own content key of the source table (info().CanonicalID, base32), +// which is deterministic for identical content: identical results across blocks +// share the same tag (columns dedupe instead of being made unique per block), +// while different results get different tags (no collision). +contentHash := func(table) { + i := table.info() + if !is_undefined(i.CanonicalID) { + return ll.base32Encode(i.CanonicalID) + } + // Field reference: dereference to its value resource. + ri := smart.resource(i.Value).info() + return ll.base32Encode(ri.CanonicalID) +} + REQUIRED_FEATURES := ["FR1", "CDR1", "FR2", "CDR2", "FR3", "CDR3", "FR4"] detectMode := func(axisSpec) { @@ -138,7 +153,6 @@ wf.prepare(func(args) { }) wf.body(func(args) { - blockId := wf.blockId().getDataAsJson() bundle := args.columns datasetSpec := bundle.getSpec(args.inputAnchor) @@ -431,40 +445,41 @@ wf.body(func(args) { }, { splitDataAndSpec: true, cpu: 1, mem: "4GiB" }) } - // Stamp every column's spec with `pl7.app/blockId` in domain (for run - // attribution) and the upstream trace (so downstream blocks see the full - // provenance). The same pFrame is published both as the block's UI output - // and as the result-pool export — one canonical resource, two consumers. - // The trace is also how the block's own UI distinguishes our property - // columns from upstream metadata columns mixed into propertiesPfHandle. - // Per-instance trace label. Resolution lives in the model - // (resolveTraceLabel in model/src/label.ts) — the model projects a single - // string into args so the workflow does not know about customBlockLabel - // vs defaultBlockLabel vs the static fallback. The fallback chain - // itself — customBlockLabel || defaultBlockLabel || "Sequence Properties" - // — matches clonotype-clustering and titeseq-analysis PR #13; only the - // location of the resolution differs (model here, inline-Tengo there). + // Stamp every column's spec with a content tag (pl7.app/contentHash, derived + // from the source table's CanonicalID instead of the per-block blockId) and + // the upstream trace. Identical results across blocks share the tag (columns + // dedupe) while different results stay distinct. The same pFrame is published + // both as the block's UI output and as the result-pool export. The trace also + // lets the block's UI distinguish our property columns from upstream metadata + // columns mixed into propertiesPfHandle. + // + // The trace step is no longer keyed on blockId (the block is no longer the + // identity of the result; its content is). `id` is optional in the trace + // schema; omitting it keeps the provenance step identical across + // content-identical runs. Per-instance trace label resolution lives in the + // model (resolveTraceLabel in model/src/label.ts). trace := pSpec.makeTrace(datasetSpec, { type: "milaboratories.sequence-properties", importance: 30, - label: args.traceLabel, - id: blockId + label: args.traceLabel }) - stamp := func(spec) { + stamp := func(spec, cHash) { return trace.inject(columnSpecs.cloneSpec(spec, - { "pl7.app/blockId": blockId }, + { "pl7.app/contentHash": cHash }, undefined)) } pf := pframes.pFrameBuilder() + scalarHash := contentHash(propertiesTsv) for _, k in maps.getKeys(scalarOut) { v := scalarOut[k] - pf.add(k, stamp(v.spec), v.data) + pf.add(k, stamp(v.spec, scalarHash), v.data) } if aaOut != undefined { + aaHash := contentHash(aaFractionTsv) for _, k in maps.getKeys(aaOut) { v := aaOut[k] - pf.add(k, stamp(v.spec), v.data) + pf.add(k, stamp(v.spec, aaHash), v.data) } } pf = pf.build() From ccadf237520cd0ab5853a5277ed879fca8eba6ee Mon Sep 17 00:00:00 2001 From: AStaroverov Date: Fri, 12 Jun 2026 13:44:22 +0200 Subject: [PATCH 2/3] Fix contentHash: read CanonicalID in a resolved process template MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit info().CanonicalID is only readable once the table resource is resolved, which it is not in main's body (the same body that creates the Python exec — the saveFile output field has Value==0, so smart.resource(i.Value) errors with "null resource id"). Move the TSV import + content-tagging + trace stamping into a new process.tpl.tengo that receives the Python output tables as resolved render inputs. Also probe info() with maps.containsKey (it is a strict map; accessing a missing key throws). Co-Authored-By: Claude Fable 5 --- workflow/src/main.tpl.tengo | 100 ++++++++------------------------- workflow/src/process.tpl.tengo | 96 +++++++++++++++++++++++++++++++ 2 files changed, 118 insertions(+), 78 deletions(-) create mode 100644 workflow/src/process.tpl.tengo diff --git a/workflow/src/main.tpl.tengo b/workflow/src/main.tpl.tengo index c27fea8..3dc19ce 100644 --- a/workflow/src/main.tpl.tengo +++ b/workflow/src/main.tpl.tengo @@ -12,8 +12,6 @@ render := import("@platforma-sdk/workflow-tengo:render") smart := import("@platforma-sdk/workflow-tengo:smart") ll := import("@platforma-sdk/workflow-tengo:ll") pframes := import("@platforma-sdk/workflow-tengo:pframes") -xsv := import("@platforma-sdk/workflow-tengo:pframes.xsv") -pSpec := import("@platforma-sdk/workflow-tengo:pframes.spec") canonical := import("@platforma-sdk/workflow-tengo:canonical") maps := import("@platforma-sdk/workflow-tengo:maps") constants := import("@platforma-sdk/workflow-tengo:constants") @@ -21,6 +19,7 @@ messages := import(":messages") columnSpecs := import(":columns") infoTpl := assets.importTemplate(":info") +processTpl := assets.importTemplate(":process") // JSON resource with sorted-key canonical bytes. smart.createJsonResource uses // Tengo's stdlib json.encode, which preserves Go's randomized map iteration — @@ -31,21 +30,6 @@ canonicalJsonResource := func(value) { return smart.createValueResource(constants.RTYPE_JSON, canonical.encode(value)) } -// Content tag for exported columns, replacing the per-block blockId. We use the -// backend's own content key of the source table (info().CanonicalID, base32), -// which is deterministic for identical content: identical results across blocks -// share the same tag (columns dedupe instead of being made unique per block), -// while different results get different tags (no collision). -contentHash := func(table) { - i := table.info() - if !is_undefined(i.CanonicalID) { - return ll.base32Encode(i.CanonicalID) - } - // Field reference: dereference to its value resource. - ri := smart.resource(i.Value).info() - return ll.base32Encode(ri.CanonicalID) -} - REQUIRED_FEATURES := ["FR1", "CDR1", "FR2", "CDR2", "FR3", "CDR3", "FR4"] detectMode := func(axisSpec) { @@ -408,10 +392,8 @@ wf.body(func(args) { }) infoBlob := infoResult.output("info", 24 * 60 * 60 * 1000) - // Build the canonical column list (no blockId, no isOutput, no trace) and - // import the Python TSV outputs into parquet-backed PColumn data. With - // canonical specs the xsv.importFile renders dedup across blocks; only - // the spec-rewrap into the unified pFrame below is per-block. + // Column-build args forwarded to the process template (no blockId, no trace + // — those are stamped there at pframe-build time). colArgs := { mode: mode, receptor: receptor, @@ -426,72 +408,34 @@ wf.body(func(args) { visibleKeyAxisSpec := columnSpecs.cloneSpec(keyAxisSpec, undefined, { "pl7.app/table/visibility": "default" }) - canonicalCols := columnSpecs.buildColumns(colArgs) - scalarOut := xsv.importFile(propertiesTsv, "tsv", { - axes: [{ column: "entity_key", spec: visibleKeyAxisSpec }], - columns: canonicalCols, - storageFormat: "Parquet", - partitionKeyLength: 0 - }, { splitDataAndSpec: true, cpu: 1, mem: "4GiB" }) - - aaOut := undefined - if mode == "peptide" { - aaCol := columnSpecs.aaFractionColumn(visibleKeyAxisSpec) - aaOut = xsv.importFile(aaFractionTsv, "tsv", { - axes: aaCol.axes, - columns: [aaCol.column], - storageFormat: "Parquet", - partitionKeyLength: 0 - }, { splitDataAndSpec: true, cpu: 1, mem: "4GiB" }) - } - // Stamp every column's spec with a content tag (pl7.app/contentHash, derived - // from the source table's CanonicalID instead of the per-block blockId) and - // the upstream trace. Identical results across blocks share the tag (columns - // dedupe) while different results stay distinct. The same pFrame is published - // both as the block's UI output and as the result-pool export. The trace also - // lets the block's UI distinguish our property columns from upstream metadata - // columns mixed into propertiesPfHandle. - // - // The trace step is no longer keyed on blockId (the block is no longer the - // identity of the result; its content is). `id` is optional in the trace - // schema; omitting it keeps the provenance step identical across - // content-identical runs. Per-instance trace label resolution lives in the - // model (resolveTraceLabel in model/src/label.ts). - trace := pSpec.makeTrace(datasetSpec, { - type: "milaboratories.sequence-properties", - importance: 30, - label: args.traceLabel + // Import + per-column content-tagging + trace stamping is deferred to + // process.tpl.tengo: it needs the Python output tables as RESOLVED render + // inputs to read their content key (info().CanonicalID), which is not + // available in this body (the same body that creates the exec). The render + // is blockId-independent, so it dedups across block instances with identical + // inputs. + processResult := render.create(processTpl, { + properties: propertiesTsv, + aaFraction: aaFractionTsv, + params: canonicalJsonResource({ + colArgs: colArgs, + visibleKeyAxisSpec: visibleKeyAxisSpec, + traceLabel: args.traceLabel, + datasetSpec: datasetSpec + }) + }, { + metaInputs: { mem: args.mem } }) - stamp := func(spec, cHash) { - return trace.inject(columnSpecs.cloneSpec(spec, - { "pl7.app/contentHash": cHash }, - undefined)) - } - - pf := pframes.pFrameBuilder() - scalarHash := contentHash(propertiesTsv) - for _, k in maps.getKeys(scalarOut) { - v := scalarOut[k] - pf.add(k, stamp(v.spec, scalarHash), v.data) - } - if aaOut != undefined { - aaHash := contentHash(aaFractionTsv) - for _, k in maps.getKeys(aaOut) { - v := aaOut[k] - pf.add(k, stamp(v.spec, aaHash), v.data) - } - } - pf = pf.build() return { outputs: { - propertiesPf: pframes.exportFrame(pf), + propertiesPf: processResult.output("propertiesPf", 24 * 60 * 60 * 1000), info: infoBlob, processingLog: processingLog }, exports: { - properties: pf + properties: processResult.output("exportPframe", 24 * 60 * 60 * 1000) } } }) diff --git a/workflow/src/process.tpl.tengo b/workflow/src/process.tpl.tengo new file mode 100644 index 0000000..134e678 --- /dev/null +++ b/workflow/src/process.tpl.tengo @@ -0,0 +1,96 @@ +// Post-processing: imports the Python TSV outputs into parquet-backed PColumn +// data and stamps each column with a content tag + the upstream trace. +// +// Split out from main.tpl.tengo so the source tables arrive as RESOLVED render +// inputs: their content key (info().CanonicalID) is only available once the +// resource exists, which it does not in main's body (the same body that creates +// the Python exec). With the tables as inputs here, contentHash resolves. + +self := import("@platforma-sdk/workflow-tengo:tpl") +xsv := import("@platforma-sdk/workflow-tengo:pframes.xsv") +pframes := import("@platforma-sdk/workflow-tengo:pframes") +pSpec := import("@platforma-sdk/workflow-tengo:pframes.spec") +maps := import("@platforma-sdk/workflow-tengo:maps") +ll := import("@platforma-sdk/workflow-tengo:ll") +smart := import("@platforma-sdk/workflow-tengo:smart") +columnSpecs := import(":columns") + +self.defineOutputs("propertiesPf", "exportPframe") + +// Content tag for exported columns, replacing the per-block blockId. The +// backend's own content key of the source table (info().CanonicalID, base32) is +// deterministic for identical content: identical results across blocks share +// the tag (columns dedupe) while different results get distinct tags. +// info() is a strict map; accessing a missing key throws, so probe with +// maps.containsKey. A resolved resource carries CanonicalID directly; a field +// reference (an exec saveFile output) does not — dereference its Value. +contentHash := func(table) { + i := table.info() + if maps.containsKey(i, "CanonicalID") { + return ll.base32Encode(i.CanonicalID) + } + ri := smart.resource(i.Value).info() + return ll.base32Encode(ri.CanonicalID) +} + +self.body(func(args) { + params := args.params + colArgs := params.colArgs + visibleKeyAxisSpec := params.visibleKeyAxisSpec + mode := colArgs.mode + + properties := args.properties + + canonicalCols := columnSpecs.buildColumns(colArgs) + scalarOut := xsv.importFile(properties, "tsv", { + axes: [{ column: "entity_key", spec: visibleKeyAxisSpec }], + columns: canonicalCols, + storageFormat: "Parquet", + partitionKeyLength: 0 + }, { splitDataAndSpec: true, cpu: 1, mem: "4GiB" }) + + aaOut := undefined + if mode == "peptide" { + aaCol := columnSpecs.aaFractionColumn(visibleKeyAxisSpec) + aaOut = xsv.importFile(args.aaFraction, "tsv", { + axes: aaCol.axes, + columns: [aaCol.column], + storageFormat: "Parquet", + partitionKeyLength: 0 + }, { splitDataAndSpec: true, cpu: 1, mem: "4GiB" }) + } + + // Trace step is not keyed on blockId — the result's content is its identity. + // `id` is optional; omitting it keeps the provenance step identical across + // content-identical runs. + trace := pSpec.makeTrace(params.datasetSpec, { + type: "milaboratories.sequence-properties", + importance: 30, + label: params.traceLabel + }) + stamp := func(spec, cHash) { + return trace.inject(columnSpecs.cloneSpec(spec, + { "pl7.app/contentHash": cHash }, + undefined)) + } + + pf := pframes.pFrameBuilder() + scalarHash := contentHash(properties) + for _, k in maps.getKeys(scalarOut) { + v := scalarOut[k] + pf.add(k, stamp(v.spec, scalarHash), v.data) + } + if aaOut != undefined { + aaHash := contentHash(args.aaFraction) + for _, k in maps.getKeys(aaOut) { + v := aaOut[k] + pf.add(k, stamp(v.spec, aaHash), v.data) + } + } + pf = pf.build() + + return { + propertiesPf: pframes.exportFrame(pf), + exportPframe: pf + } +}) From de5c45f81411723705a32b5b97fd6d2833f5a5fa Mon Sep 17 00:00:00 2001 From: AStaroverov Date: Thu, 18 Jun 2026 17:02:52 +0200 Subject: [PATCH 3/3] refactor: replace blockId with contentHash for exported columns and update related specs --- .changeset/content-hash-namespacing.md | 5 -- test/src/wf.test.ts | 4 +- workflow/src/columns.lib.tengo | 8 +-- workflow/src/main.tpl.tengo | 85 +++++++++++++++++------ workflow/src/process.tpl.tengo | 96 -------------------------- 5 files changed, 69 insertions(+), 129 deletions(-) delete mode 100644 .changeset/content-hash-namespacing.md delete mode 100644 workflow/src/process.tpl.tengo diff --git a/.changeset/content-hash-namespacing.md b/.changeset/content-hash-namespacing.md deleted file mode 100644 index da3fdc1..0000000 --- a/.changeset/content-hash-namespacing.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -"@platforma-open/milaboratories.sequence-properties.workflow": patch ---- - -Namespace exported property columns by a content hash of the source table (pl7.app/contentHash, derived from the backend CanonicalID) instead of the per-block blockId, and drop the blockId-keyed trace step id. Identical results across blocks/projects now produce content-identical columns that dedupe downstream instead of being made unique per block; different results stay distinct. The heavy Python step and Parquet import already deduped; this extends dedup to the exported pframe identity. diff --git a/test/src/wf.test.ts b/test/src/wf.test.ts index 6706372..4719d14 100644 --- a/test/src/wf.test.ts +++ b/test/src/wf.test.ts @@ -127,7 +127,7 @@ describe('edge cases', () => { describe('downstream consumption', () => { it.todo('score columns are discoverable by a downstream block via the result pool'); it.todo('output PColumns carry pl7.app/trace stamped with this block\'s label'); - it.todo('export PFrame stamps pl7.app/contentHash on score column domains'); + it.todo('export PFrame stamps blockId on score column domains'); }); // --------------------------------------------------------------------------- @@ -160,6 +160,6 @@ describe('dedup', () => { describe('cross-block composition', () => { it.todo('canary: real MiXCR fastq → sequence-properties detects sc IG and emits scores'); it.todo('end-to-end: peptide-extraction → sequence-properties → lead-selection'); - it.todo('two sequence-properties instances on identical input share columns via pl7.app/contentHash'); + it.todo('two sequence-properties instances in one project disambiguate via blockId'); it.todo('VDJ chain: full-coverage MiXCR → seq-properties → lead-selection'); }); diff --git a/workflow/src/columns.lib.tengo b/workflow/src/columns.lib.tengo index 7f985f6..ae56e05 100644 --- a/workflow/src/columns.lib.tengo +++ b/workflow/src/columns.lib.tengo @@ -6,13 +6,13 @@ // Returns `[{column, id, naRegex, // allowNA, spec}, ...]` with feature- // specific domains and annotations. -// No content tag, no isOutput, no trace +// No blockId, no isOutput, no trace // — those are stamped on by the caller // at pframe-build time. // - `aaFractionColumn()` — 2-axis AA-fraction column descriptor // (peptide mode only). // - `cloneSpec(spec, dExtras, aExtras)` — spec-cloning helper used by the -// caller to layer contentHash / isOutput / +// caller to layer blockId / isOutput / // any other per-consumer overrides. // // `args` shape: @@ -70,7 +70,7 @@ makeCol := func(tsvCol, valName, valueType, label, domain, annotations) { } // --------------------------------------------------------------------------- -// Canonical column lists per scope. No `toPlot`, no content tag — the public +// Canonical column lists per scope. No `toPlot`, no `blockId` — the public // getters layer those on per consumer. // --------------------------------------------------------------------------- @@ -386,7 +386,7 @@ buildColumns := func(args) { // --------------------------------------------------------------------------- // Spec-cloning helper. Builds a fresh spec dict with optional domain and // annotation extras. Callers use this to layer per-consumer overrides -// (e.g. `pl7.app/contentHash` in domain, `pl7.app/isOutput` in annotations). +// (e.g. `pl7.app/blockId` in domain, `pl7.app/isOutput` in annotations). // --------------------------------------------------------------------------- cloneSpec := func(spec, domainExtras, annotationExtras) { diff --git a/workflow/src/main.tpl.tengo b/workflow/src/main.tpl.tengo index 3dc19ce..603fcf7 100644 --- a/workflow/src/main.tpl.tengo +++ b/workflow/src/main.tpl.tengo @@ -12,6 +12,8 @@ render := import("@platforma-sdk/workflow-tengo:render") smart := import("@platforma-sdk/workflow-tengo:smart") ll := import("@platforma-sdk/workflow-tengo:ll") pframes := import("@platforma-sdk/workflow-tengo:pframes") +xsv := import("@platforma-sdk/workflow-tengo:pframes.xsv") +pSpec := import("@platforma-sdk/workflow-tengo:pframes.spec") canonical := import("@platforma-sdk/workflow-tengo:canonical") maps := import("@platforma-sdk/workflow-tengo:maps") constants := import("@platforma-sdk/workflow-tengo:constants") @@ -19,7 +21,6 @@ messages := import(":messages") columnSpecs := import(":columns") infoTpl := assets.importTemplate(":info") -processTpl := assets.importTemplate(":process") // JSON resource with sorted-key canonical bytes. smart.createJsonResource uses // Tengo's stdlib json.encode, which preserves Go's randomized map iteration — @@ -137,6 +138,7 @@ wf.prepare(func(args) { }) wf.body(func(args) { + blockId := wf.blockId().getDataAsJson() bundle := args.columns datasetSpec := bundle.getSpec(args.inputAnchor) @@ -392,8 +394,10 @@ wf.body(func(args) { }) infoBlob := infoResult.output("info", 24 * 60 * 60 * 1000) - // Column-build args forwarded to the process template (no blockId, no trace - // — those are stamped there at pframe-build time). + // Build the canonical column list (no blockId, no isOutput, no trace) and + // import the Python TSV outputs into parquet-backed PColumn data. With + // canonical specs the xsv.importFile renders dedup across blocks; only + // the spec-rewrap into the unified pFrame below is per-block. colArgs := { mode: mode, receptor: receptor, @@ -408,34 +412,71 @@ wf.body(func(args) { visibleKeyAxisSpec := columnSpecs.cloneSpec(keyAxisSpec, undefined, { "pl7.app/table/visibility": "default" }) + canonicalCols := columnSpecs.buildColumns(colArgs) + scalarOut := xsv.importFile(propertiesTsv, "tsv", { + axes: [{ column: "entity_key", spec: visibleKeyAxisSpec }], + columns: canonicalCols, + storageFormat: "Parquet", + partitionKeyLength: 0 + }, { splitDataAndSpec: true, cpu: 1, mem: "4GiB" }) + + aaOut := undefined + if mode == "peptide" { + aaCol := columnSpecs.aaFractionColumn(visibleKeyAxisSpec) + aaOut = xsv.importFile(aaFractionTsv, "tsv", { + axes: aaCol.axes, + columns: [aaCol.column], + storageFormat: "Parquet", + partitionKeyLength: 0 + }, { splitDataAndSpec: true, cpu: 1, mem: "4GiB" }) + } - // Import + per-column content-tagging + trace stamping is deferred to - // process.tpl.tengo: it needs the Python output tables as RESOLVED render - // inputs to read their content key (info().CanonicalID), which is not - // available in this body (the same body that creates the exec). The render - // is blockId-independent, so it dedups across block instances with identical - // inputs. - processResult := render.create(processTpl, { - properties: propertiesTsv, - aaFraction: aaFractionTsv, - params: canonicalJsonResource({ - colArgs: colArgs, - visibleKeyAxisSpec: visibleKeyAxisSpec, - traceLabel: args.traceLabel, - datasetSpec: datasetSpec - }) - }, { - metaInputs: { mem: args.mem } + // Stamp every column's spec with `pl7.app/blockId` in domain (for run + // attribution) and the upstream trace (so downstream blocks see the full + // provenance). The same pFrame is published both as the block's UI output + // and as the result-pool export — one canonical resource, two consumers. + // The trace is also how the block's own UI distinguishes our property + // columns from upstream metadata columns mixed into propertiesPfHandle. + // Per-instance trace label. Resolution lives in the model + // (resolveTraceLabel in model/src/label.ts) — the model projects a single + // string into args so the workflow does not know about customBlockLabel + // vs defaultBlockLabel vs the static fallback. The fallback chain + // itself — customBlockLabel || defaultBlockLabel || "Sequence Properties" + // — matches clonotype-clustering and titeseq-analysis PR #13; only the + // location of the resolution differs (model here, inline-Tengo there). + trace := pSpec.makeTrace(datasetSpec, { + type: "milaboratories.sequence-properties", + importance: 30, + label: args.traceLabel, + id: blockId }) + stamp := func(spec) { + return trace.inject(columnSpecs.cloneSpec(spec, + { "pl7.app/blockId": blockId }, + undefined)) + } + + pf := pframes.pFrameBuilder() + for _, k in maps.getKeys(scalarOut) { + v := scalarOut[k] + pf.add(k, stamp(v.spec), v.data) + } + if aaOut != undefined { + for _, k in maps.getKeys(aaOut) { + v := aaOut[k] + pf.add(k, stamp(v.spec), v.data) + } + } + pf = pf.build() return { outputs: { - propertiesPf: processResult.output("propertiesPf", 24 * 60 * 60 * 1000), + propertiesPf: pframes.exportFrame(pf), info: infoBlob, processingLog: processingLog }, exports: { - properties: processResult.output("exportPframe", 24 * 60 * 60 * 1000) + properties: pf } } }) diff --git a/workflow/src/process.tpl.tengo b/workflow/src/process.tpl.tengo deleted file mode 100644 index 134e678..0000000 --- a/workflow/src/process.tpl.tengo +++ /dev/null @@ -1,96 +0,0 @@ -// Post-processing: imports the Python TSV outputs into parquet-backed PColumn -// data and stamps each column with a content tag + the upstream trace. -// -// Split out from main.tpl.tengo so the source tables arrive as RESOLVED render -// inputs: their content key (info().CanonicalID) is only available once the -// resource exists, which it does not in main's body (the same body that creates -// the Python exec). With the tables as inputs here, contentHash resolves. - -self := import("@platforma-sdk/workflow-tengo:tpl") -xsv := import("@platforma-sdk/workflow-tengo:pframes.xsv") -pframes := import("@platforma-sdk/workflow-tengo:pframes") -pSpec := import("@platforma-sdk/workflow-tengo:pframes.spec") -maps := import("@platforma-sdk/workflow-tengo:maps") -ll := import("@platforma-sdk/workflow-tengo:ll") -smart := import("@platforma-sdk/workflow-tengo:smart") -columnSpecs := import(":columns") - -self.defineOutputs("propertiesPf", "exportPframe") - -// Content tag for exported columns, replacing the per-block blockId. The -// backend's own content key of the source table (info().CanonicalID, base32) is -// deterministic for identical content: identical results across blocks share -// the tag (columns dedupe) while different results get distinct tags. -// info() is a strict map; accessing a missing key throws, so probe with -// maps.containsKey. A resolved resource carries CanonicalID directly; a field -// reference (an exec saveFile output) does not — dereference its Value. -contentHash := func(table) { - i := table.info() - if maps.containsKey(i, "CanonicalID") { - return ll.base32Encode(i.CanonicalID) - } - ri := smart.resource(i.Value).info() - return ll.base32Encode(ri.CanonicalID) -} - -self.body(func(args) { - params := args.params - colArgs := params.colArgs - visibleKeyAxisSpec := params.visibleKeyAxisSpec - mode := colArgs.mode - - properties := args.properties - - canonicalCols := columnSpecs.buildColumns(colArgs) - scalarOut := xsv.importFile(properties, "tsv", { - axes: [{ column: "entity_key", spec: visibleKeyAxisSpec }], - columns: canonicalCols, - storageFormat: "Parquet", - partitionKeyLength: 0 - }, { splitDataAndSpec: true, cpu: 1, mem: "4GiB" }) - - aaOut := undefined - if mode == "peptide" { - aaCol := columnSpecs.aaFractionColumn(visibleKeyAxisSpec) - aaOut = xsv.importFile(args.aaFraction, "tsv", { - axes: aaCol.axes, - columns: [aaCol.column], - storageFormat: "Parquet", - partitionKeyLength: 0 - }, { splitDataAndSpec: true, cpu: 1, mem: "4GiB" }) - } - - // Trace step is not keyed on blockId — the result's content is its identity. - // `id` is optional; omitting it keeps the provenance step identical across - // content-identical runs. - trace := pSpec.makeTrace(params.datasetSpec, { - type: "milaboratories.sequence-properties", - importance: 30, - label: params.traceLabel - }) - stamp := func(spec, cHash) { - return trace.inject(columnSpecs.cloneSpec(spec, - { "pl7.app/contentHash": cHash }, - undefined)) - } - - pf := pframes.pFrameBuilder() - scalarHash := contentHash(properties) - for _, k in maps.getKeys(scalarOut) { - v := scalarOut[k] - pf.add(k, stamp(v.spec, scalarHash), v.data) - } - if aaOut != undefined { - aaHash := contentHash(args.aaFraction) - for _, k in maps.getKeys(aaOut) { - v := aaOut[k] - pf.add(k, stamp(v.spec, aaHash), v.data) - } - } - pf = pf.build() - - return { - propertiesPf: pframes.exportFrame(pf), - exportPframe: pf - } -})