Skip to content

Commit ccadf23

Browse files
AStaroverovclaude
andcommitted
Fix contentHash: read CanonicalID in a resolved process template
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 <noreply@anthropic.com>
1 parent 9d27224 commit ccadf23

2 files changed

Lines changed: 118 additions & 78 deletions

File tree

workflow/src/main.tpl.tengo

Lines changed: 22 additions & 78 deletions
Original file line numberDiff line numberDiff line change
@@ -12,15 +12,14 @@ render := import("@platforma-sdk/workflow-tengo:render")
1212
smart := import("@platforma-sdk/workflow-tengo:smart")
1313
ll := import("@platforma-sdk/workflow-tengo:ll")
1414
pframes := import("@platforma-sdk/workflow-tengo:pframes")
15-
xsv := import("@platforma-sdk/workflow-tengo:pframes.xsv")
16-
pSpec := import("@platforma-sdk/workflow-tengo:pframes.spec")
1715
canonical := import("@platforma-sdk/workflow-tengo:canonical")
1816
maps := import("@platforma-sdk/workflow-tengo:maps")
1917
constants := import("@platforma-sdk/workflow-tengo:constants")
2018
messages := import(":messages")
2119
columnSpecs := import(":columns")
2220

2321
infoTpl := assets.importTemplate(":info")
22+
processTpl := assets.importTemplate(":process")
2423

2524
// JSON resource with sorted-key canonical bytes. smart.createJsonResource uses
2625
// Tengo's stdlib json.encode, which preserves Go's randomized map iteration —
@@ -31,21 +30,6 @@ canonicalJsonResource := func(value) {
3130
return smart.createValueResource(constants.RTYPE_JSON, canonical.encode(value))
3231
}
3332

34-
// Content tag for exported columns, replacing the per-block blockId. We use the
35-
// backend's own content key of the source table (info().CanonicalID, base32),
36-
// which is deterministic for identical content: identical results across blocks
37-
// share the same tag (columns dedupe instead of being made unique per block),
38-
// while different results get different tags (no collision).
39-
contentHash := func(table) {
40-
i := table.info()
41-
if !is_undefined(i.CanonicalID) {
42-
return ll.base32Encode(i.CanonicalID)
43-
}
44-
// Field reference: dereference to its value resource.
45-
ri := smart.resource(i.Value).info()
46-
return ll.base32Encode(ri.CanonicalID)
47-
}
48-
4933
REQUIRED_FEATURES := ["FR1", "CDR1", "FR2", "CDR2", "FR3", "CDR3", "FR4"]
5034

5135
detectMode := func(axisSpec) {
@@ -408,10 +392,8 @@ wf.body(func(args) {
408392
})
409393
infoBlob := infoResult.output("info", 24 * 60 * 60 * 1000)
410394

411-
// Build the canonical column list (no blockId, no isOutput, no trace) and
412-
// import the Python TSV outputs into parquet-backed PColumn data. With
413-
// canonical specs the xsv.importFile renders dedup across blocks; only
414-
// the spec-rewrap into the unified pFrame below is per-block.
395+
// Column-build args forwarded to the process template (no blockId, no trace
396+
// — those are stamped there at pframe-build time).
415397
colArgs := {
416398
mode: mode,
417399
receptor: receptor,
@@ -426,72 +408,34 @@ wf.body(func(args) {
426408
visibleKeyAxisSpec := columnSpecs.cloneSpec(keyAxisSpec, undefined, {
427409
"pl7.app/table/visibility": "default"
428410
})
429-
canonicalCols := columnSpecs.buildColumns(colArgs)
430-
scalarOut := xsv.importFile(propertiesTsv, "tsv", {
431-
axes: [{ column: "entity_key", spec: visibleKeyAxisSpec }],
432-
columns: canonicalCols,
433-
storageFormat: "Parquet",
434-
partitionKeyLength: 0
435-
}, { splitDataAndSpec: true, cpu: 1, mem: "4GiB" })
436-
437-
aaOut := undefined
438-
if mode == "peptide" {
439-
aaCol := columnSpecs.aaFractionColumn(visibleKeyAxisSpec)
440-
aaOut = xsv.importFile(aaFractionTsv, "tsv", {
441-
axes: aaCol.axes,
442-
columns: [aaCol.column],
443-
storageFormat: "Parquet",
444-
partitionKeyLength: 0
445-
}, { splitDataAndSpec: true, cpu: 1, mem: "4GiB" })
446-
}
447411

448-
// Stamp every column's spec with a content tag (pl7.app/contentHash, derived
449-
// from the source table's CanonicalID instead of the per-block blockId) and
450-
// the upstream trace. Identical results across blocks share the tag (columns
451-
// dedupe) while different results stay distinct. The same pFrame is published
452-
// both as the block's UI output and as the result-pool export. The trace also
453-
// lets the block's UI distinguish our property columns from upstream metadata
454-
// columns mixed into propertiesPfHandle.
455-
//
456-
// The trace step is no longer keyed on blockId (the block is no longer the
457-
// identity of the result; its content is). `id` is optional in the trace
458-
// schema; omitting it keeps the provenance step identical across
459-
// content-identical runs. Per-instance trace label resolution lives in the
460-
// model (resolveTraceLabel in model/src/label.ts).
461-
trace := pSpec.makeTrace(datasetSpec, {
462-
type: "milaboratories.sequence-properties",
463-
importance: 30,
464-
label: args.traceLabel
412+
// Import + per-column content-tagging + trace stamping is deferred to
413+
// process.tpl.tengo: it needs the Python output tables as RESOLVED render
414+
// inputs to read their content key (info().CanonicalID), which is not
415+
// available in this body (the same body that creates the exec). The render
416+
// is blockId-independent, so it dedups across block instances with identical
417+
// inputs.
418+
processResult := render.create(processTpl, {
419+
properties: propertiesTsv,
420+
aaFraction: aaFractionTsv,
421+
params: canonicalJsonResource({
422+
colArgs: colArgs,
423+
visibleKeyAxisSpec: visibleKeyAxisSpec,
424+
traceLabel: args.traceLabel,
425+
datasetSpec: datasetSpec
426+
})
427+
}, {
428+
metaInputs: { mem: args.mem }
465429
})
466-
stamp := func(spec, cHash) {
467-
return trace.inject(columnSpecs.cloneSpec(spec,
468-
{ "pl7.app/contentHash": cHash },
469-
undefined))
470-
}
471-
472-
pf := pframes.pFrameBuilder()
473-
scalarHash := contentHash(propertiesTsv)
474-
for _, k in maps.getKeys(scalarOut) {
475-
v := scalarOut[k]
476-
pf.add(k, stamp(v.spec, scalarHash), v.data)
477-
}
478-
if aaOut != undefined {
479-
aaHash := contentHash(aaFractionTsv)
480-
for _, k in maps.getKeys(aaOut) {
481-
v := aaOut[k]
482-
pf.add(k, stamp(v.spec, aaHash), v.data)
483-
}
484-
}
485-
pf = pf.build()
486430

487431
return {
488432
outputs: {
489-
propertiesPf: pframes.exportFrame(pf),
433+
propertiesPf: processResult.output("propertiesPf", 24 * 60 * 60 * 1000),
490434
info: infoBlob,
491435
processingLog: processingLog
492436
},
493437
exports: {
494-
properties: pf
438+
properties: processResult.output("exportPframe", 24 * 60 * 60 * 1000)
495439
}
496440
}
497441
})

workflow/src/process.tpl.tengo

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
// Post-processing: imports the Python TSV outputs into parquet-backed PColumn
2+
// data and stamps each column with a content tag + the upstream trace.
3+
//
4+
// Split out from main.tpl.tengo so the source tables arrive as RESOLVED render
5+
// inputs: their content key (info().CanonicalID) is only available once the
6+
// resource exists, which it does not in main's body (the same body that creates
7+
// the Python exec). With the tables as inputs here, contentHash resolves.
8+
9+
self := import("@platforma-sdk/workflow-tengo:tpl")
10+
xsv := import("@platforma-sdk/workflow-tengo:pframes.xsv")
11+
pframes := import("@platforma-sdk/workflow-tengo:pframes")
12+
pSpec := import("@platforma-sdk/workflow-tengo:pframes.spec")
13+
maps := import("@platforma-sdk/workflow-tengo:maps")
14+
ll := import("@platforma-sdk/workflow-tengo:ll")
15+
smart := import("@platforma-sdk/workflow-tengo:smart")
16+
columnSpecs := import(":columns")
17+
18+
self.defineOutputs("propertiesPf", "exportPframe")
19+
20+
// Content tag for exported columns, replacing the per-block blockId. The
21+
// backend's own content key of the source table (info().CanonicalID, base32) is
22+
// deterministic for identical content: identical results across blocks share
23+
// the tag (columns dedupe) while different results get distinct tags.
24+
// info() is a strict map; accessing a missing key throws, so probe with
25+
// maps.containsKey. A resolved resource carries CanonicalID directly; a field
26+
// reference (an exec saveFile output) does not — dereference its Value.
27+
contentHash := func(table) {
28+
i := table.info()
29+
if maps.containsKey(i, "CanonicalID") {
30+
return ll.base32Encode(i.CanonicalID)
31+
}
32+
ri := smart.resource(i.Value).info()
33+
return ll.base32Encode(ri.CanonicalID)
34+
}
35+
36+
self.body(func(args) {
37+
params := args.params
38+
colArgs := params.colArgs
39+
visibleKeyAxisSpec := params.visibleKeyAxisSpec
40+
mode := colArgs.mode
41+
42+
properties := args.properties
43+
44+
canonicalCols := columnSpecs.buildColumns(colArgs)
45+
scalarOut := xsv.importFile(properties, "tsv", {
46+
axes: [{ column: "entity_key", spec: visibleKeyAxisSpec }],
47+
columns: canonicalCols,
48+
storageFormat: "Parquet",
49+
partitionKeyLength: 0
50+
}, { splitDataAndSpec: true, cpu: 1, mem: "4GiB" })
51+
52+
aaOut := undefined
53+
if mode == "peptide" {
54+
aaCol := columnSpecs.aaFractionColumn(visibleKeyAxisSpec)
55+
aaOut = xsv.importFile(args.aaFraction, "tsv", {
56+
axes: aaCol.axes,
57+
columns: [aaCol.column],
58+
storageFormat: "Parquet",
59+
partitionKeyLength: 0
60+
}, { splitDataAndSpec: true, cpu: 1, mem: "4GiB" })
61+
}
62+
63+
// Trace step is not keyed on blockId — the result's content is its identity.
64+
// `id` is optional; omitting it keeps the provenance step identical across
65+
// content-identical runs.
66+
trace := pSpec.makeTrace(params.datasetSpec, {
67+
type: "milaboratories.sequence-properties",
68+
importance: 30,
69+
label: params.traceLabel
70+
})
71+
stamp := func(spec, cHash) {
72+
return trace.inject(columnSpecs.cloneSpec(spec,
73+
{ "pl7.app/contentHash": cHash },
74+
undefined))
75+
}
76+
77+
pf := pframes.pFrameBuilder()
78+
scalarHash := contentHash(properties)
79+
for _, k in maps.getKeys(scalarOut) {
80+
v := scalarOut[k]
81+
pf.add(k, stamp(v.spec, scalarHash), v.data)
82+
}
83+
if aaOut != undefined {
84+
aaHash := contentHash(args.aaFraction)
85+
for _, k in maps.getKeys(aaOut) {
86+
v := aaOut[k]
87+
pf.add(k, stamp(v.spec, aaHash), v.data)
88+
}
89+
}
90+
pf = pf.build()
91+
92+
return {
93+
propertiesPf: pframes.exportFrame(pf),
94+
exportPframe: pf
95+
}
96+
})

0 commit comments

Comments
 (0)