Validation patterns and compliance checking for Kailash SDK including parameter validation, DataFlow pattern validation, connection validation, absolute import validation, workflow structure...
Validation patterns and compliance checking for Kailash SDK development.
results["node_id"]["result"] (not .result)grep -I for binary skip, \.egg-info/ exclusion shape/redteam audit; sub-package collection-gate patterns; same-shard sweep §4a| Validation | Catches | Key Pattern |
|---|---|---|
| Parameters | Missing/wrong-type params | Check before workflow.build() |
| Connections | Wrong 4-param format, nonexistent nodes | (src_id, src_param, tgt_id, tgt_param) |
| Workflow | Duplicate IDs, dead-ends, no entry point | Structural integrity |
| DataFlow | .result access, UUID conversion |
results["id"]["result"] |
| Imports | Relative imports, circular deps | Absolute imports only |
| Security | Hardcoded secrets, SQL injection | Env vars, parameterized queries |
from kailash.validation import WorkflowValidator
validator = WorkflowValidator(workflow)
results = validator.validate_all()
if not results.is_valid:
for error in results.errors:
print(f"Error: {error}")
.build() before executepython -m kailash.validation.cli validate-all
python -m kailash.validation.cli check-security
Some failure modes are invisible at Tier 1 unit and Tier 2 integration because each step in a multi-step pipeline works in isolation — the break is at the seam between steps. A pipeline like km.train(df) → km.register(result, name=...) can have 100% unit coverage on km.train AND 100% integration coverage on km.register while the chain fails because result is missing a field the next step needs (the "fake-integration" failure mode).
Defense: A release-blocking regression tier above Tier 3 — a test that executes the full pipeline against real infrastructure AND asserts a deterministic fingerprint over the output. Any change that alters observable behavior flips the fingerprint and blocks release.
# DO — release-blocking regression with pinned SHA-256 fingerprint
@pytest.mark.regression
@pytest.mark.release_blocking
async def test_readme_quick_start_end_to_end(real_conn):
# 1. Execute the README Quick Start verbatim
result = await km.train(df, target="churned")
registered = await km.register(result, name="demo")
# 2. Compute fingerprint over deterministic output
fingerprint = hashlib.sha256(
json.dumps(registered.artifact_uris, sort_keys=True).encode()
).hexdigest()
# 3. Assert against pinned value (pinned in spec)
assert fingerprint == "c962060cf467cc732df355ec9e1212cfb0d7534a3eed4480b511adad5a9ceb00"
# DO NOT — rely on unit + integration alone
def test_km_train_unit(): ... # passes, km.train works in isolation
def test_km_register_integration(): ... # passes, km.register works given a valid result
# Chain still breaks because result is missing `.trainable` back-reference.
A() → B(A's output) → C(B's output) is the user's primary ergonomickailash.ml facade → dataflow.ml → nexus.ml handoff)Session 2026-04-23 kailash-ml 1.0.0 M1 W33b shard — regression test caught the km.train → km.register trainable-field gap (commit 15033fa6) that unit tests couldn't. Pinned fingerprint: specs/ml-engines-v2.md §16.3. See skills/34-kailash-ml/m1-release-wave.md § "Release-Blocking README Quick Start Regression".
gold-standards-validator - Compliance checkingpattern-expert - Pattern validationtesting-specialist - Test validation