From bddb2aef4dda52c7d65e828909999bd572467d90 Mon Sep 17 00:00:00 2001 From: Nelson Parente Date: Tue, 19 May 2026 21:47:40 +0100 Subject: [PATCH 1/5] feat(python): add workflow context propagation quickstart sample Ports Cassie Coyle's Go-SDK reference example (dapr/go-sdk#823) to Python. Demonstrates PropagationScope.LINEAGE and PropagationScope.OWN_HISTORY in a fraud-detection payment scenario with a 3-level workflow hierarchy. Requires Dapr 1.18+ (dapr/dapr#9810) and dapr-ext-workflow 1.18+ (dapr/python-sdk#1025). Signed-off-by: Nelson Parente --- .../python/sdk-context-propagation/README.md | 152 ++++++ .../python/sdk-context-propagation/dapr.yaml | 7 + .../order-processor/app.py | 440 ++++++++++++++++++ .../order-processor/requirements.txt | 2 + 4 files changed, 601 insertions(+) create mode 100644 workflows/python/sdk-context-propagation/README.md create mode 100644 workflows/python/sdk-context-propagation/dapr.yaml create mode 100644 workflows/python/sdk-context-propagation/order-processor/app.py create mode 100644 workflows/python/sdk-context-propagation/order-processor/requirements.txt diff --git a/workflows/python/sdk-context-propagation/README.md b/workflows/python/sdk-context-propagation/README.md new file mode 100644 index 000000000..a4ce5368c --- /dev/null +++ b/workflows/python/sdk-context-propagation/README.md @@ -0,0 +1,152 @@ +# Dapr Workflow — Context Propagation (Python SDK) + +This quickstart demonstrates **workflow history propagation**, a new feature in Dapr 1.18 that lets a parent workflow share its execution history with child workflows and activities. Downstream services can inspect that history to make trust-aware decisions — without any external state store or custom messaging. + +> **Runtime requirement**: Dapr 1.18+ ([dapr/dapr#9810](https://github.com/dapr/dapr/pull/9810)) +> **SDK requirement**: `dapr-ext-workflow>=1.18.0rc0` ([dapr/python-sdk#1025](https://github.com/dapr/python-sdk/pull/1025)) +> **Proposal**: [dapr/proposals#102](https://github.com/dapr/proposals/issues/102) + +## What is workflow context propagation? + +When a parent workflow calls a child workflow or activity it can optionally attach a tamper-evident snapshot of its own execution history. The receiver reads that snapshot via `ctx.get_propagated_history()` and queries it by workflow name and activity name — letting it verify that the correct upstream steps ran before it proceeds. + +### Two propagation modes + +| Mode | Constant | What the receiver sees | +|------|----------|----------------------| +| **Own history** | `PropagationScope.OWN_HISTORY` | Only the direct caller's events | +| **Lineage** | `PropagationScope.LINEAGE` | Caller's events **plus** any ancestor history the caller itself received | + +## Scenario: Credit-card payment with fraud detection + +``` +MerchantCheckout (root) + └─ validate_merchant (activity, no propagation) + └─ ProcessPayment (child wf, LINEAGE) + └─ validate_card (activity, no propagation) + └─ check_spending_limits (activity, no propagation) + └─ FraudDetection (grandchild wf, LINEAGE) + | reads MerchantCheckout/validate_merchant + | ProcessPayment/validate_card + | ProcessPayment/check_spending_limits + └─ settle_payment (activity, OWN_HISTORY) + reads ProcessPayment events only +``` + +`FraudDetection` uses `PropagationScope.LINEAGE` to see the **full ancestor chain** — it can verify both the merchant validation (performed by the grandparent) and the card/limit checks (performed by the parent) before approving the transaction. + +`settle_payment` uses `PropagationScope.OWN_HISTORY` to see only the **direct caller's events** — a trust-boundary mode that limits visibility to what `ProcessPayment` itself executed. + +## Python API surface + +```python +# Parent workflow — propagate LINEAGE when calling a child workflow +result = yield ctx.call_child_workflow( + fraud_detection, + input=req_json, + propagation=wf.PropagationScope.LINEAGE, +) + +# Parent workflow — propagate OWN_HISTORY when calling an activity +settlement = yield ctx.call_activity( + settle_payment, + input=req_json, + propagation=wf.PropagationScope.OWN_HISTORY, +) + +# Child workflow (or activity) — read the propagated history +history = ctx.get_propagated_history() # returns PropagatedHistory | None + +if history is not None: + process_wf = history.get_workflow_by_name('ProcessPayment') # raises PropagationNotFoundError if missing + card_act = process_wf.get_activity_by_name('validate_card') + print(card_act.completed) # bool + print(card_act.output) # JSON string +``` + +Key types exported from `dapr.ext.workflow`: +- `PropagationScope` — enum with `LINEAGE` and `OWN_HISTORY` +- `PropagatedHistory` — top-level history object; call `.get_workflows()` or `.get_workflow_by_name(name)` +- `WorkflowResult` — per-workflow slice; call `.get_activity_by_name(name)` or `.get_child_workflow_by_name(name)` +- `ActivityResult` — has `.completed`, `.output` fields +- `PropagationNotFoundError` — raised when a named workflow/activity is not in the history + +## Prerequisites + +- [Dapr CLI](https://docs.dapr.io/getting-started/install-dapr-cli/) 1.18+ +- Dapr runtime 1.18+ initialized (`dapr init`) +- Python 3.9+ +- Redis (started automatically by `dapr init`) + +## Run the sample + +```sh +cd workflows/python/sdk-context-propagation + +# Install dependencies +pip3 install -r order-processor/requirements.txt + +# Run with Dapr +dapr run -f . +``` + +## Expected output + +``` +============================================ += WORKFLOW HISTORY PROPAGATION DEMO = +============================================ + + Flow: MerchantCheckout -> validate_merchant + -> ProcessPayment (child wf, LINEAGE) + -> validate_card -> check_spending_limits + -> FraudDetection (child wf, LINEAGE) <-- sees MerchantCheckout + ProcessPayment events + -> settle_payment (activity, OWN_HISTORY) <-- sees only ProcessPayment events + + [main] Started workflow instance: checkout-001 + [MerchantCheckout] Starting checkout for merchant merchant-abc + [MerchantCheckout] Step 1: validate_merchant (no propagation) + [ValidateMerchant] Validating merchant merchant-abc + [MerchantCheckout] Step 1 complete: merchant valid + [MerchantCheckout] Step 2: ProcessPayment child wf (PropagationScope.LINEAGE) + [ProcessPayment] Starting payment ****4242 149.99 USD + [ProcessPayment] Step 1: validate_card (no propagation) + [ValidateCard] Validating card ****4242 (propagated history: none) + [ProcessPayment] Step 1 complete: card valid + [ProcessPayment] Step 2: check_spending_limits (no propagation) + [CheckSpendingLimits] Checking 149.99 USD (propagated history: none) + [ProcessPayment] Step 2 complete: within limits + [ProcessPayment] Step 3: FraudDetection child wf (PropagationScope.LINEAGE) + [FraudDetection] Received propagated history with workflows: ['MerchantCheckout', 'ProcessPayment'] + [FraudDetection] Verification: + MerchantCheckout/validate_merchant: completed=True + ProcessPayment/validate_card: completed=True + ProcessPayment/check_spending_limits: completed=True + [FraudDetection] APPROVED (risk=0.10) + [ProcessPayment] Step 3 complete: fraud check passed (risk=0.10) + [ProcessPayment] Step 4: settle_payment (PropagationScope.OWN_HISTORY) + [SettlePayment] Propagated workflows: ['ProcessPayment'] + [SettlePayment] SETTLED: txn-merchant-abc-1748000000000 + [ProcessPayment] Step 4 complete: settled (txn=txn-merchant-abc-...) + [ProcessPayment] COMPLETE: payment settled: ... + [MerchantCheckout] COMPLETE: payment settled: ... + [main] Workflow completed! Output: "payment settled: ..." + +======================== += COMPLETE = +======================== +``` + +## Stop the sample + +```sh +dapr stop -f . +``` + +## References + +- [Proposal: Workflow History Propagation (dapr/proposals#102)](https://github.com/dapr/proposals/issues/102) +- [Runtime PR: dapr/dapr#9810](https://github.com/dapr/dapr/pull/9810) +- [Python SDK PR: dapr/python-sdk#1025](https://github.com/dapr/python-sdk/pull/1025) +- [Go SDK reference: dapr/go-sdk#823](https://github.com/dapr/go-sdk/pull/823) +- [Dapr Workflow documentation](https://docs.dapr.io/developing-applications/building-blocks/workflow/) diff --git a/workflows/python/sdk-context-propagation/dapr.yaml b/workflows/python/sdk-context-propagation/dapr.yaml new file mode 100644 index 000000000..e05735188 --- /dev/null +++ b/workflows/python/sdk-context-propagation/dapr.yaml @@ -0,0 +1,7 @@ +version: 1 +common: + resourcesPath: ../../components +apps: + - appID: order-processor + appDirPath: ./order-processor/ + command: ["python3", "app.py"] diff --git a/workflows/python/sdk-context-propagation/order-processor/app.py b/workflows/python/sdk-context-propagation/order-processor/app.py new file mode 100644 index 000000000..be2abac78 --- /dev/null +++ b/workflows/python/sdk-context-propagation/order-processor/app.py @@ -0,0 +1,440 @@ +# -*- coding: utf-8 -*- +# Copyright 2026 The Dapr Authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Workflow history propagation quickstart (Python SDK). + +Scenario: credit-card payment processing with fraud detection. + +Flow: + MerchantCheckout (root workflow) + └─ ValidateMerchant (activity, no propagation) + └─ ProcessPayment (child workflow, PropagationScope.LINEAGE) + └─ ValidateCard (activity, no propagation) + └─ CheckSpendingLimits (activity, no propagation) + └─ FraudDetection (child workflow, PropagationScope.LINEAGE) + | reads: MerchantCheckout/ValidateMerchant + | ProcessPayment/ValidateCard + | ProcessPayment/CheckSpendingLimits + └─ SettlePayment (activity, PropagationScope.OWN_HISTORY) + reads: ProcessPayment events only (no ancestor chain) + +This requires Dapr 1.18+ (dapr/dapr#9810) and dapr-ext-workflow 1.18+ +(dapr/python-sdk#1025). Against an older sidecar the propagation field is +silently dropped and ctx.get_propagated_history() returns None. +""" + +from __future__ import annotations + +import json +import time +from dataclasses import asdict, dataclass +from typing import Optional + +import dapr.ext.workflow as wf + +# --------------------------------------------------------------------------- +# Data models +# --------------------------------------------------------------------------- + +@dataclass +class PaymentRequest: + card_last4: str + amount: float + currency: str + merchant_id: str + description: str + + def to_json(self) -> str: + return json.dumps(asdict(self)) + + @classmethod + def from_json(cls, data: str) -> "PaymentRequest": + return cls(**json.loads(data)) + + +@dataclass +class FraudCheckResult: + risk_score: float + approved: bool + reason: str + event_count: int + + +@dataclass +class SettlementResult: + transaction_id: str + status: str + event_count: int + + +# --------------------------------------------------------------------------- +# Workflow runtime +# --------------------------------------------------------------------------- + +wfr = wf.WorkflowRuntime() + + +# --------------------------------------------------------------------------- +# Activities +# --------------------------------------------------------------------------- + +@wfr.activity(name='validate_merchant') +def validate_merchant(ctx: wf.WorkflowActivityContext, req_json: str) -> bool: + req = PaymentRequest.from_json(req_json) + print(f' [ValidateMerchant] Validating merchant {req.merchant_id}', flush=True) + return True + + +@wfr.activity(name='validate_card') +def validate_card(ctx: wf.WorkflowActivityContext, req_json: str) -> bool: + req = PaymentRequest.from_json(req_json) + # This activity receives no propagated history (called without propagation= param) + history = ctx.get_propagated_history() + print( + f' [ValidateCard] Validating card ****{req.card_last4} ' + f'(propagated history: {_describe_activity_history(history)})', + flush=True, + ) + return True + + +@wfr.activity(name='check_spending_limits') +def check_spending_limits(ctx: wf.WorkflowActivityContext, req_json: str) -> bool: + req = PaymentRequest.from_json(req_json) + history = ctx.get_propagated_history() + print( + f' [CheckSpendingLimits] Checking {req.amount} {req.currency} ' + f'(propagated history: {_describe_activity_history(history)})', + flush=True, + ) + return req.amount <= 10000 + + +@wfr.activity(name='settle_payment') +def settle_payment(ctx: wf.WorkflowActivityContext, req_json: str) -> str: + """Receives PropagationScope.OWN_HISTORY from ProcessPayment — sees only + the ProcessPayment workflow's events, not the MerchantCheckout ancestor.""" + req = PaymentRequest.from_json(req_json) + history = ctx.get_propagated_history() + + event_count = 0 + if history is not None: + workflows = history.get_workflows() + event_count = sum(1 for _ in workflows) # coarse count; use history.events for full count + print(f' [SettlePayment] Propagated workflows: {[w.name for w in workflows]}', flush=True) + for wf_result in workflows: + print( + f' [SettlePayment] workflow: name={wf_result.name} app={wf_result.app_id}', + flush=True, + ) + else: + print(' [SettlePayment] No propagated history received', flush=True) + + txn_id = f'txn-{req.merchant_id}-{int(time.time() * 1000)}' + print(f' [SettlePayment] SETTLED: {txn_id}', flush=True) + return json.dumps(asdict(SettlementResult( + transaction_id=txn_id, + status='settled', + event_count=event_count, + ))) + + +# --------------------------------------------------------------------------- +# Child workflows +# --------------------------------------------------------------------------- + +@wfr.workflow(name='FraudDetection') +def fraud_detection(ctx: wf.DaprWorkflowContext, req_json: str): + """Grandchild workflow that inspects the full ancestor chain. + + Receives PropagationScope.LINEAGE from ProcessPayment, so its + get_propagated_history() contains events from both MerchantCheckout + and ProcessPayment. + """ + req = PaymentRequest.from_json(req_json) + print( + f' [FraudDetection] Checking payment ****{req.card_last4} {req.amount} {req.currency}', + flush=True, + ) + + history = ctx.get_propagated_history() + if history is None: + print( + ' [FraudDetection] WARNING: no propagated history — sidecar may not support 1.18+', + flush=True, + ) + result = FraudCheckResult( + risk_score=1.0, + approved=False, + reason='no execution history provided — cannot verify caller pipeline', + event_count=0, + ) + return json.dumps(asdict(result)) + + workflows = history.get_workflows() + print( + f' [FraudDetection] Received propagated history with workflows: ' + f'{[w.name for w in workflows]}', + flush=True, + ) + + # Verify the ancestor chain includes the required steps. + try: + merchant_wf = history.get_workflow_by_name('MerchantCheckout') + except wf.PropagationNotFoundError: + return json.dumps(asdict(FraudCheckResult( + risk_score=0.9, + approved=False, + reason='MerchantCheckout missing from propagated history', + event_count=0, + ))) + + try: + process_wf = history.get_workflow_by_name('ProcessPayment') + except wf.PropagationNotFoundError: + return json.dumps(asdict(FraudCheckResult( + risk_score=0.9, + approved=False, + reason='ProcessPayment missing from propagated history', + event_count=0, + ))) + + try: + merchant_act = merchant_wf.get_activity_by_name('validate_merchant') + except wf.PropagationNotFoundError: + return json.dumps(asdict(FraudCheckResult( + risk_score=0.85, + approved=False, + reason='validate_merchant not found in MerchantCheckout history', + event_count=0, + ))) + + try: + card_act = process_wf.get_activity_by_name('validate_card') + spending_act = process_wf.get_activity_by_name('check_spending_limits') + except wf.PropagationNotFoundError as exc: + return json.dumps(asdict(FraudCheckResult( + risk_score=0.85, + approved=False, + reason=f'required activity missing from ProcessPayment history: {exc}', + event_count=0, + ))) + + print( + f' [FraudDetection] Verification:\n' + f' MerchantCheckout/validate_merchant: completed={merchant_act.completed}\n' + f' ProcessPayment/validate_card: completed={card_act.completed}\n' + f' ProcessPayment/check_spending_limits: completed={spending_act.completed}', + flush=True, + ) + + if not (merchant_act.completed and card_act.completed and spending_act.completed): + return json.dumps(asdict(FraudCheckResult( + risk_score=0.9, + approved=False, + reason='required upstream checks not completed in propagated history', + event_count=len(workflows), + ))) + + risk_score = 0.3 if req.amount > 1000 else 0.1 + print(f' [FraudDetection] APPROVED (risk={risk_score:.2f})', flush=True) + return json.dumps(asdict(FraudCheckResult( + risk_score=risk_score, + approved=True, + reason='all upstream checks verified in propagated history', + event_count=len(workflows), + ))) + + +@wfr.workflow(name='ProcessPayment') +def process_payment(ctx: wf.DaprWorkflowContext, req_json: str): + """Child workflow — orchestrates card validation, fraud check, and settlement. + + Receives PropagationScope.LINEAGE from MerchantCheckout, so it holds + the full ancestor chain when calling its own children. + """ + req = PaymentRequest.from_json(req_json) + print( + f' [ProcessPayment] Starting payment ****{req.card_last4} ' + f'{req.amount} {req.currency}', + flush=True, + ) + + # Step 1: Validate card (no propagation) + print(' [ProcessPayment] Step 1: validate_card (no propagation)', flush=True) + card_valid = yield ctx.call_activity(validate_card, input=req_json) + if not card_valid: + return 'payment declined: invalid card' + print(' [ProcessPayment] Step 1 complete: card valid', flush=True) + + # Step 2: Check spending limits (no propagation) + print(' [ProcessPayment] Step 2: check_spending_limits (no propagation)', flush=True) + within_limits = yield ctx.call_activity(check_spending_limits, input=req_json) + if not within_limits: + return 'payment declined: spending limit exceeded' + print(' [ProcessPayment] Step 2 complete: within limits', flush=True) + + # Step 3: Fraud detection child workflow with LINEAGE propagation. + # The grandchild sees both MerchantCheckout AND ProcessPayment events. + print( + ' [ProcessPayment] Step 3: FraudDetection child wf ' + '(PropagationScope.LINEAGE)', + flush=True, + ) + fraud_json = yield ctx.call_child_workflow( + fraud_detection, + input=req_json, + propagation=wf.PropagationScope.LINEAGE, + ) + fraud_result = FraudCheckResult(**json.loads(fraud_json)) + if not fraud_result.approved: + return ( + f'payment declined: fraud check failed ' + f'(risk={fraud_result.risk_score:.2f}, reason={fraud_result.reason})' + ) + print( + f' [ProcessPayment] Step 3 complete: fraud check passed ' + f'(risk={fraud_result.risk_score:.2f})', + flush=True, + ) + + # Step 4: Settle the payment with OWN_HISTORY propagation. + # SettlePayment only sees ProcessPayment's own events, not MerchantCheckout. + print( + ' [ProcessPayment] Step 4: settle_payment (PropagationScope.OWN_HISTORY)', + flush=True, + ) + settlement_json = yield ctx.call_activity( + settle_payment, + input=req_json, + propagation=wf.PropagationScope.OWN_HISTORY, + ) + settlement = SettlementResult(**json.loads(settlement_json)) + print( + f' [ProcessPayment] Step 4 complete: settled (txn={settlement.transaction_id})', + flush=True, + ) + + result = ( + f'payment settled: txn={settlement.transaction_id}, ' + f'card=****{req.card_last4}, amount={req.amount} {req.currency}' + ) + print(f' [ProcessPayment] COMPLETE: {result}', flush=True) + return result + + +@wfr.workflow(name='MerchantCheckout') +def merchant_checkout(ctx: wf.DaprWorkflowContext, req_json: str): + """Root workflow — validates the merchant then delegates payment to a child + workflow with full LINEAGE propagation so the grandchild FraudDetection + can inspect the complete ancestor chain. + """ + req = PaymentRequest.from_json(req_json) + print( + f' [MerchantCheckout] Starting checkout for merchant {req.merchant_id}', + flush=True, + ) + + # Step 1: Validate merchant (no propagation — plain activity) + print(' [MerchantCheckout] Step 1: validate_merchant (no propagation)', flush=True) + yield ctx.call_activity(validate_merchant, input=req_json) + print(' [MerchantCheckout] Step 1 complete: merchant valid', flush=True) + + # Step 2: Delegate to ProcessPayment with LINEAGE propagation. + # ProcessPayment inherits this workflow's history plus any it received from above. + print( + ' [MerchantCheckout] Step 2: ProcessPayment child wf ' + '(PropagationScope.LINEAGE)', + flush=True, + ) + result = yield ctx.call_child_workflow( + process_payment, + input=req_json, + propagation=wf.PropagationScope.LINEAGE, + ) + + print(f' [MerchantCheckout] COMPLETE: {result}', flush=True) + return result + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _describe_activity_history(history: Optional[wf.PropagatedHistory]) -> str: + if history is None: + return 'none' + workflows = history.get_workflows() + return f'{len(workflows)} workflow(s)' + + +def _banner(msg: str) -> str: + line = '=' * (len(msg) + 4) + return f'{line}\n= {msg} =\n{line}' + + +# --------------------------------------------------------------------------- +# Entry point +# --------------------------------------------------------------------------- + +if __name__ == '__main__': + wfr.start() + + req = PaymentRequest( + card_last4='4242', + amount=149.99, + currency='USD', + merchant_id='merchant-abc', + description='Online purchase', + ) + + print(_banner('WORKFLOW HISTORY PROPAGATION DEMO'), flush=True) + print(flush=True) + print(' Flow: MerchantCheckout -> validate_merchant', flush=True) + print(' -> ProcessPayment (child wf, LINEAGE)', flush=True) + print(' -> validate_card -> check_spending_limits', flush=True) + print(' -> FraudDetection (child wf, LINEAGE) <-- sees MerchantCheckout + ProcessPayment events', flush=True) + print(' -> settle_payment (activity, OWN_HISTORY) <-- sees only ProcessPayment events', flush=True) + print(flush=True) + + wf_client = wf.DaprWorkflowClient() + instance_id = wf_client.schedule_new_workflow( + workflow=merchant_checkout, + input=req.to_json(), + instance_id='checkout-001', + ) + print(f' [main] Started workflow instance: {instance_id}', flush=True) + + try: + state = wf_client.wait_for_workflow_completion( + instance_id=instance_id, + timeout_in_seconds=30, + ) + if state is None: + print(' [main] Workflow not found!', flush=True) + elif state.runtime_status.name == 'COMPLETED': + print( + f' [main] Workflow completed! Output: {state.serialized_output}', + flush=True, + ) + else: + print( + f' [main] Workflow ended with status: {state.runtime_status.name}', + flush=True, + ) + except TimeoutError: + print(' [main] Workflow timed out!', flush=True) + + print(flush=True) + print(_banner('COMPLETE'), flush=True) + + wfr.shutdown() diff --git a/workflows/python/sdk-context-propagation/order-processor/requirements.txt b/workflows/python/sdk-context-propagation/order-processor/requirements.txt new file mode 100644 index 000000000..4b21b8080 --- /dev/null +++ b/workflows/python/sdk-context-propagation/order-processor/requirements.txt @@ -0,0 +1,2 @@ +dapr>=1.18.0rc0 +dapr-ext-workflow>=1.18.0rc0 From e8aaa4ba183ec3943f9c8df4e03cbcb11c5199e9 Mon Sep 17 00:00:00 2001 From: Nelson Parente Date: Wed, 20 May 2026 22:00:50 +0100 Subject: [PATCH 2/5] feat(python): rewrite quickstart to patient-intake scenario Aligns the Python workflow history propagation quickstart with the canonical Go reference (dapr/go-sdk#823, dapr/quickstarts#1315) so all SDK quickstarts share the same patient intake / e-prescribing scenario. - Swap credit-card/fraud scenario for patient-intake/e-prescribing - Adopt PatientIntake -> PrescribeMedication -> ComplianceAudit hierarchy - Add is_replaying guards around all print() calls inside workflows - Use Cassie's PascalCase activity/workflow names for cross-SDK consistency Signed-off-by: Nelson Parente --- .../python/sdk-context-propagation/README.md | 143 ++--- .../order-processor/app.py | 505 ++++++++++-------- 2 files changed, 360 insertions(+), 288 deletions(-) diff --git a/workflows/python/sdk-context-propagation/README.md b/workflows/python/sdk-context-propagation/README.md index a4ce5368c..6a34d7a0d 100644 --- a/workflows/python/sdk-context-propagation/README.md +++ b/workflows/python/sdk-context-propagation/README.md @@ -17,40 +17,44 @@ When a parent workflow calls a child workflow or activity it can optionally atta | **Own history** | `PropagationScope.OWN_HISTORY` | Only the direct caller's events | | **Lineage** | `PropagationScope.LINEAGE` | Caller's events **plus** any ancestor history the caller itself received | -## Scenario: Credit-card payment with fraud detection +## Scenario: Patient intake / e-prescribing + +A compliance audit and a pharmacy dispense step refuse to act unless the propagated history proves the required upstream checks (insurance, allergies, drug interactions) actually ran. ``` -MerchantCheckout (root) - └─ validate_merchant (activity, no propagation) - └─ ProcessPayment (child wf, LINEAGE) - └─ validate_card (activity, no propagation) - └─ check_spending_limits (activity, no propagation) - └─ FraudDetection (grandchild wf, LINEAGE) - | reads MerchantCheckout/validate_merchant - | ProcessPayment/validate_card - | ProcessPayment/check_spending_limits - └─ settle_payment (activity, OWN_HISTORY) - reads ProcessPayment events only +PatientIntake (root) + └─ VerifyInsurance (activity, no propagation) + └─ PrescribeMedication (child wf, LINEAGE) + └─ CheckAllergies (activity, no propagation) + └─ ScreenDrugInteractions (activity, no propagation) + └─ ComplianceAudit (grandchild wf, LINEAGE) + | reads PatientIntake/VerifyInsurance + | PrescribeMedication/CheckAllergies + | PrescribeMedication/ScreenDrugInteractions + └─ DispenseMedication (activity, OWN_HISTORY) + reads PrescribeMedication events only ``` -`FraudDetection` uses `PropagationScope.LINEAGE` to see the **full ancestor chain** — it can verify both the merchant validation (performed by the grandparent) and the card/limit checks (performed by the parent) before approving the transaction. +`ComplianceAudit` uses `PropagationScope.LINEAGE` to see the **full ancestor chain** — it can verify both the insurance check (performed by the grandparent `PatientIntake`) and the allergy/interaction checks (performed by the parent `PrescribeMedication`) before approving the prescription. + +`DispenseMedication` uses `PropagationScope.OWN_HISTORY` to see only the **direct caller's events** — a trust-boundary mode that limits visibility to what `PrescribeMedication` itself executed. The pharmacy dispense system doesn't need (or get to see) the upstream patient-intake chain. -`settle_payment` uses `PropagationScope.OWN_HISTORY` to see only the **direct caller's events** — a trust-boundary mode that limits visibility to what `ProcessPayment` itself executed. +This sample mirrors the canonical Go reference [dapr/go-sdk#823](https://github.com/dapr/go-sdk/pull/823) and the [Go quickstart](https://github.com/dapr/quickstarts/pull/1315). ## Python API surface ```python # Parent workflow — propagate LINEAGE when calling a child workflow result = yield ctx.call_child_workflow( - fraud_detection, - input=req_json, + compliance_audit, + input=rec_json, propagation=wf.PropagationScope.LINEAGE, ) # Parent workflow — propagate OWN_HISTORY when calling an activity -settlement = yield ctx.call_activity( - settle_payment, - input=req_json, +dispense = yield ctx.call_activity( + dispense_medication, + input=rec_json, propagation=wf.PropagationScope.OWN_HISTORY, ) @@ -58,10 +62,10 @@ settlement = yield ctx.call_activity( history = ctx.get_propagated_history() # returns PropagatedHistory | None if history is not None: - process_wf = history.get_workflow_by_name('ProcessPayment') # raises PropagationNotFoundError if missing - card_act = process_wf.get_activity_by_name('validate_card') - print(card_act.completed) # bool - print(card_act.output) # JSON string + intake_wf = history.get_workflow_by_name('PatientIntake') # raises PropagationNotFoundError if missing + insurance = intake_wf.get_activity_by_name('VerifyInsurance') + print(insurance.completed) # bool + print(insurance.output) # JSON string ``` Key types exported from `dapr.ext.workflow`: @@ -71,6 +75,8 @@ Key types exported from `dapr.ext.workflow`: - `ActivityResult` — has `.completed`, `.output` fields - `PropagationNotFoundError` — raised when a named workflow/activity is not in the history +> **Replay safety**: workflow code runs many times during durable execution. Guard side-effecting calls — including `print()` — with `if not ctx.is_replaying:` so they only fire on the live execution, not on each replay. + ## Prerequisites - [Dapr CLI](https://docs.dapr.io/getting-started/install-dapr-cli/) 1.18+ @@ -93,50 +99,56 @@ dapr run -f . ## Expected output ``` -============================================ -= WORKFLOW HISTORY PROPAGATION DEMO = -============================================ - - Flow: MerchantCheckout -> validate_merchant - -> ProcessPayment (child wf, LINEAGE) - -> validate_card -> check_spending_limits - -> FraudDetection (child wf, LINEAGE) <-- sees MerchantCheckout + ProcessPayment events - -> settle_payment (activity, OWN_HISTORY) <-- sees only ProcessPayment events - - [main] Started workflow instance: checkout-001 - [MerchantCheckout] Starting checkout for merchant merchant-abc - [MerchantCheckout] Step 1: validate_merchant (no propagation) - [ValidateMerchant] Validating merchant merchant-abc - [MerchantCheckout] Step 1 complete: merchant valid - [MerchantCheckout] Step 2: ProcessPayment child wf (PropagationScope.LINEAGE) - [ProcessPayment] Starting payment ****4242 149.99 USD - [ProcessPayment] Step 1: validate_card (no propagation) - [ValidateCard] Validating card ****4242 (propagated history: none) - [ProcessPayment] Step 1 complete: card valid - [ProcessPayment] Step 2: check_spending_limits (no propagation) - [CheckSpendingLimits] Checking 149.99 USD (propagated history: none) - [ProcessPayment] Step 2 complete: within limits - [ProcessPayment] Step 3: FraudDetection child wf (PropagationScope.LINEAGE) - [FraudDetection] Received propagated history with workflows: ['MerchantCheckout', 'ProcessPayment'] - [FraudDetection] Verification: - MerchantCheckout/validate_merchant: completed=True - ProcessPayment/validate_card: completed=True - ProcessPayment/check_spending_limits: completed=True - [FraudDetection] APPROVED (risk=0.10) - [ProcessPayment] Step 3 complete: fraud check passed (risk=0.10) - [ProcessPayment] Step 4: settle_payment (PropagationScope.OWN_HISTORY) - [SettlePayment] Propagated workflows: ['ProcessPayment'] - [SettlePayment] SETTLED: txn-merchant-abc-1748000000000 - [ProcessPayment] Step 4 complete: settled (txn=txn-merchant-abc-...) - [ProcessPayment] COMPLETE: payment settled: ... - [MerchantCheckout] COMPLETE: payment settled: ... - [main] Workflow completed! Output: "payment settled: ..." - -======================== +============================================================ += WORKFLOW HISTORY PROPAGATION DEMO — PATIENT INTAKE = +============================================================ + + Flow: PatientIntake -> VerifyInsurance + -> PrescribeMedication (child wf, LINEAGE) + -> CheckAllergies -> ScreenDrugInteractions + -> ComplianceAudit (child wf, LINEAGE) <-- sees PatientIntake + PrescribeMedication events + -> DispenseMedication (activity, OWN_HISTORY) <-- sees only PrescribeMedication events + + [main] Started workflow instance: intake-001 + [PatientIntake] Starting intake for patient P-1042 + [PatientIntake] Step 1: VerifyInsurance (no propagation) + [VerifyInsurance] Checking coverage for patient P-1042 + [PatientIntake] Step 1 complete: insurance verified + [PatientIntake] Step 2: PrescribeMedication child wf (PropagationScope.LINEAGE) + [PrescribeMedication] Starting prescription: amoxicillin 500mg for bacterial sinusitis + [PrescribeMedication] Step 1: CheckAllergies (no propagation) + [CheckAllergies] Screening P-1042 for amoxicillin + [PrescribeMedication] Step 1 complete: allergy clear + [PrescribeMedication] Step 2: ScreenDrugInteractions (no propagation) + [ScreenDrugInteractions] Screening amoxicillin 500mg for P-1042 + [PrescribeMedication] Step 2 complete: no interactions + [PrescribeMedication] Step 3: ComplianceAudit child wf (PropagationScope.LINEAGE) + [ComplianceAudit] Auditing prescription for patient P-1042 + [ComplianceAudit] Received propagated history with workflows: ['PatientIntake', 'PrescribeMedication'] + [ComplianceAudit] Verification: + PatientIntake/VerifyInsurance: completed=True + PrescribeMedication/CheckAllergies: completed=True + PrescribeMedication/ScreenDrugInteractions: completed=True + [ComplianceAudit] APPROVED (risk=0.10) + [PrescribeMedication] Step 3 complete: compliance audit passed (risk=0.10, 2 workflow(s) verified) + [PrescribeMedication] Step 4: DispenseMedication (PropagationScope.OWN_HISTORY) + [DispenseMedication] Propagated workflows: ['PrescribeMedication'] + [DispenseMedication] workflow: name=PrescribeMedication app=order-processor + [DispenseMedication] DISPENSED: rx-P-1042-... (amoxicillin 500mg) + [PrescribeMedication] Step 4 complete: dispensed (id=rx-P-1042-..., 1 workflow(s) verified) + [PrescribeMedication] COMPLETE: dispensed: id=rx-P-1042-..., patient=P-1042, drug=amoxicillin 500mg + [PatientIntake] COMPLETE: dispensed: id=rx-P-1042-..., patient=P-1042, drug=amoxicillin 500mg + [main] Workflow completed! Output: "dispensed: ..." + +================ = COMPLETE = -======================== +================ ``` +## Standalone-mode note + +In standalone mode the sidecar will log `propagating unsigned workflow history to ...` warnings — these are expected. Without `WorkflowHistorySigning` enabled, propagated history chunks aren't cryptographically signed, which is fine for a local `dapr run` demo. Signing the chunks within an mTLS trust boundary is a production concern handled at the cluster/control-plane level and is out of scope for this quickstart. + ## Stop the sample ```sh @@ -148,5 +160,6 @@ dapr stop -f . - [Proposal: Workflow History Propagation (dapr/proposals#102)](https://github.com/dapr/proposals/issues/102) - [Runtime PR: dapr/dapr#9810](https://github.com/dapr/dapr/pull/9810) - [Python SDK PR: dapr/python-sdk#1025](https://github.com/dapr/python-sdk/pull/1025) -- [Go SDK reference: dapr/go-sdk#823](https://github.com/dapr/go-sdk/pull/823) +- [Canonical Go SDK reference: dapr/go-sdk#823](https://github.com/dapr/go-sdk/pull/823) +- [Sibling Go quickstart: dapr/quickstarts#1315](https://github.com/dapr/quickstarts/pull/1315) - [Dapr Workflow documentation](https://docs.dapr.io/developing-applications/building-blocks/workflow/) diff --git a/workflows/python/sdk-context-propagation/order-processor/app.py b/workflows/python/sdk-context-propagation/order-processor/app.py index be2abac78..1c282c469 100644 --- a/workflows/python/sdk-context-propagation/order-processor/app.py +++ b/workflows/python/sdk-context-propagation/order-processor/app.py @@ -12,20 +12,23 @@ """Workflow history propagation quickstart (Python SDK). -Scenario: credit-card payment processing with fraud detection. +Scenario: patient intake / e-prescribing pipeline. A compliance audit and a +pharmacy dispense step refuse to act unless the propagated history proves +the required upstream checks (insurance, allergies, drug interactions) +actually ran. Flow: - MerchantCheckout (root workflow) - └─ ValidateMerchant (activity, no propagation) - └─ ProcessPayment (child workflow, PropagationScope.LINEAGE) - └─ ValidateCard (activity, no propagation) - └─ CheckSpendingLimits (activity, no propagation) - └─ FraudDetection (child workflow, PropagationScope.LINEAGE) - | reads: MerchantCheckout/ValidateMerchant - | ProcessPayment/ValidateCard - | ProcessPayment/CheckSpendingLimits - └─ SettlePayment (activity, PropagationScope.OWN_HISTORY) - reads: ProcessPayment events only (no ancestor chain) + PatientIntake (root workflow) + |- VerifyInsurance (activity, no propagation) + `- PrescribeMedication (child workflow, PropagationScope.LINEAGE) + |- CheckAllergies (activity, no propagation) + |- ScreenDrugInteractions (activity, no propagation) + |- ComplianceAudit (grandchild wf, PropagationScope.LINEAGE) + | reads: PatientIntake/VerifyInsurance + | PrescribeMedication/CheckAllergies + | PrescribeMedication/ScreenDrugInteractions + `- DispenseMedication (activity, PropagationScope.OWN_HISTORY) + reads: PrescribeMedication events only (no PatientIntake) This requires Dapr 1.18+ (dapr/dapr#9810) and dapr-ext-workflow 1.18+ (dapr/python-sdk#1025). Against an older sidecar the propagation field is @@ -37,7 +40,6 @@ import json import time from dataclasses import asdict, dataclass -from typing import Optional import dapr.ext.workflow as wf @@ -46,32 +48,34 @@ # --------------------------------------------------------------------------- @dataclass -class PaymentRequest: - card_last4: str - amount: float - currency: str - merchant_id: str - description: str +class PatientRecord: + patient_id: str + name: str + dob: str + mrn: str + condition: str + medication: str + dosage: float def to_json(self) -> str: return json.dumps(asdict(self)) @classmethod - def from_json(cls, data: str) -> "PaymentRequest": + def from_json(cls, data: str) -> "PatientRecord": return cls(**json.loads(data)) @dataclass -class FraudCheckResult: +class ComplianceResult: + compliant: bool risk_score: float - approved: bool reason: str event_count: int @dataclass -class SettlementResult: - transaction_id: str +class DispenseResult: + dispense_id: str status: str event_count: int @@ -87,63 +91,69 @@ class SettlementResult: # Activities # --------------------------------------------------------------------------- -@wfr.activity(name='validate_merchant') -def validate_merchant(ctx: wf.WorkflowActivityContext, req_json: str) -> bool: - req = PaymentRequest.from_json(req_json) - print(f' [ValidateMerchant] Validating merchant {req.merchant_id}', flush=True) +@wfr.activity(name='VerifyInsurance') +def verify_insurance(ctx: wf.WorkflowActivityContext, rec_json: str) -> bool: + rec = PatientRecord.from_json(rec_json) + print(f' [VerifyInsurance] Checking coverage for patient {rec.patient_id}', flush=True) return True -@wfr.activity(name='validate_card') -def validate_card(ctx: wf.WorkflowActivityContext, req_json: str) -> bool: - req = PaymentRequest.from_json(req_json) - # This activity receives no propagated history (called without propagation= param) - history = ctx.get_propagated_history() +@wfr.activity(name='CheckAllergies') +def check_allergies(ctx: wf.WorkflowActivityContext, rec_json: str) -> bool: + rec = PatientRecord.from_json(rec_json) print( - f' [ValidateCard] Validating card ****{req.card_last4} ' - f'(propagated history: {_describe_activity_history(history)})', + f' [CheckAllergies] Screening {rec.patient_id} for {rec.medication}', flush=True, ) return True -@wfr.activity(name='check_spending_limits') -def check_spending_limits(ctx: wf.WorkflowActivityContext, req_json: str) -> bool: - req = PaymentRequest.from_json(req_json) - history = ctx.get_propagated_history() +@wfr.activity(name='ScreenDrugInteractions') +def screen_drug_interactions(ctx: wf.WorkflowActivityContext, rec_json: str) -> bool: + rec = PatientRecord.from_json(rec_json) print( - f' [CheckSpendingLimits] Checking {req.amount} {req.currency} ' - f'(propagated history: {_describe_activity_history(history)})', + f' [ScreenDrugInteractions] Screening {rec.medication} {rec.dosage:.0f}mg for {rec.patient_id}', flush=True, ) - return req.amount <= 10000 + return True + +@wfr.activity(name='DispenseMedication') +def dispense_medication(ctx: wf.WorkflowActivityContext, rec_json: str) -> str: + """Receives PropagationScope.OWN_HISTORY from PrescribeMedication — sees only + the PrescribeMedication workflow's events, not the PatientIntake ancestor. -@wfr.activity(name='settle_payment') -def settle_payment(ctx: wf.WorkflowActivityContext, req_json: str) -> str: - """Receives PropagationScope.OWN_HISTORY from ProcessPayment — sees only - the ProcessPayment workflow's events, not the MerchantCheckout ancestor.""" - req = PaymentRequest.from_json(req_json) + The pharmacy dispense system intentionally does not get to see the + upstream patient-intake chain; it only needs proof that the prescribing + workflow itself ran the right checks. + """ + rec = PatientRecord.from_json(rec_json) history = ctx.get_propagated_history() event_count = 0 if history is not None: workflows = history.get_workflows() - event_count = sum(1 for _ in workflows) # coarse count; use history.events for full count - print(f' [SettlePayment] Propagated workflows: {[w.name for w in workflows]}', flush=True) + event_count = len(workflows) + print( + f' [DispenseMedication] Propagated workflows: {[w.name for w in workflows]}', + flush=True, + ) for wf_result in workflows: print( - f' [SettlePayment] workflow: name={wf_result.name} app={wf_result.app_id}', + f' [DispenseMedication] workflow: name={wf_result.name} app={wf_result.app_id}', flush=True, ) else: - print(' [SettlePayment] No propagated history received', flush=True) + print(' [DispenseMedication] No propagated history received', flush=True) - txn_id = f'txn-{req.merchant_id}-{int(time.time() * 1000)}' - print(f' [SettlePayment] SETTLED: {txn_id}', flush=True) - return json.dumps(asdict(SettlementResult( - transaction_id=txn_id, - status='settled', + dispense_id = f'rx-{rec.patient_id}-{int(time.time() * 1000)}' + print( + f' [DispenseMedication] DISPENSED: {dispense_id} ({rec.medication} {rec.dosage:.0f}mg)', + flush=True, + ) + return json.dumps(asdict(DispenseResult( + dispense_id=dispense_id, + status='dispensed', event_count=event_count, ))) @@ -152,217 +162,271 @@ def settle_payment(ctx: wf.WorkflowActivityContext, req_json: str) -> str: # Child workflows # --------------------------------------------------------------------------- -@wfr.workflow(name='FraudDetection') -def fraud_detection(ctx: wf.DaprWorkflowContext, req_json: str): +@wfr.workflow(name='ComplianceAudit') +def compliance_audit(ctx: wf.DaprWorkflowContext, rec_json: str): """Grandchild workflow that inspects the full ancestor chain. - Receives PropagationScope.LINEAGE from ProcessPayment, so its - get_propagated_history() contains events from both MerchantCheckout - and ProcessPayment. + Receives PropagationScope.LINEAGE from PrescribeMedication, so its + get_propagated_history() contains events from both PatientIntake and + PrescribeMedication. It refuses to approve dispensing unless the required + upstream steps (insurance, allergies, drug interactions) are all present + and completed in the propagated history. """ - req = PaymentRequest.from_json(req_json) - print( - f' [FraudDetection] Checking payment ****{req.card_last4} {req.amount} {req.currency}', - flush=True, - ) - - history = ctx.get_propagated_history() - if history is None: + rec = PatientRecord.from_json(rec_json) + if not ctx.is_replaying: print( - ' [FraudDetection] WARNING: no propagated history — sidecar may not support 1.18+', + f' [ComplianceAudit] Auditing prescription for patient {rec.patient_id}', flush=True, ) - result = FraudCheckResult( + + history = ctx.get_propagated_history() + if history is None: + if not ctx.is_replaying: + print( + ' [ComplianceAudit] WARNING: no propagated history — sidecar may not support 1.18+', + flush=True, + ) + print( + ' [ComplianceAudit] BLOCKED — cannot verify upstream pipeline without history', + flush=True, + ) + return json.dumps(asdict(ComplianceResult( + compliant=False, risk_score=1.0, - approved=False, reason='no execution history provided — cannot verify caller pipeline', event_count=0, - ) - return json.dumps(asdict(result)) + ))) workflows = history.get_workflows() - print( - f' [FraudDetection] Received propagated history with workflows: ' - f'{[w.name for w in workflows]}', - flush=True, - ) + if not ctx.is_replaying: + print( + f' [ComplianceAudit] Received propagated history with workflows: ' + f'{[w.name for w in workflows]}', + flush=True, + ) - # Verify the ancestor chain includes the required steps. + # Verify the ancestor chain includes the required workflows. try: - merchant_wf = history.get_workflow_by_name('MerchantCheckout') + intake_wf = history.get_workflow_by_name('PatientIntake') except wf.PropagationNotFoundError: - return json.dumps(asdict(FraudCheckResult( + return json.dumps(asdict(ComplianceResult( + compliant=False, risk_score=0.9, - approved=False, - reason='MerchantCheckout missing from propagated history', - event_count=0, + reason='PatientIntake missing from propagated history', + event_count=len(workflows), ))) try: - process_wf = history.get_workflow_by_name('ProcessPayment') + prescribe_wf = history.get_workflow_by_name('PrescribeMedication') except wf.PropagationNotFoundError: - return json.dumps(asdict(FraudCheckResult( + return json.dumps(asdict(ComplianceResult( + compliant=False, risk_score=0.9, - approved=False, - reason='ProcessPayment missing from propagated history', - event_count=0, + reason='PrescribeMedication missing from propagated history', + event_count=len(workflows), ))) + # Verify the required activity completions are recorded. try: - merchant_act = merchant_wf.get_activity_by_name('validate_merchant') + insurance_act = intake_wf.get_activity_by_name('VerifyInsurance') except wf.PropagationNotFoundError: - return json.dumps(asdict(FraudCheckResult( + return json.dumps(asdict(ComplianceResult( + compliant=False, risk_score=0.85, - approved=False, - reason='validate_merchant not found in MerchantCheckout history', - event_count=0, + reason='VerifyInsurance not found in PatientIntake history', + event_count=len(workflows), ))) try: - card_act = process_wf.get_activity_by_name('validate_card') - spending_act = process_wf.get_activity_by_name('check_spending_limits') + allergies_act = prescribe_wf.get_activity_by_name('CheckAllergies') + interactions_act = prescribe_wf.get_activity_by_name('ScreenDrugInteractions') except wf.PropagationNotFoundError as exc: - return json.dumps(asdict(FraudCheckResult( + return json.dumps(asdict(ComplianceResult( + compliant=False, risk_score=0.85, - approved=False, - reason=f'required activity missing from ProcessPayment history: {exc}', - event_count=0, + reason=f'required activity missing from PrescribeMedication history: {exc}', + event_count=len(workflows), ))) - print( - f' [FraudDetection] Verification:\n' - f' MerchantCheckout/validate_merchant: completed={merchant_act.completed}\n' - f' ProcessPayment/validate_card: completed={card_act.completed}\n' - f' ProcessPayment/check_spending_limits: completed={spending_act.completed}', - flush=True, - ) + if not ctx.is_replaying: + print(' [ComplianceAudit] Verification:', flush=True) + print( + f' PatientIntake/VerifyInsurance: completed={insurance_act.completed}', + flush=True, + ) + print( + f' PrescribeMedication/CheckAllergies: completed={allergies_act.completed}', + flush=True, + ) + print( + f' PrescribeMedication/ScreenDrugInteractions: completed={interactions_act.completed}', + flush=True, + ) - if not (merchant_act.completed and card_act.completed and spending_act.completed): - return json.dumps(asdict(FraudCheckResult( + if not (insurance_act.completed and allergies_act.completed and interactions_act.completed): + if not ctx.is_replaying: + print( + ' [ComplianceAudit] BLOCKED — required upstream checks not completed', + flush=True, + ) + return json.dumps(asdict(ComplianceResult( + compliant=False, risk_score=0.9, - approved=False, reason='required upstream checks not completed in propagated history', event_count=len(workflows), ))) - risk_score = 0.3 if req.amount > 1000 else 0.1 - print(f' [FraudDetection] APPROVED (risk={risk_score:.2f})', flush=True) - return json.dumps(asdict(FraudCheckResult( + risk_score = 0.3 if rec.dosage > 1000 else 0.1 + if not ctx.is_replaying: + print(f' [ComplianceAudit] APPROVED (risk={risk_score:.2f})', flush=True) + return json.dumps(asdict(ComplianceResult( + compliant=True, risk_score=risk_score, - approved=True, reason='all upstream checks verified in propagated history', event_count=len(workflows), ))) -@wfr.workflow(name='ProcessPayment') -def process_payment(ctx: wf.DaprWorkflowContext, req_json: str): - """Child workflow — orchestrates card validation, fraud check, and settlement. +@wfr.workflow(name='PrescribeMedication') +def prescribe_medication(ctx: wf.DaprWorkflowContext, rec_json: str): + """Child workflow — orchestrates allergy + interaction screening, compliance + audit, and dispensing. - Receives PropagationScope.LINEAGE from MerchantCheckout, so it holds - the full ancestor chain when calling its own children. + Receives PropagationScope.LINEAGE from PatientIntake, so it holds the full + ancestor chain when calling its own children. Calls ComplianceAudit with + LINEAGE (audit needs to see the grandparent) and DispenseMedication with + OWN_HISTORY (pharmacy only sees the prescribing step, not the intake). """ - req = PaymentRequest.from_json(req_json) - print( - f' [ProcessPayment] Starting payment ****{req.card_last4} ' - f'{req.amount} {req.currency}', - flush=True, - ) + rec = PatientRecord.from_json(rec_json) + if not ctx.is_replaying: + print( + f' [PrescribeMedication] Starting prescription: {rec.medication} ' + f'{rec.dosage:.0f}mg for {rec.condition}', + flush=True, + ) - # Step 1: Validate card (no propagation) - print(' [ProcessPayment] Step 1: validate_card (no propagation)', flush=True) - card_valid = yield ctx.call_activity(validate_card, input=req_json) - if not card_valid: - return 'payment declined: invalid card' - print(' [ProcessPayment] Step 1 complete: card valid', flush=True) - - # Step 2: Check spending limits (no propagation) - print(' [ProcessPayment] Step 2: check_spending_limits (no propagation)', flush=True) - within_limits = yield ctx.call_activity(check_spending_limits, input=req_json) - if not within_limits: - return 'payment declined: spending limit exceeded' - print(' [ProcessPayment] Step 2 complete: within limits', flush=True) - - # Step 3: Fraud detection child workflow with LINEAGE propagation. - # The grandchild sees both MerchantCheckout AND ProcessPayment events. - print( - ' [ProcessPayment] Step 3: FraudDetection child wf ' - '(PropagationScope.LINEAGE)', - flush=True, - ) - fraud_json = yield ctx.call_child_workflow( - fraud_detection, - input=req_json, + # Step 1: Allergy check (no propagation — plain activity) + if not ctx.is_replaying: + print( + ' [PrescribeMedication] Step 1: CheckAllergies (no propagation)', + flush=True, + ) + allergy_clear = yield ctx.call_activity(check_allergies, input=rec_json) + if not allergy_clear: + return 'prescription declined: known allergy' + if not ctx.is_replaying: + print(' [PrescribeMedication] Step 1 complete: allergy clear', flush=True) + + # Step 2: Drug interaction screen (no propagation) + if not ctx.is_replaying: + print( + ' [PrescribeMedication] Step 2: ScreenDrugInteractions (no propagation)', + flush=True, + ) + interactions_clear = yield ctx.call_activity(screen_drug_interactions, input=rec_json) + if not interactions_clear: + return 'prescription declined: drug interaction risk' + if not ctx.is_replaying: + print(' [PrescribeMedication] Step 2 complete: no interactions', flush=True) + + # Step 3: Compliance audit grandchild workflow with LINEAGE propagation. + # The grandchild sees both PatientIntake AND PrescribeMedication events. + if not ctx.is_replaying: + print( + ' [PrescribeMedication] Step 3: ComplianceAudit child wf ' + '(PropagationScope.LINEAGE)', + flush=True, + ) + audit_json = yield ctx.call_child_workflow( + compliance_audit, + input=rec_json, propagation=wf.PropagationScope.LINEAGE, ) - fraud_result = FraudCheckResult(**json.loads(fraud_json)) - if not fraud_result.approved: + audit = ComplianceResult(**json.loads(audit_json)) + if not audit.compliant: return ( - f'payment declined: fraud check failed ' - f'(risk={fraud_result.risk_score:.2f}, reason={fraud_result.reason})' + f'prescription blocked: compliance audit failed ' + f'(risk={audit.risk_score:.2f}, reason={audit.reason})' + ) + if not ctx.is_replaying: + print( + f' [PrescribeMedication] Step 3 complete: compliance audit passed ' + f'(risk={audit.risk_score:.2f}, {audit.event_count} workflow(s) verified)', + flush=True, ) - print( - f' [ProcessPayment] Step 3 complete: fraud check passed ' - f'(risk={fraud_result.risk_score:.2f})', - flush=True, - ) - # Step 4: Settle the payment with OWN_HISTORY propagation. - # SettlePayment only sees ProcessPayment's own events, not MerchantCheckout. - print( - ' [ProcessPayment] Step 4: settle_payment (PropagationScope.OWN_HISTORY)', - flush=True, - ) - settlement_json = yield ctx.call_activity( - settle_payment, - input=req_json, + # Step 4: Dispense the medication with OWN_HISTORY propagation. + # DispenseMedication only sees PrescribeMedication's events, not PatientIntake. + if not ctx.is_replaying: + print( + ' [PrescribeMedication] Step 4: DispenseMedication ' + '(PropagationScope.OWN_HISTORY)', + flush=True, + ) + dispense_json = yield ctx.call_activity( + dispense_medication, + input=rec_json, propagation=wf.PropagationScope.OWN_HISTORY, ) - settlement = SettlementResult(**json.loads(settlement_json)) - print( - f' [ProcessPayment] Step 4 complete: settled (txn={settlement.transaction_id})', - flush=True, - ) + dispense = DispenseResult(**json.loads(dispense_json)) + if not ctx.is_replaying: + print( + f' [PrescribeMedication] Step 4 complete: dispensed ' + f'(id={dispense.dispense_id}, {dispense.event_count} workflow(s) verified)', + flush=True, + ) result = ( - f'payment settled: txn={settlement.transaction_id}, ' - f'card=****{req.card_last4}, amount={req.amount} {req.currency}' + f'dispensed: id={dispense.dispense_id}, patient={rec.patient_id}, ' + f'drug={rec.medication} {rec.dosage:.0f}mg' ) - print(f' [ProcessPayment] COMPLETE: {result}', flush=True) + if not ctx.is_replaying: + print(f' [PrescribeMedication] COMPLETE: {result}', flush=True) return result -@wfr.workflow(name='MerchantCheckout') -def merchant_checkout(ctx: wf.DaprWorkflowContext, req_json: str): - """Root workflow — validates the merchant then delegates payment to a child - workflow with full LINEAGE propagation so the grandchild FraudDetection - can inspect the complete ancestor chain. +@wfr.workflow(name='PatientIntake') +def patient_intake(ctx: wf.DaprWorkflowContext, rec_json: str): + """Root workflow — verifies the patient's insurance then delegates the + prescription to a child workflow with full LINEAGE propagation so the + grandchild ComplianceAudit can inspect the complete ancestor chain. """ - req = PaymentRequest.from_json(req_json) - print( - f' [MerchantCheckout] Starting checkout for merchant {req.merchant_id}', - flush=True, - ) - - # Step 1: Validate merchant (no propagation — plain activity) - print(' [MerchantCheckout] Step 1: validate_merchant (no propagation)', flush=True) - yield ctx.call_activity(validate_merchant, input=req_json) - print(' [MerchantCheckout] Step 1 complete: merchant valid', flush=True) + rec = PatientRecord.from_json(rec_json) + if not ctx.is_replaying: + print( + f' [PatientIntake] Starting intake for patient {rec.patient_id}', + flush=True, + ) - # Step 2: Delegate to ProcessPayment with LINEAGE propagation. - # ProcessPayment inherits this workflow's history plus any it received from above. - print( - ' [MerchantCheckout] Step 2: ProcessPayment child wf ' - '(PropagationScope.LINEAGE)', - flush=True, - ) + # Step 1: Verify insurance (no propagation — plain activity) + if not ctx.is_replaying: + print( + ' [PatientIntake] Step 1: VerifyInsurance (no propagation)', + flush=True, + ) + insured = yield ctx.call_activity(verify_insurance, input=rec_json) + if not insured: + return 'intake declined: insurance not on file' + if not ctx.is_replaying: + print(' [PatientIntake] Step 1 complete: insurance verified', flush=True) + + # Step 2: Delegate to PrescribeMedication with LINEAGE propagation. + # PrescribeMedication inherits this workflow's history so its own + # grandchild ComplianceAudit can verify the complete ancestor chain. + if not ctx.is_replaying: + print( + ' [PatientIntake] Step 2: PrescribeMedication child wf ' + '(PropagationScope.LINEAGE)', + flush=True, + ) result = yield ctx.call_child_workflow( - process_payment, - input=req_json, + prescribe_medication, + input=rec_json, propagation=wf.PropagationScope.LINEAGE, ) - print(f' [MerchantCheckout] COMPLETE: {result}', flush=True) + if not ctx.is_replaying: + print(f' [PatientIntake] COMPLETE: {result}', flush=True) return result @@ -370,13 +434,6 @@ def merchant_checkout(ctx: wf.DaprWorkflowContext, req_json: str): # Helpers # --------------------------------------------------------------------------- -def _describe_activity_history(history: Optional[wf.PropagatedHistory]) -> str: - if history is None: - return 'none' - workflows = history.get_workflows() - return f'{len(workflows)} workflow(s)' - - def _banner(msg: str) -> str: line = '=' * (len(msg) + 4) return f'{line}\n= {msg} =\n{line}' @@ -389,28 +446,30 @@ def _banner(msg: str) -> str: if __name__ == '__main__': wfr.start() - req = PaymentRequest( - card_last4='4242', - amount=149.99, - currency='USD', - merchant_id='merchant-abc', - description='Online purchase', + rec = PatientRecord( + patient_id='P-1042', + name='Jane Doe', + dob='1985-06-12', + mrn='MRN-77231', + condition='bacterial sinusitis', + medication='amoxicillin', + dosage=500, ) - print(_banner('WORKFLOW HISTORY PROPAGATION DEMO'), flush=True) + print(_banner('WORKFLOW HISTORY PROPAGATION DEMO — PATIENT INTAKE'), flush=True) print(flush=True) - print(' Flow: MerchantCheckout -> validate_merchant', flush=True) - print(' -> ProcessPayment (child wf, LINEAGE)', flush=True) - print(' -> validate_card -> check_spending_limits', flush=True) - print(' -> FraudDetection (child wf, LINEAGE) <-- sees MerchantCheckout + ProcessPayment events', flush=True) - print(' -> settle_payment (activity, OWN_HISTORY) <-- sees only ProcessPayment events', flush=True) + print(' Flow: PatientIntake -> VerifyInsurance', flush=True) + print(' -> PrescribeMedication (child wf, LINEAGE)', flush=True) + print(' -> CheckAllergies -> ScreenDrugInteractions', flush=True) + print(' -> ComplianceAudit (child wf, LINEAGE) <-- sees PatientIntake + PrescribeMedication events', flush=True) + print(' -> DispenseMedication (activity, OWN_HISTORY) <-- sees only PrescribeMedication events', flush=True) print(flush=True) wf_client = wf.DaprWorkflowClient() instance_id = wf_client.schedule_new_workflow( - workflow=merchant_checkout, - input=req.to_json(), - instance_id='checkout-001', + workflow=patient_intake, + input=rec.to_json(), + instance_id='intake-001', ) print(f' [main] Started workflow instance: {instance_id}', flush=True) From 7721d061a3a22b58dcb6ca21e525b2c362563d59 Mon Sep 17 00:00:00 2001 From: Nelson Parente Date: Thu, 28 May 2026 22:26:07 +0100 Subject: [PATCH 3/5] feat(python): align history propagation quickstart with canonical Go Aligns the Python workflow history propagation quickstart with Cassie's canonical Go version that merged into release-1.18 (dapr/quickstarts#1315, tutorials/workflow/go/history-propagation). Changes to match the canonical structure: - Move from workflows/python/sdk-context-propagation/order-processor/ to tutorials/workflow/python/history-propagation/, matching the Go sibling at tutorials/workflow/go/history-propagation/ - Split the monolithic app.py into app.py / workflow.py / models.py, mirroring main.go / workflow.go / models.go - Add ForwardLineage flag to PatientRecord and run both scenarios back-to-back (happy path + negative), purging state after each so the app exits on its own - Make DispenseMedication refuse to dispense when no propagated history is received, returning a refused DispenseResult with a Reason field - Match the Go README: title, propagation-scope table, STEP markers, expected output for both scenarios - dapr.yaml: appID=patient-app (was order-processor), resourcesPath=../../resources (was ../../components), appLogDestination/daprdLogDestination=console to match sister Python tutorials and Cassie's Go example - Use correct Python SDK API names: get_last_workflow_by_name / get_last_activity_by_name (the earlier draft used get_workflow_by_name / get_activity_by_name, which were renamed in dapr/python-sdk#1025) Signed-off-by: Nelson Parente --- .../python/history-propagation/README.md | 198 +++++++ .../python/history-propagation/app.py | 143 +++++ .../python/history-propagation/dapr.yaml | 9 + .../python/history-propagation/makefile | 2 + .../python/history-propagation/models.py | 85 +++ .../history-propagation/requirements.txt | 2 + .../python/history-propagation/workflow.py | 462 ++++++++++++++++ .../python/sdk-context-propagation/README.md | 165 ------ .../python/sdk-context-propagation/dapr.yaml | 7 - .../order-processor/app.py | 499 ------------------ .../order-processor/requirements.txt | 2 - 11 files changed, 901 insertions(+), 673 deletions(-) create mode 100644 tutorials/workflow/python/history-propagation/README.md create mode 100644 tutorials/workflow/python/history-propagation/app.py create mode 100644 tutorials/workflow/python/history-propagation/dapr.yaml create mode 100644 tutorials/workflow/python/history-propagation/makefile create mode 100644 tutorials/workflow/python/history-propagation/models.py create mode 100644 tutorials/workflow/python/history-propagation/requirements.txt create mode 100644 tutorials/workflow/python/history-propagation/workflow.py delete mode 100644 workflows/python/sdk-context-propagation/README.md delete mode 100644 workflows/python/sdk-context-propagation/dapr.yaml delete mode 100644 workflows/python/sdk-context-propagation/order-processor/app.py delete mode 100644 workflows/python/sdk-context-propagation/order-processor/requirements.txt diff --git a/tutorials/workflow/python/history-propagation/README.md b/tutorials/workflow/python/history-propagation/README.md new file mode 100644 index 000000000..4d81da50c --- /dev/null +++ b/tutorials/workflow/python/history-propagation/README.md @@ -0,0 +1,198 @@ +# Dapr Workflow History Propagation — Patient Intake + +This example demonstrates how Dapr workflows can propagate their execution +history to child workflows and activities, so downstream consumers can +inspect the full (or partial) execution context of their caller. See the +[Workflow history propagation](https://docs.dapr.io/developing-applications/building-blocks/workflow/workflow-history-propagation/) +docs for the concept overview. + +The scenario is a patient intake / e-prescribing pipeline: a compliance +audit and a pharmacy dispense step refuse to act unless they can see +proof — in the propagated history — that the required upstream checks +(insurance, allergies, drug interactions) actually ran. + +This is the Python sibling of the canonical Go quickstart in +[`tutorials/workflow/go/history-propagation`](../../go/history-propagation), +itself based on [dapr/go-sdk#823](https://github.com/dapr/go-sdk/pull/823). +Python runtime support landed in +[dapr/python-sdk#1025](https://github.com/dapr/python-sdk/pull/1025). + +## Workflow architecture + +``` +PatientIntake (workflow) +├── VerifyInsurance (activity, no propagation) +└── PrescribeMedication (child workflow, PropagationScope.LINEAGE) + ├── CheckAllergies (activity, no propagation) + ├── ScreenDrugInteractions (activity, no propagation) + ├── ComplianceAudit (child workflow, PropagationScope.LINEAGE) + │ → sees PatientIntake + PrescribeMedication events + └── DispenseMedication (activity, PropagationScope.OWN_HISTORY) + → sees PrescribeMedication events only + → refuses to dispense if the screening lineage is missing +``` + +### Propagation scope + +| Mode | What it sends | Use case | +|------|---------------|----------| +| `PropagationScope.LINEAGE` | Caller's own events + any ancestor events it received | Full chain-of-custody verification (compliance audits) | +| `PropagationScope.OWN_HISTORY` | Caller's own events only (no ancestor chain) | Trust boundary — downstream only sees the immediate caller (pharmacy dispense) | + +### Key demonstration + +- **ComplianceAudit** receives the full lineage via `PropagationScope.LINEAGE` — + it verifies that `VerifyInsurance` ran in the grandparent workflow + (PatientIntake), plus `CheckAllergies` and `ScreenDrugInteractions` + ran in PrescribeMedication. + +- **DispenseMedication** receives only PrescribeMedication's history via + `PropagationScope.OWN_HISTORY`. The PatientIntake ancestral history is + excluded — the pharmacy system doesn't need (or get to see) the upstream + chain. Before dispensing, the pharmacy verifies that `CheckAllergies` and + `ScreenDrugInteractions` completed in the propagated history. + +### Scenarios + +The demo runs two scenarios back-to-back to show both the happy path and +the pharmacy's safety check: + +1. **Lineage forwarded → pharmacy dispenses.** `PrescribeMedication` calls + `DispenseMedication` with `PropagationScope.OWN_HISTORY`. The pharmacy + sees the completed allergy and interaction screens in the propagated + history and fills the prescription. + +2. **Lineage withheld → pharmacy refuses.** `PrescribeMedication` calls + `DispenseMedication` **without** history propagation (simulating an + upstream system that fails to forward its lineage). With no propagated + history to prove the prescription was screened, the pharmacy refuses to + dispense and returns a `refused` result explaining what was missing. + +## Python API surface + +```python +# Parent workflow — propagate LINEAGE when calling a child workflow +result = yield ctx.call_child_workflow( + compliance_audit, + input=rec_json, + propagation=wf.PropagationScope.LINEAGE, +) + +# Parent workflow — propagate OWN_HISTORY when calling an activity +dispense = yield ctx.call_activity( + dispense_medication, + input=rec_json, + propagation=wf.PropagationScope.OWN_HISTORY, +) + +# Child workflow (or activity) — read the propagated history +history = ctx.get_propagated_history() # PropagatedHistory | None + +if history is not None: + intake_wf = history.get_last_workflow_by_name('PatientIntake') + insurance = intake_wf.get_last_activity_by_name('VerifyInsurance') + print(insurance.completed) # bool + print(insurance.output) # JSON string +``` + +Key symbols exported from `dapr.ext.workflow`: + +- `PropagationScope` — enum with `LINEAGE` and `OWN_HISTORY` +- `PropagatedHistory` — top-level history object; `.get_workflows()`, + `.get_last_workflow_by_name(name)`, `.events`, `.scope`, `.get_app_ids()` +- `WorkflowResult` — per-workflow slice; `.get_last_activity_by_name(name)` +- `ActivityResult` — `.completed`, `.output` +- `PropagationNotFoundError` — raised when a named workflow/activity is + not present in the history + +> **Replay safety:** workflow code runs many times during durable +> execution. Guard side-effecting calls — including `print()` — with +> `if not ctx.is_replaying:` so they only fire on the live execution. + +## Running this example + +Requires Dapr `1.18.0+` (workflow history propagation), +`dapr-ext-workflow>=1.18.0rc0`, and `dapr>=1.18.0rc0`. + +Install the Python dependencies: + + + +```bash +pip3 install -r requirements.txt && echo "patient-intake deps OK" +``` + + + +Run the demo: + + + +```bash +dapr run -f . +``` + + + +The app runs both scenarios once and exits on its own — no Ctrl+C needed. + +In scenario 1 (lineage forwarded) you'll see the pharmacy dispense: + +``` +[ComplianceAudit] Received propagated history: 15 events (scope: LINEAGE) +[ComplianceAudit] APPROVED (risk=0.10) +[DispenseMedication] Dispense request: amoxicillin 500mg ... (propagated history: 12 events, scope=OWN_HISTORY) +[DispenseMedication] DISPENSED: rx-P-1042-... +``` + +In scenario 2 (lineage withheld) the pharmacy refuses: + +``` +[PrescribeMedication] Step 4: CallActivity(DispenseMedication) + -> NO history propagation (negative scenario) +[DispenseMedication] Dispense request: penicillin 500mg ... (propagated history: none) +[DispenseMedication] REFUSED — no propagated history; cannot verify screening for P-2087 +[PrescribeMedication] Step 4 BLOCKED: pharmacy refused to dispense (missing lineage: no propagated history received from prescriber) +``` + +In standalone mode the sidecar will log +`propagating unsigned workflow history to ...` warnings — these are +expected. Without `WorkflowHistorySigning` enabled, propagated history +chunks aren't cryptographically signed, which is fine for a local +`dapr run` demo. Signing the chunks within an mTLS trust boundary is a +production concern handled at the cluster/control-plane level and is out +of scope for this quickstart. + +## Files + +``` +history-propagation/ +├── README.md # this file +├── dapr.yaml # `dapr run -f .` config (appID, resources, command) +├── makefile # wires the example into `make validate` +├── app.py # registry + worker setup, schedules both scenarios +├── models.py # PatientRecord, ComplianceResult, DispenseResult +├── workflow.py # workflow + activity definitions, history helpers +└── requirements.txt # Python deps +``` diff --git a/tutorials/workflow/python/history-propagation/app.py b/tutorials/workflow/python/history-propagation/app.py new file mode 100644 index 000000000..07d59f6a5 --- /dev/null +++ b/tutorials/workflow/python/history-propagation/app.py @@ -0,0 +1,143 @@ +# -*- coding: utf-8 -*- +# Copyright 2026 The Dapr Authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Workflow history propagation quickstart — patient intake / e-prescribing. + +A root PatientIntake workflow orders a prescription via a child +PrescribeMedication workflow, which in turn runs a ComplianceAudit child +workflow and a DispenseMedication activity. The compliance audit and the +dispensing step inspect the propagated execution history of their callers +to verify that the required upstream checks (insurance, allergies, drug +interactions) actually ran before they make a decision. + +Two scenarios are scheduled back-to-back: + 1. Lineage forwarded — PrescribeMedication propagates its history to + DispenseMedication; the pharmacy dispenses. + 2. Lineage withheld — PrescribeMedication does NOT propagate history; + the pharmacy refuses to dispense. + +Both workflows run to completion and have their state purged so the app +exits on its own — no Ctrl+C needed. +""" + +from __future__ import annotations + +import dapr.ext.workflow as wf + +from models import PatientRecord +from workflow import patient_intake, wfr + + +def _banner(msg: str) -> str: + line = '=' * (len(msg) + 4) + return f'{line}\n= {msg} =\n{line}' + + +def _run_scenario(client: wf.DaprWorkflowClient, title: str, instance_id: str, rec: PatientRecord) -> None: + """Schedule one PatientIntake run, wait for it, print the result, purge state.""" + print(flush=True) + print(_banner(title), flush=True) + + started_id = client.schedule_new_workflow( + workflow=patient_intake, + input=rec.to_json(), + instance_id=instance_id, + ) + print(f' [main] Started workflow: {started_id}', flush=True) + + state = client.wait_for_workflow_completion( + instance_id=started_id, + timeout_in_seconds=30, + ) + if state is None: + print(' [main] Workflow not found!', flush=True) + return + if state.runtime_status.name != 'COMPLETED': + print( + f' [main] Workflow ended with status: {state.runtime_status.name}', + flush=True, + ) + else: + print(f' [main] Result: {state.serialized_output}', flush=True) + + try: + client.purge_workflow(started_id) + except Exception as exc: + print(f' [main] failed to purge: {exc}', flush=True) + + +def main() -> None: + wfr.start() + + print(_banner('WORKFLOW HISTORY PROPAGATION DEMO — PATIENT INTAKE'), flush=True) + print(flush=True) + print(' Flow: PatientIntake -> VerifyInsurance', flush=True) + print(' -> PrescribeMedication (child wf, lineage)', flush=True) + print(' -> CheckAllergies -> ScreenDrugInteractions', flush=True) + print( + ' -> ComplianceAudit (child wf, lineage) ' + '<-- sees PatientIntake + PrescribeMedication events', + flush=True, + ) + print( + ' -> DispenseMedication (activity, own only) ' + '<-- sees only PrescribeMedication events', + flush=True, + ) + + client = wf.DaprWorkflowClient() + + # Scenario 1 (happy path): PrescribeMedication forwards its own history to + # the pharmacy, which verifies the upstream screening and dispenses. + _run_scenario( + client, + 'SCENARIO 1: lineage forwarded — pharmacy dispenses', + 'intake-ok', + PatientRecord( + patient_id='P-1042', + name='Jane Doe', + dob='1985-06-12', + mrn='MRN-77231', + condition='bacterial sinusitis', + medication='amoxicillin', + dosage=500, + forward_lineage=True, + ), + ) + + # Scenario 2 (negative): PrescribeMedication dispenses WITHOUT propagating + # its history, so the pharmacy receives no lineage and refuses to dispense. + _run_scenario( + client, + 'SCENARIO 2: lineage withheld — pharmacy refuses', + 'intake-missing-lineage', + PatientRecord( + patient_id='P-2087', + name='John Roe', + dob='1979-03-04', + mrn='MRN-55810', + condition='strep throat', + medication='penicillin', + dosage=500, + forward_lineage=False, + ), + ) + + print(flush=True) + print(_banner('COMPLETE'), flush=True) + + client.close() + wfr.shutdown() + + +if __name__ == '__main__': + main() diff --git a/tutorials/workflow/python/history-propagation/dapr.yaml b/tutorials/workflow/python/history-propagation/dapr.yaml new file mode 100644 index 000000000..705539ca5 --- /dev/null +++ b/tutorials/workflow/python/history-propagation/dapr.yaml @@ -0,0 +1,9 @@ +version: 1 +common: + resourcesPath: ../../resources +apps: + - appID: patient-app + appDirPath: ./ + command: ["python3", "app.py"] + appLogDestination: console + daprdLogDestination: console diff --git a/tutorials/workflow/python/history-propagation/makefile b/tutorials/workflow/python/history-propagation/makefile new file mode 100644 index 000000000..3170ab7cf --- /dev/null +++ b/tutorials/workflow/python/history-propagation/makefile @@ -0,0 +1,2 @@ +include ../../../../docker.mk +include ../../../../validate.mk diff --git a/tutorials/workflow/python/history-propagation/models.py b/tutorials/workflow/python/history-propagation/models.py new file mode 100644 index 000000000..b4a6bc187 --- /dev/null +++ b/tutorials/workflow/python/history-propagation/models.py @@ -0,0 +1,85 @@ +# -*- coding: utf-8 -*- +# Copyright 2026 The Dapr Authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Data models for the workflow history propagation quickstart.""" + +from __future__ import annotations + +import json +from dataclasses import asdict, dataclass + + +@dataclass +class PatientRecord: + """A patient intake submitted by the front desk. + + Name/DOB/MRN are protected health information (PHI) in a real deployment + and would be candidates for redaction when propagated downstream — a + future addition for history propagation. `forward_lineage` controls + whether PrescribeMedication propagates its own history to the + DispenseMedication activity. + """ + + patient_id: str + name: str + dob: str + mrn: str + condition: str + medication: str + dosage: float + forward_lineage: bool + + def to_json(self) -> str: + return json.dumps(asdict(self)) + + @classmethod + def from_json(cls, data: str) -> 'PatientRecord': + return cls(**json.loads(data)) + + +@dataclass +class ComplianceResult: + """Output of the ComplianceAudit child workflow.""" + + compliant: bool + risk_score: float + reason: str + event_count: int + + def to_json(self) -> str: + return json.dumps(asdict(self)) + + @classmethod + def from_json(cls, data: str) -> 'ComplianceResult': + return cls(**json.loads(data)) + + +@dataclass +class DispenseResult: + """Output of the DispenseMedication activity. + + `status` is ``"dispensed"`` when the pharmacy filled the prescription, or + ``"refused"`` when it could not verify the prescribing pipeline in the + propagated history (``reason`` explains what was missing). + """ + + dispense_id: str + status: str + reason: str + event_count: int + + def to_json(self) -> str: + return json.dumps(asdict(self)) + + @classmethod + def from_json(cls, data: str) -> 'DispenseResult': + return cls(**json.loads(data)) diff --git a/tutorials/workflow/python/history-propagation/requirements.txt b/tutorials/workflow/python/history-propagation/requirements.txt new file mode 100644 index 000000000..591b1d704 --- /dev/null +++ b/tutorials/workflow/python/history-propagation/requirements.txt @@ -0,0 +1,2 @@ +dapr==1.18.0rc0 +dapr-ext-workflow==1.18.0rc0 diff --git a/tutorials/workflow/python/history-propagation/workflow.py b/tutorials/workflow/python/history-propagation/workflow.py new file mode 100644 index 000000000..2eaea04ce --- /dev/null +++ b/tutorials/workflow/python/history-propagation/workflow.py @@ -0,0 +1,462 @@ +# -*- coding: utf-8 -*- +# Copyright 2026 The Dapr Authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Workflow + activity definitions for the history propagation quickstart. + +PatientIntake (root workflow) + |- VerifyInsurance (activity, no propagation) + `- PrescribeMedication (child workflow, PropagationScope.LINEAGE) + |- CheckAllergies (activity, no propagation) + |- ScreenDrugInteractions (activity, no propagation) + |- ComplianceAudit (grandchild wf, PropagationScope.LINEAGE) + | reads: PatientIntake/VerifyInsurance + | PrescribeMedication/CheckAllergies + | PrescribeMedication/ScreenDrugInteractions + `- DispenseMedication (activity, PropagationScope.OWN_HISTORY + when forward_lineage=True, else no propagation) + reads: PrescribeMedication events only (no PatientIntake chain) +""" + +from __future__ import annotations + +import time + +import dapr.ext.workflow as wf + +from models import ComplianceResult, DispenseResult, PatientRecord + +wfr = wf.WorkflowRuntime() + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _describe_scope(scope: wf.PropagationScope | None) -> str: + if scope is None: + return 'NONE' + return scope.name + + +def _describe_history(history: wf.PropagatedHistory | None) -> str: + if history is None: + return 'none' + return f'{len(history.events)} events, scope={_describe_scope(history.scope)}' + + +# --------------------------------------------------------------------------- +# Activities +# --------------------------------------------------------------------------- + + +@wfr.activity(name='VerifyInsurance') +def verify_insurance(ctx: wf.WorkflowActivityContext, rec_json: str) -> bool: + """Check the patient has active coverage on file.""" + rec = PatientRecord.from_json(rec_json) + print(f' [VerifyInsurance] Checking coverage for patient {rec.patient_id}', flush=True) + return True + + +@wfr.activity(name='CheckAllergies') +def check_allergies(ctx: wf.WorkflowActivityContext, rec_json: str) -> bool: + """Look up the patient's allergy list and clear or block the prescription.""" + rec = PatientRecord.from_json(rec_json) + history = ctx.get_propagated_history() + print( + f' [CheckAllergies] Screening {rec.patient_id} for {rec.medication} ' + f'(propagated history: {_describe_history(history)})', + flush=True, + ) + return True + + +@wfr.activity(name='ScreenDrugInteractions') +def screen_drug_interactions(ctx: wf.WorkflowActivityContext, rec_json: str) -> bool: + """Check the candidate prescription against the patient's active medication list.""" + rec = PatientRecord.from_json(rec_json) + history = ctx.get_propagated_history() + print( + f' [ScreenDrugInteractions] Screening {rec.medication} {rec.dosage:.0f}mg ' + f'for {rec.patient_id} (propagated history: {_describe_history(history)})', + flush=True, + ) + return True + + +@wfr.activity(name='DispenseMedication') +def dispense_medication(ctx: wf.WorkflowActivityContext, rec_json: str) -> str: + """Issue the final prescription. + + Pharmacy policy: no lineage, no dispense. Without propagated history the + pharmacy cannot prove the prescription was screened, so it refuses. + """ + rec = PatientRecord.from_json(rec_json) + history = ctx.get_propagated_history() + print( + f' [DispenseMedication] Dispense request: {rec.medication} {rec.dosage:.0f}mg ' + f'for {rec.patient_id} (propagated history: {_describe_history(history)})', + flush=True, + ) + + if history is None: + print( + f' [DispenseMedication] REFUSED — no propagated history; ' + f'cannot verify screening for {rec.patient_id}', + flush=True, + ) + refused = DispenseResult( + dispense_id='', + status='refused', + reason='missing lineage: no propagated history received from prescriber', + event_count=0, + ) + return refused.to_json() + + event_count = len(history.events) + print(f' [DispenseMedication] Apps in chain: {history.get_app_ids()}', flush=True) + for wf_result in history.get_workflows(): + print( + f' [DispenseMedication] workflow: app={wf_result.app_id}, ' + f'name={wf_result.name}, instance={wf_result.instance_id}', + flush=True, + ) + + try: + prescribe_wf = history.get_last_workflow_by_name('PrescribeMedication') + except wf.PropagationNotFoundError: + print( + f' [DispenseMedication] REFUSED — propagated history is missing the ' + f'PrescribeMedication lineage for {rec.patient_id}', + flush=True, + ) + refused = DispenseResult( + dispense_id='', + status='refused', + reason='missing lineage: PrescribeMedication not present in propagated history', + event_count=event_count, + ) + return refused.to_json() + + try: + allergies_act = prescribe_wf.get_last_activity_by_name('CheckAllergies') + interactions_act = prescribe_wf.get_last_activity_by_name('ScreenDrugInteractions') + except wf.PropagationNotFoundError: + print( + f' [DispenseMedication] REFUSED — required screening not verified in ' + f'propagated history for {rec.patient_id}', + flush=True, + ) + refused = DispenseResult( + dispense_id='', + status='refused', + reason='missing lineage: allergy/interaction screening not verified in propagated history', + event_count=event_count, + ) + return refused.to_json() + + if not (allergies_act.completed and interactions_act.completed): + print( + f' [DispenseMedication] REFUSED — required screening not verified in ' + f'propagated history for {rec.patient_id}', + flush=True, + ) + refused = DispenseResult( + dispense_id='', + status='refused', + reason='missing lineage: allergy/interaction screening not verified in propagated history', + event_count=event_count, + ) + return refused.to_json() + + dispense_id = f'rx-{rec.patient_id}-{int(time.time() * 1000)}' + print(f' [DispenseMedication] DISPENSED: {dispense_id}', flush=True) + dispensed = DispenseResult( + dispense_id=dispense_id, + status='dispensed', + reason='', + event_count=event_count, + ) + return dispensed.to_json() + + +# --------------------------------------------------------------------------- +# Child workflows +# --------------------------------------------------------------------------- + + +@wfr.workflow(name='ComplianceAudit') +def compliance_audit(ctx: wf.DaprWorkflowContext, rec_json: str): + """Inspect the parent's propagated history to verify the prescribing pipeline ran. + + Refuses to approve dispensing unless the required upstream steps (insurance, + allergies, interactions) are all present and completed in the propagated history. + """ + rec = PatientRecord.from_json(rec_json) + if not ctx.is_replaying: + print( + f' [ComplianceAudit] Auditing prescription for patient {rec.patient_id}', + flush=True, + ) + + history = ctx.get_propagated_history() + if history is None: + if not ctx.is_replaying: + print(' [ComplianceAudit] WARNING: No propagated history received!', flush=True) + print( + ' [ComplianceAudit] BLOCKED — cannot verify upstream pipeline without history', + flush=True, + ) + no_history = ComplianceResult( + compliant=False, + risk_score=1.0, + reason='no execution history provided — cannot verify caller pipeline', + event_count=0, + ) + return no_history.to_json() + + if not ctx.is_replaying: + print( + f' [ComplianceAudit] Received propagated history: ' + f'{len(history.events)} events (scope: {_describe_scope(history.scope)})', + flush=True, + ) + print(f' [ComplianceAudit] Apps in chain: {history.get_app_ids()}', flush=True) + for wf_result in history.get_workflows(): + print( + f' [ComplianceAudit] workflow: app={wf_result.app_id}, ' + f'name={wf_result.name}, instance={wf_result.instance_id}', + flush=True, + ) + + try: + intake_wf = history.get_last_workflow_by_name('PatientIntake') + prescribe_wf = history.get_last_workflow_by_name('PrescribeMedication') + insurance = intake_wf.get_last_activity_by_name('VerifyInsurance') + allergies = prescribe_wf.get_last_activity_by_name('CheckAllergies') + interactions = prescribe_wf.get_last_activity_by_name('ScreenDrugInteractions') + except wf.PropagationNotFoundError as exc: + if not ctx.is_replaying: + print(f' [ComplianceAudit] BLOCKED — {exc}', flush=True) + missing = ComplianceResult( + compliant=False, + risk_score=0.9, + reason=f'required item missing from propagated history: {exc}', + event_count=len(history.events), + ) + return missing.to_json() + + if not ctx.is_replaying: + print(' [ComplianceAudit] Verification:', flush=True) + print( + f' [ComplianceAudit] PatientIntake/VerifyInsurance: ' + f'completed={insurance.completed}', + flush=True, + ) + print( + f' [ComplianceAudit] PrescribeMedication/CheckAllergies: ' + f'completed={allergies.completed}', + flush=True, + ) + print( + f' [ComplianceAudit] PrescribeMedication/ScreenDrugInteractions: ' + f'completed={interactions.completed}', + flush=True, + ) + + all_completed = insurance.completed and allergies.completed and interactions.completed + if not all_completed: + if not ctx.is_replaying: + print( + ' [ComplianceAudit] BLOCKED — required upstream checks not completed', + flush=True, + ) + blocked = ComplianceResult( + compliant=False, + risk_score=0.9, + reason='required upstream checks not completed in propagated history', + event_count=len(history.events), + ) + return blocked.to_json() + + risk_score = 0.3 if rec.dosage > 1000 else 0.1 + if not ctx.is_replaying: + print(f' [ComplianceAudit] APPROVED (risk={risk_score:.2f})', flush=True) + approved = ComplianceResult( + compliant=True, + risk_score=risk_score, + reason='all upstream checks verified in propagated history', + event_count=len(history.events), + ) + return approved.to_json() + + +@wfr.workflow(name='PrescribeMedication') +def prescribe_medication(ctx: wf.DaprWorkflowContext, rec_json: str): + """Orchestrate an e-prescription. + + Checks the patient's allergies, screens for drug interactions, runs a + compliance audit (as a child wf with full lineage), and dispenses the + medication (as an activity with workflow-level propagation only). + """ + rec = PatientRecord.from_json(rec_json) + if not ctx.is_replaying: + print( + f' [PrescribeMedication] Starting prescription: {rec.medication} ' + f'{rec.dosage:.0f}mg for {rec.condition}', + flush=True, + ) + + # Step 1: Allergy check (no propagation — plain activity) + if not ctx.is_replaying: + print( + ' [PrescribeMedication] Step 1: CallActivity(CheckAllergies) — no propagation', + flush=True, + ) + allergy_clear = yield ctx.call_activity(check_allergies, input=rec_json) + if not allergy_clear: + return 'prescription declined: known allergy' + if not ctx.is_replaying: + print(' [PrescribeMedication] Step 1 complete: allergy clear', flush=True) + + # Step 2: Drug interaction screen (no propagation — plain activity) + if not ctx.is_replaying: + print( + ' [PrescribeMedication] Step 2: CallActivity(ScreenDrugInteractions) ' + '— no propagation', + flush=True, + ) + interactions_clear = yield ctx.call_activity(screen_drug_interactions, input=rec_json) + if not interactions_clear: + return 'prescription declined: drug interaction risk' + if not ctx.is_replaying: + print(' [PrescribeMedication] Step 2 complete: no interactions', flush=True) + + # Step 3: Compliance audit as a child wf. + # PropagationScope.LINEAGE — include our events AND any ancestral history we received. + if not ctx.is_replaying: + print( + ' [PrescribeMedication] Step 3: CallChildWorkflow(ComplianceAudit)', + flush=True, + ) + print( + ' -> propagation=PropagationScope.LINEAGE', + flush=True, + ) + audit_json = yield ctx.call_child_workflow( + compliance_audit, + input=rec_json, + propagation=wf.PropagationScope.LINEAGE, + ) + audit = ComplianceResult.from_json(audit_json) + if not audit.compliant: + return ( + f'prescription blocked: compliance audit failed ' + f'(risk={audit.risk_score:.2f}, reason={audit.reason})' + ) + if not ctx.is_replaying: + print( + f' [PrescribeMedication] Step 3 complete: compliance audit passed ' + f'(risk={audit.risk_score:.2f}, {audit.event_count} events verified)', + flush=True, + ) + + # Step 4: Dispense the medication. + # Happy path: attach PropagationScope.OWN_HISTORY — the pharmacy sees our + # events only (no ancestral chain) and can verify the screens ran. + # Negative scenario (rec.forward_lineage == False): omit propagation, so the + # pharmacy receives no lineage and refuses to dispense. + if not ctx.is_replaying: + if rec.forward_lineage: + print( + ' [PrescribeMedication] Step 4: CallActivity(DispenseMedication)', + flush=True, + ) + print( + ' -> propagation=PropagationScope.OWN_HISTORY', + flush=True, + ) + else: + print( + ' [PrescribeMedication] Step 4: CallActivity(DispenseMedication)', + flush=True, + ) + print( + ' -> NO history propagation (negative scenario)', + flush=True, + ) + if rec.forward_lineage: + dispense_json = yield ctx.call_activity( + dispense_medication, + input=rec_json, + propagation=wf.PropagationScope.OWN_HISTORY, + ) + else: + dispense_json = yield ctx.call_activity(dispense_medication, input=rec_json) + dispense = DispenseResult.from_json(dispense_json) + if dispense.status != 'dispensed': + if not ctx.is_replaying: + print( + f' [PrescribeMedication] Step 4 BLOCKED: pharmacy refused to dispense ' + f'({dispense.reason})', + flush=True, + ) + return f'prescription not dispensed: pharmacy refused ({dispense.reason})' + if not ctx.is_replaying: + print( + f' [PrescribeMedication] Step 4 complete: dispensed ' + f'(id={dispense.dispense_id}, {dispense.event_count} events verified)', + flush=True, + ) + + result = ( + f'dispensed: id={dispense.dispense_id}, patient={rec.patient_id}, ' + f'drug={rec.medication} {rec.dosage:.0f}mg' + ) + if not ctx.is_replaying: + print(f' [PrescribeMedication] COMPLETE: {result}', flush=True) + return result + + +@wfr.workflow(name='PatientIntake') +def patient_intake(ctx: wf.DaprWorkflowContext, rec_json: str): + """Top-level workflow. + + Verifies the patient's insurance, then calls PrescribeMedication as a child + workflow with PropagationScope.LINEAGE, giving PrescribeMedication ancestral + history to forward (or not) downstream. + """ + rec = PatientRecord.from_json(rec_json) + if not ctx.is_replaying: + print(f' [PatientIntake] Starting intake for patient {rec.patient_id}', flush=True) + print( + ' [PatientIntake] Step 1: CallActivity(VerifyInsurance) — no propagation', + flush=True, + ) + insured = yield ctx.call_activity(verify_insurance, input=rec_json) + if not insured: + return 'intake declined: insurance not on file' + if not ctx.is_replaying: + print(' [PatientIntake] Step 1 complete: insurance verified', flush=True) + print(' [PatientIntake] Step 2: CallChildWorkflow(PrescribeMedication)', flush=True) + print( + ' -> propagation=PropagationScope.LINEAGE', + flush=True, + ) + result = yield ctx.call_child_workflow( + prescribe_medication, + input=rec_json, + propagation=wf.PropagationScope.LINEAGE, + ) + if not ctx.is_replaying: + print(f' [PatientIntake] COMPLETE: {result}', flush=True) + return result diff --git a/workflows/python/sdk-context-propagation/README.md b/workflows/python/sdk-context-propagation/README.md deleted file mode 100644 index 6a34d7a0d..000000000 --- a/workflows/python/sdk-context-propagation/README.md +++ /dev/null @@ -1,165 +0,0 @@ -# Dapr Workflow — Context Propagation (Python SDK) - -This quickstart demonstrates **workflow history propagation**, a new feature in Dapr 1.18 that lets a parent workflow share its execution history with child workflows and activities. Downstream services can inspect that history to make trust-aware decisions — without any external state store or custom messaging. - -> **Runtime requirement**: Dapr 1.18+ ([dapr/dapr#9810](https://github.com/dapr/dapr/pull/9810)) -> **SDK requirement**: `dapr-ext-workflow>=1.18.0rc0` ([dapr/python-sdk#1025](https://github.com/dapr/python-sdk/pull/1025)) -> **Proposal**: [dapr/proposals#102](https://github.com/dapr/proposals/issues/102) - -## What is workflow context propagation? - -When a parent workflow calls a child workflow or activity it can optionally attach a tamper-evident snapshot of its own execution history. The receiver reads that snapshot via `ctx.get_propagated_history()` and queries it by workflow name and activity name — letting it verify that the correct upstream steps ran before it proceeds. - -### Two propagation modes - -| Mode | Constant | What the receiver sees | -|------|----------|----------------------| -| **Own history** | `PropagationScope.OWN_HISTORY` | Only the direct caller's events | -| **Lineage** | `PropagationScope.LINEAGE` | Caller's events **plus** any ancestor history the caller itself received | - -## Scenario: Patient intake / e-prescribing - -A compliance audit and a pharmacy dispense step refuse to act unless the propagated history proves the required upstream checks (insurance, allergies, drug interactions) actually ran. - -``` -PatientIntake (root) - └─ VerifyInsurance (activity, no propagation) - └─ PrescribeMedication (child wf, LINEAGE) - └─ CheckAllergies (activity, no propagation) - └─ ScreenDrugInteractions (activity, no propagation) - └─ ComplianceAudit (grandchild wf, LINEAGE) - | reads PatientIntake/VerifyInsurance - | PrescribeMedication/CheckAllergies - | PrescribeMedication/ScreenDrugInteractions - └─ DispenseMedication (activity, OWN_HISTORY) - reads PrescribeMedication events only -``` - -`ComplianceAudit` uses `PropagationScope.LINEAGE` to see the **full ancestor chain** — it can verify both the insurance check (performed by the grandparent `PatientIntake`) and the allergy/interaction checks (performed by the parent `PrescribeMedication`) before approving the prescription. - -`DispenseMedication` uses `PropagationScope.OWN_HISTORY` to see only the **direct caller's events** — a trust-boundary mode that limits visibility to what `PrescribeMedication` itself executed. The pharmacy dispense system doesn't need (or get to see) the upstream patient-intake chain. - -This sample mirrors the canonical Go reference [dapr/go-sdk#823](https://github.com/dapr/go-sdk/pull/823) and the [Go quickstart](https://github.com/dapr/quickstarts/pull/1315). - -## Python API surface - -```python -# Parent workflow — propagate LINEAGE when calling a child workflow -result = yield ctx.call_child_workflow( - compliance_audit, - input=rec_json, - propagation=wf.PropagationScope.LINEAGE, -) - -# Parent workflow — propagate OWN_HISTORY when calling an activity -dispense = yield ctx.call_activity( - dispense_medication, - input=rec_json, - propagation=wf.PropagationScope.OWN_HISTORY, -) - -# Child workflow (or activity) — read the propagated history -history = ctx.get_propagated_history() # returns PropagatedHistory | None - -if history is not None: - intake_wf = history.get_workflow_by_name('PatientIntake') # raises PropagationNotFoundError if missing - insurance = intake_wf.get_activity_by_name('VerifyInsurance') - print(insurance.completed) # bool - print(insurance.output) # JSON string -``` - -Key types exported from `dapr.ext.workflow`: -- `PropagationScope` — enum with `LINEAGE` and `OWN_HISTORY` -- `PropagatedHistory` — top-level history object; call `.get_workflows()` or `.get_workflow_by_name(name)` -- `WorkflowResult` — per-workflow slice; call `.get_activity_by_name(name)` or `.get_child_workflow_by_name(name)` -- `ActivityResult` — has `.completed`, `.output` fields -- `PropagationNotFoundError` — raised when a named workflow/activity is not in the history - -> **Replay safety**: workflow code runs many times during durable execution. Guard side-effecting calls — including `print()` — with `if not ctx.is_replaying:` so they only fire on the live execution, not on each replay. - -## Prerequisites - -- [Dapr CLI](https://docs.dapr.io/getting-started/install-dapr-cli/) 1.18+ -- Dapr runtime 1.18+ initialized (`dapr init`) -- Python 3.9+ -- Redis (started automatically by `dapr init`) - -## Run the sample - -```sh -cd workflows/python/sdk-context-propagation - -# Install dependencies -pip3 install -r order-processor/requirements.txt - -# Run with Dapr -dapr run -f . -``` - -## Expected output - -``` -============================================================ -= WORKFLOW HISTORY PROPAGATION DEMO — PATIENT INTAKE = -============================================================ - - Flow: PatientIntake -> VerifyInsurance - -> PrescribeMedication (child wf, LINEAGE) - -> CheckAllergies -> ScreenDrugInteractions - -> ComplianceAudit (child wf, LINEAGE) <-- sees PatientIntake + PrescribeMedication events - -> DispenseMedication (activity, OWN_HISTORY) <-- sees only PrescribeMedication events - - [main] Started workflow instance: intake-001 - [PatientIntake] Starting intake for patient P-1042 - [PatientIntake] Step 1: VerifyInsurance (no propagation) - [VerifyInsurance] Checking coverage for patient P-1042 - [PatientIntake] Step 1 complete: insurance verified - [PatientIntake] Step 2: PrescribeMedication child wf (PropagationScope.LINEAGE) - [PrescribeMedication] Starting prescription: amoxicillin 500mg for bacterial sinusitis - [PrescribeMedication] Step 1: CheckAllergies (no propagation) - [CheckAllergies] Screening P-1042 for amoxicillin - [PrescribeMedication] Step 1 complete: allergy clear - [PrescribeMedication] Step 2: ScreenDrugInteractions (no propagation) - [ScreenDrugInteractions] Screening amoxicillin 500mg for P-1042 - [PrescribeMedication] Step 2 complete: no interactions - [PrescribeMedication] Step 3: ComplianceAudit child wf (PropagationScope.LINEAGE) - [ComplianceAudit] Auditing prescription for patient P-1042 - [ComplianceAudit] Received propagated history with workflows: ['PatientIntake', 'PrescribeMedication'] - [ComplianceAudit] Verification: - PatientIntake/VerifyInsurance: completed=True - PrescribeMedication/CheckAllergies: completed=True - PrescribeMedication/ScreenDrugInteractions: completed=True - [ComplianceAudit] APPROVED (risk=0.10) - [PrescribeMedication] Step 3 complete: compliance audit passed (risk=0.10, 2 workflow(s) verified) - [PrescribeMedication] Step 4: DispenseMedication (PropagationScope.OWN_HISTORY) - [DispenseMedication] Propagated workflows: ['PrescribeMedication'] - [DispenseMedication] workflow: name=PrescribeMedication app=order-processor - [DispenseMedication] DISPENSED: rx-P-1042-... (amoxicillin 500mg) - [PrescribeMedication] Step 4 complete: dispensed (id=rx-P-1042-..., 1 workflow(s) verified) - [PrescribeMedication] COMPLETE: dispensed: id=rx-P-1042-..., patient=P-1042, drug=amoxicillin 500mg - [PatientIntake] COMPLETE: dispensed: id=rx-P-1042-..., patient=P-1042, drug=amoxicillin 500mg - [main] Workflow completed! Output: "dispensed: ..." - -================ -= COMPLETE = -================ -``` - -## Standalone-mode note - -In standalone mode the sidecar will log `propagating unsigned workflow history to ...` warnings — these are expected. Without `WorkflowHistorySigning` enabled, propagated history chunks aren't cryptographically signed, which is fine for a local `dapr run` demo. Signing the chunks within an mTLS trust boundary is a production concern handled at the cluster/control-plane level and is out of scope for this quickstart. - -## Stop the sample - -```sh -dapr stop -f . -``` - -## References - -- [Proposal: Workflow History Propagation (dapr/proposals#102)](https://github.com/dapr/proposals/issues/102) -- [Runtime PR: dapr/dapr#9810](https://github.com/dapr/dapr/pull/9810) -- [Python SDK PR: dapr/python-sdk#1025](https://github.com/dapr/python-sdk/pull/1025) -- [Canonical Go SDK reference: dapr/go-sdk#823](https://github.com/dapr/go-sdk/pull/823) -- [Sibling Go quickstart: dapr/quickstarts#1315](https://github.com/dapr/quickstarts/pull/1315) -- [Dapr Workflow documentation](https://docs.dapr.io/developing-applications/building-blocks/workflow/) diff --git a/workflows/python/sdk-context-propagation/dapr.yaml b/workflows/python/sdk-context-propagation/dapr.yaml deleted file mode 100644 index e05735188..000000000 --- a/workflows/python/sdk-context-propagation/dapr.yaml +++ /dev/null @@ -1,7 +0,0 @@ -version: 1 -common: - resourcesPath: ../../components -apps: - - appID: order-processor - appDirPath: ./order-processor/ - command: ["python3", "app.py"] diff --git a/workflows/python/sdk-context-propagation/order-processor/app.py b/workflows/python/sdk-context-propagation/order-processor/app.py deleted file mode 100644 index 1c282c469..000000000 --- a/workflows/python/sdk-context-propagation/order-processor/app.py +++ /dev/null @@ -1,499 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2026 The Dapr Authors -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# http://www.apache.org/licenses/LICENSE-2.0 -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""Workflow history propagation quickstart (Python SDK). - -Scenario: patient intake / e-prescribing pipeline. A compliance audit and a -pharmacy dispense step refuse to act unless the propagated history proves -the required upstream checks (insurance, allergies, drug interactions) -actually ran. - -Flow: - PatientIntake (root workflow) - |- VerifyInsurance (activity, no propagation) - `- PrescribeMedication (child workflow, PropagationScope.LINEAGE) - |- CheckAllergies (activity, no propagation) - |- ScreenDrugInteractions (activity, no propagation) - |- ComplianceAudit (grandchild wf, PropagationScope.LINEAGE) - | reads: PatientIntake/VerifyInsurance - | PrescribeMedication/CheckAllergies - | PrescribeMedication/ScreenDrugInteractions - `- DispenseMedication (activity, PropagationScope.OWN_HISTORY) - reads: PrescribeMedication events only (no PatientIntake) - -This requires Dapr 1.18+ (dapr/dapr#9810) and dapr-ext-workflow 1.18+ -(dapr/python-sdk#1025). Against an older sidecar the propagation field is -silently dropped and ctx.get_propagated_history() returns None. -""" - -from __future__ import annotations - -import json -import time -from dataclasses import asdict, dataclass - -import dapr.ext.workflow as wf - -# --------------------------------------------------------------------------- -# Data models -# --------------------------------------------------------------------------- - -@dataclass -class PatientRecord: - patient_id: str - name: str - dob: str - mrn: str - condition: str - medication: str - dosage: float - - def to_json(self) -> str: - return json.dumps(asdict(self)) - - @classmethod - def from_json(cls, data: str) -> "PatientRecord": - return cls(**json.loads(data)) - - -@dataclass -class ComplianceResult: - compliant: bool - risk_score: float - reason: str - event_count: int - - -@dataclass -class DispenseResult: - dispense_id: str - status: str - event_count: int - - -# --------------------------------------------------------------------------- -# Workflow runtime -# --------------------------------------------------------------------------- - -wfr = wf.WorkflowRuntime() - - -# --------------------------------------------------------------------------- -# Activities -# --------------------------------------------------------------------------- - -@wfr.activity(name='VerifyInsurance') -def verify_insurance(ctx: wf.WorkflowActivityContext, rec_json: str) -> bool: - rec = PatientRecord.from_json(rec_json) - print(f' [VerifyInsurance] Checking coverage for patient {rec.patient_id}', flush=True) - return True - - -@wfr.activity(name='CheckAllergies') -def check_allergies(ctx: wf.WorkflowActivityContext, rec_json: str) -> bool: - rec = PatientRecord.from_json(rec_json) - print( - f' [CheckAllergies] Screening {rec.patient_id} for {rec.medication}', - flush=True, - ) - return True - - -@wfr.activity(name='ScreenDrugInteractions') -def screen_drug_interactions(ctx: wf.WorkflowActivityContext, rec_json: str) -> bool: - rec = PatientRecord.from_json(rec_json) - print( - f' [ScreenDrugInteractions] Screening {rec.medication} {rec.dosage:.0f}mg for {rec.patient_id}', - flush=True, - ) - return True - - -@wfr.activity(name='DispenseMedication') -def dispense_medication(ctx: wf.WorkflowActivityContext, rec_json: str) -> str: - """Receives PropagationScope.OWN_HISTORY from PrescribeMedication — sees only - the PrescribeMedication workflow's events, not the PatientIntake ancestor. - - The pharmacy dispense system intentionally does not get to see the - upstream patient-intake chain; it only needs proof that the prescribing - workflow itself ran the right checks. - """ - rec = PatientRecord.from_json(rec_json) - history = ctx.get_propagated_history() - - event_count = 0 - if history is not None: - workflows = history.get_workflows() - event_count = len(workflows) - print( - f' [DispenseMedication] Propagated workflows: {[w.name for w in workflows]}', - flush=True, - ) - for wf_result in workflows: - print( - f' [DispenseMedication] workflow: name={wf_result.name} app={wf_result.app_id}', - flush=True, - ) - else: - print(' [DispenseMedication] No propagated history received', flush=True) - - dispense_id = f'rx-{rec.patient_id}-{int(time.time() * 1000)}' - print( - f' [DispenseMedication] DISPENSED: {dispense_id} ({rec.medication} {rec.dosage:.0f}mg)', - flush=True, - ) - return json.dumps(asdict(DispenseResult( - dispense_id=dispense_id, - status='dispensed', - event_count=event_count, - ))) - - -# --------------------------------------------------------------------------- -# Child workflows -# --------------------------------------------------------------------------- - -@wfr.workflow(name='ComplianceAudit') -def compliance_audit(ctx: wf.DaprWorkflowContext, rec_json: str): - """Grandchild workflow that inspects the full ancestor chain. - - Receives PropagationScope.LINEAGE from PrescribeMedication, so its - get_propagated_history() contains events from both PatientIntake and - PrescribeMedication. It refuses to approve dispensing unless the required - upstream steps (insurance, allergies, drug interactions) are all present - and completed in the propagated history. - """ - rec = PatientRecord.from_json(rec_json) - if not ctx.is_replaying: - print( - f' [ComplianceAudit] Auditing prescription for patient {rec.patient_id}', - flush=True, - ) - - history = ctx.get_propagated_history() - if history is None: - if not ctx.is_replaying: - print( - ' [ComplianceAudit] WARNING: no propagated history — sidecar may not support 1.18+', - flush=True, - ) - print( - ' [ComplianceAudit] BLOCKED — cannot verify upstream pipeline without history', - flush=True, - ) - return json.dumps(asdict(ComplianceResult( - compliant=False, - risk_score=1.0, - reason='no execution history provided — cannot verify caller pipeline', - event_count=0, - ))) - - workflows = history.get_workflows() - if not ctx.is_replaying: - print( - f' [ComplianceAudit] Received propagated history with workflows: ' - f'{[w.name for w in workflows]}', - flush=True, - ) - - # Verify the ancestor chain includes the required workflows. - try: - intake_wf = history.get_workflow_by_name('PatientIntake') - except wf.PropagationNotFoundError: - return json.dumps(asdict(ComplianceResult( - compliant=False, - risk_score=0.9, - reason='PatientIntake missing from propagated history', - event_count=len(workflows), - ))) - - try: - prescribe_wf = history.get_workflow_by_name('PrescribeMedication') - except wf.PropagationNotFoundError: - return json.dumps(asdict(ComplianceResult( - compliant=False, - risk_score=0.9, - reason='PrescribeMedication missing from propagated history', - event_count=len(workflows), - ))) - - # Verify the required activity completions are recorded. - try: - insurance_act = intake_wf.get_activity_by_name('VerifyInsurance') - except wf.PropagationNotFoundError: - return json.dumps(asdict(ComplianceResult( - compliant=False, - risk_score=0.85, - reason='VerifyInsurance not found in PatientIntake history', - event_count=len(workflows), - ))) - - try: - allergies_act = prescribe_wf.get_activity_by_name('CheckAllergies') - interactions_act = prescribe_wf.get_activity_by_name('ScreenDrugInteractions') - except wf.PropagationNotFoundError as exc: - return json.dumps(asdict(ComplianceResult( - compliant=False, - risk_score=0.85, - reason=f'required activity missing from PrescribeMedication history: {exc}', - event_count=len(workflows), - ))) - - if not ctx.is_replaying: - print(' [ComplianceAudit] Verification:', flush=True) - print( - f' PatientIntake/VerifyInsurance: completed={insurance_act.completed}', - flush=True, - ) - print( - f' PrescribeMedication/CheckAllergies: completed={allergies_act.completed}', - flush=True, - ) - print( - f' PrescribeMedication/ScreenDrugInteractions: completed={interactions_act.completed}', - flush=True, - ) - - if not (insurance_act.completed and allergies_act.completed and interactions_act.completed): - if not ctx.is_replaying: - print( - ' [ComplianceAudit] BLOCKED — required upstream checks not completed', - flush=True, - ) - return json.dumps(asdict(ComplianceResult( - compliant=False, - risk_score=0.9, - reason='required upstream checks not completed in propagated history', - event_count=len(workflows), - ))) - - risk_score = 0.3 if rec.dosage > 1000 else 0.1 - if not ctx.is_replaying: - print(f' [ComplianceAudit] APPROVED (risk={risk_score:.2f})', flush=True) - return json.dumps(asdict(ComplianceResult( - compliant=True, - risk_score=risk_score, - reason='all upstream checks verified in propagated history', - event_count=len(workflows), - ))) - - -@wfr.workflow(name='PrescribeMedication') -def prescribe_medication(ctx: wf.DaprWorkflowContext, rec_json: str): - """Child workflow — orchestrates allergy + interaction screening, compliance - audit, and dispensing. - - Receives PropagationScope.LINEAGE from PatientIntake, so it holds the full - ancestor chain when calling its own children. Calls ComplianceAudit with - LINEAGE (audit needs to see the grandparent) and DispenseMedication with - OWN_HISTORY (pharmacy only sees the prescribing step, not the intake). - """ - rec = PatientRecord.from_json(rec_json) - if not ctx.is_replaying: - print( - f' [PrescribeMedication] Starting prescription: {rec.medication} ' - f'{rec.dosage:.0f}mg for {rec.condition}', - flush=True, - ) - - # Step 1: Allergy check (no propagation — plain activity) - if not ctx.is_replaying: - print( - ' [PrescribeMedication] Step 1: CheckAllergies (no propagation)', - flush=True, - ) - allergy_clear = yield ctx.call_activity(check_allergies, input=rec_json) - if not allergy_clear: - return 'prescription declined: known allergy' - if not ctx.is_replaying: - print(' [PrescribeMedication] Step 1 complete: allergy clear', flush=True) - - # Step 2: Drug interaction screen (no propagation) - if not ctx.is_replaying: - print( - ' [PrescribeMedication] Step 2: ScreenDrugInteractions (no propagation)', - flush=True, - ) - interactions_clear = yield ctx.call_activity(screen_drug_interactions, input=rec_json) - if not interactions_clear: - return 'prescription declined: drug interaction risk' - if not ctx.is_replaying: - print(' [PrescribeMedication] Step 2 complete: no interactions', flush=True) - - # Step 3: Compliance audit grandchild workflow with LINEAGE propagation. - # The grandchild sees both PatientIntake AND PrescribeMedication events. - if not ctx.is_replaying: - print( - ' [PrescribeMedication] Step 3: ComplianceAudit child wf ' - '(PropagationScope.LINEAGE)', - flush=True, - ) - audit_json = yield ctx.call_child_workflow( - compliance_audit, - input=rec_json, - propagation=wf.PropagationScope.LINEAGE, - ) - audit = ComplianceResult(**json.loads(audit_json)) - if not audit.compliant: - return ( - f'prescription blocked: compliance audit failed ' - f'(risk={audit.risk_score:.2f}, reason={audit.reason})' - ) - if not ctx.is_replaying: - print( - f' [PrescribeMedication] Step 3 complete: compliance audit passed ' - f'(risk={audit.risk_score:.2f}, {audit.event_count} workflow(s) verified)', - flush=True, - ) - - # Step 4: Dispense the medication with OWN_HISTORY propagation. - # DispenseMedication only sees PrescribeMedication's events, not PatientIntake. - if not ctx.is_replaying: - print( - ' [PrescribeMedication] Step 4: DispenseMedication ' - '(PropagationScope.OWN_HISTORY)', - flush=True, - ) - dispense_json = yield ctx.call_activity( - dispense_medication, - input=rec_json, - propagation=wf.PropagationScope.OWN_HISTORY, - ) - dispense = DispenseResult(**json.loads(dispense_json)) - if not ctx.is_replaying: - print( - f' [PrescribeMedication] Step 4 complete: dispensed ' - f'(id={dispense.dispense_id}, {dispense.event_count} workflow(s) verified)', - flush=True, - ) - - result = ( - f'dispensed: id={dispense.dispense_id}, patient={rec.patient_id}, ' - f'drug={rec.medication} {rec.dosage:.0f}mg' - ) - if not ctx.is_replaying: - print(f' [PrescribeMedication] COMPLETE: {result}', flush=True) - return result - - -@wfr.workflow(name='PatientIntake') -def patient_intake(ctx: wf.DaprWorkflowContext, rec_json: str): - """Root workflow — verifies the patient's insurance then delegates the - prescription to a child workflow with full LINEAGE propagation so the - grandchild ComplianceAudit can inspect the complete ancestor chain. - """ - rec = PatientRecord.from_json(rec_json) - if not ctx.is_replaying: - print( - f' [PatientIntake] Starting intake for patient {rec.patient_id}', - flush=True, - ) - - # Step 1: Verify insurance (no propagation — plain activity) - if not ctx.is_replaying: - print( - ' [PatientIntake] Step 1: VerifyInsurance (no propagation)', - flush=True, - ) - insured = yield ctx.call_activity(verify_insurance, input=rec_json) - if not insured: - return 'intake declined: insurance not on file' - if not ctx.is_replaying: - print(' [PatientIntake] Step 1 complete: insurance verified', flush=True) - - # Step 2: Delegate to PrescribeMedication with LINEAGE propagation. - # PrescribeMedication inherits this workflow's history so its own - # grandchild ComplianceAudit can verify the complete ancestor chain. - if not ctx.is_replaying: - print( - ' [PatientIntake] Step 2: PrescribeMedication child wf ' - '(PropagationScope.LINEAGE)', - flush=True, - ) - result = yield ctx.call_child_workflow( - prescribe_medication, - input=rec_json, - propagation=wf.PropagationScope.LINEAGE, - ) - - if not ctx.is_replaying: - print(f' [PatientIntake] COMPLETE: {result}', flush=True) - return result - - -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- - -def _banner(msg: str) -> str: - line = '=' * (len(msg) + 4) - return f'{line}\n= {msg} =\n{line}' - - -# --------------------------------------------------------------------------- -# Entry point -# --------------------------------------------------------------------------- - -if __name__ == '__main__': - wfr.start() - - rec = PatientRecord( - patient_id='P-1042', - name='Jane Doe', - dob='1985-06-12', - mrn='MRN-77231', - condition='bacterial sinusitis', - medication='amoxicillin', - dosage=500, - ) - - print(_banner('WORKFLOW HISTORY PROPAGATION DEMO — PATIENT INTAKE'), flush=True) - print(flush=True) - print(' Flow: PatientIntake -> VerifyInsurance', flush=True) - print(' -> PrescribeMedication (child wf, LINEAGE)', flush=True) - print(' -> CheckAllergies -> ScreenDrugInteractions', flush=True) - print(' -> ComplianceAudit (child wf, LINEAGE) <-- sees PatientIntake + PrescribeMedication events', flush=True) - print(' -> DispenseMedication (activity, OWN_HISTORY) <-- sees only PrescribeMedication events', flush=True) - print(flush=True) - - wf_client = wf.DaprWorkflowClient() - instance_id = wf_client.schedule_new_workflow( - workflow=patient_intake, - input=rec.to_json(), - instance_id='intake-001', - ) - print(f' [main] Started workflow instance: {instance_id}', flush=True) - - try: - state = wf_client.wait_for_workflow_completion( - instance_id=instance_id, - timeout_in_seconds=30, - ) - if state is None: - print(' [main] Workflow not found!', flush=True) - elif state.runtime_status.name == 'COMPLETED': - print( - f' [main] Workflow completed! Output: {state.serialized_output}', - flush=True, - ) - else: - print( - f' [main] Workflow ended with status: {state.runtime_status.name}', - flush=True, - ) - except TimeoutError: - print(' [main] Workflow timed out!', flush=True) - - print(flush=True) - print(_banner('COMPLETE'), flush=True) - - wfr.shutdown() diff --git a/workflows/python/sdk-context-propagation/order-processor/requirements.txt b/workflows/python/sdk-context-propagation/order-processor/requirements.txt deleted file mode 100644 index 4b21b8080..000000000 --- a/workflows/python/sdk-context-propagation/order-processor/requirements.txt +++ /dev/null @@ -1,2 +0,0 @@ -dapr>=1.18.0rc0 -dapr-ext-workflow>=1.18.0rc0 From 015fb51fbb89e6d55d5a9b9ef21ba43357a9c8e3 Mon Sep 17 00:00:00 2001 From: Nelson Parente Date: Fri, 29 May 2026 13:37:56 +0100 Subject: [PATCH 4/5] docs(python): address Alice's review on history-propagation README + code MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - README: drop the Go-sibling reference paragraph - README: simplify the requirements line to just "Requires Dapr 1.18+" - README: drop the "exits on its own — no Ctrl+C needed" sentence - README: drop the "## Files" tree at the bottom - workflow.py: trim repetitive docstrings and inline Step comments; the step prints already carry the same information Signed-off-by: Nelson Parente --- .../python/history-propagation/README.md | 24 +-------- .../python/history-propagation/workflow.py | 51 ++++--------------- 2 files changed, 10 insertions(+), 65 deletions(-) diff --git a/tutorials/workflow/python/history-propagation/README.md b/tutorials/workflow/python/history-propagation/README.md index 4d81da50c..46a52c6ee 100644 --- a/tutorials/workflow/python/history-propagation/README.md +++ b/tutorials/workflow/python/history-propagation/README.md @@ -11,12 +11,6 @@ audit and a pharmacy dispense step refuse to act unless they can see proof — in the propagated history — that the required upstream checks (insurance, allergies, drug interactions) actually ran. -This is the Python sibling of the canonical Go quickstart in -[`tutorials/workflow/go/history-propagation`](../../go/history-propagation), -itself based on [dapr/go-sdk#823](https://github.com/dapr/go-sdk/pull/823). -Python runtime support landed in -[dapr/python-sdk#1025](https://github.com/dapr/python-sdk/pull/1025). - ## Workflow architecture ``` @@ -111,8 +105,7 @@ Key symbols exported from `dapr.ext.workflow`: ## Running this example -Requires Dapr `1.18.0+` (workflow history propagation), -`dapr-ext-workflow>=1.18.0rc0`, and `dapr>=1.18.0rc0`. +Requires Dapr `1.18+`. Install the Python dependencies: @@ -155,8 +148,6 @@ dapr run -f . -The app runs both scenarios once and exits on its own — no Ctrl+C needed. - In scenario 1 (lineage forwarded) you'll see the pharmacy dispense: ``` @@ -183,16 +174,3 @@ chunks aren't cryptographically signed, which is fine for a local `dapr run` demo. Signing the chunks within an mTLS trust boundary is a production concern handled at the cluster/control-plane level and is out of scope for this quickstart. - -## Files - -``` -history-propagation/ -├── README.md # this file -├── dapr.yaml # `dapr run -f .` config (appID, resources, command) -├── makefile # wires the example into `make validate` -├── app.py # registry + worker setup, schedules both scenarios -├── models.py # PatientRecord, ComplianceResult, DispenseResult -├── workflow.py # workflow + activity definitions, history helpers -└── requirements.txt # Python deps -``` diff --git a/tutorials/workflow/python/history-propagation/workflow.py b/tutorials/workflow/python/history-propagation/workflow.py index 2eaea04ce..15826a722 100644 --- a/tutorials/workflow/python/history-propagation/workflow.py +++ b/tutorials/workflow/python/history-propagation/workflow.py @@ -61,7 +61,6 @@ def _describe_history(history: wf.PropagatedHistory | None) -> str: @wfr.activity(name='VerifyInsurance') def verify_insurance(ctx: wf.WorkflowActivityContext, rec_json: str) -> bool: - """Check the patient has active coverage on file.""" rec = PatientRecord.from_json(rec_json) print(f' [VerifyInsurance] Checking coverage for patient {rec.patient_id}', flush=True) return True @@ -69,7 +68,6 @@ def verify_insurance(ctx: wf.WorkflowActivityContext, rec_json: str) -> bool: @wfr.activity(name='CheckAllergies') def check_allergies(ctx: wf.WorkflowActivityContext, rec_json: str) -> bool: - """Look up the patient's allergy list and clear or block the prescription.""" rec = PatientRecord.from_json(rec_json) history = ctx.get_propagated_history() print( @@ -82,7 +80,6 @@ def check_allergies(ctx: wf.WorkflowActivityContext, rec_json: str) -> bool: @wfr.activity(name='ScreenDrugInteractions') def screen_drug_interactions(ctx: wf.WorkflowActivityContext, rec_json: str) -> bool: - """Check the candidate prescription against the patient's active medication list.""" rec = PatientRecord.from_json(rec_json) history = ctx.get_propagated_history() print( @@ -95,11 +92,7 @@ def screen_drug_interactions(ctx: wf.WorkflowActivityContext, rec_json: str) -> @wfr.activity(name='DispenseMedication') def dispense_medication(ctx: wf.WorkflowActivityContext, rec_json: str) -> str: - """Issue the final prescription. - - Pharmacy policy: no lineage, no dispense. Without propagated history the - pharmacy cannot prove the prescription was screened, so it refuses. - """ + """Refuses to dispense unless the propagated history proves screening ran.""" rec = PatientRecord.from_json(rec_json) history = ctx.get_propagated_history() print( @@ -196,11 +189,7 @@ def dispense_medication(ctx: wf.WorkflowActivityContext, rec_json: str) -> str: @wfr.workflow(name='ComplianceAudit') def compliance_audit(ctx: wf.DaprWorkflowContext, rec_json: str): - """Inspect the parent's propagated history to verify the prescribing pipeline ran. - - Refuses to approve dispensing unless the required upstream steps (insurance, - allergies, interactions) are all present and completed in the propagated history. - """ + """Approves only if the propagated history shows insurance + screening completed.""" rec = PatientRecord.from_json(rec_json) if not ctx.is_replaying: print( @@ -302,12 +291,6 @@ def compliance_audit(ctx: wf.DaprWorkflowContext, rec_json: str): @wfr.workflow(name='PrescribeMedication') def prescribe_medication(ctx: wf.DaprWorkflowContext, rec_json: str): - """Orchestrate an e-prescription. - - Checks the patient's allergies, screens for drug interactions, runs a - compliance audit (as a child wf with full lineage), and dispenses the - medication (as an activity with workflow-level propagation only). - """ rec = PatientRecord.from_json(rec_json) if not ctx.is_replaying: print( @@ -316,7 +299,6 @@ def prescribe_medication(ctx: wf.DaprWorkflowContext, rec_json: str): flush=True, ) - # Step 1: Allergy check (no propagation — plain activity) if not ctx.is_replaying: print( ' [PrescribeMedication] Step 1: CallActivity(CheckAllergies) — no propagation', @@ -328,7 +310,6 @@ def prescribe_medication(ctx: wf.DaprWorkflowContext, rec_json: str): if not ctx.is_replaying: print(' [PrescribeMedication] Step 1 complete: allergy clear', flush=True) - # Step 2: Drug interaction screen (no propagation — plain activity) if not ctx.is_replaying: print( ' [PrescribeMedication] Step 2: CallActivity(ScreenDrugInteractions) ' @@ -341,8 +322,6 @@ def prescribe_medication(ctx: wf.DaprWorkflowContext, rec_json: str): if not ctx.is_replaying: print(' [PrescribeMedication] Step 2 complete: no interactions', flush=True) - # Step 3: Compliance audit as a child wf. - # PropagationScope.LINEAGE — include our events AND any ancestral history we received. if not ctx.is_replaying: print( ' [PrescribeMedication] Step 3: CallChildWorkflow(ComplianceAudit)', @@ -370,26 +349,20 @@ def prescribe_medication(ctx: wf.DaprWorkflowContext, rec_json: str): flush=True, ) - # Step 4: Dispense the medication. - # Happy path: attach PropagationScope.OWN_HISTORY — the pharmacy sees our - # events only (no ancestral chain) and can verify the screens ran. - # Negative scenario (rec.forward_lineage == False): omit propagation, so the - # pharmacy receives no lineage and refuses to dispense. + # Step 4 demonstrates the two propagation modes side-by-side: forward_lineage=True + # attaches OWN_HISTORY so the pharmacy can verify upstream screening; + # forward_lineage=False omits propagation, so the pharmacy refuses to dispense. if not ctx.is_replaying: + print( + ' [PrescribeMedication] Step 4: CallActivity(DispenseMedication)', + flush=True, + ) if rec.forward_lineage: - print( - ' [PrescribeMedication] Step 4: CallActivity(DispenseMedication)', - flush=True, - ) print( ' -> propagation=PropagationScope.OWN_HISTORY', flush=True, ) else: - print( - ' [PrescribeMedication] Step 4: CallActivity(DispenseMedication)', - flush=True, - ) print( ' -> NO history propagation (negative scenario)', flush=True, @@ -429,12 +402,6 @@ def prescribe_medication(ctx: wf.DaprWorkflowContext, rec_json: str): @wfr.workflow(name='PatientIntake') def patient_intake(ctx: wf.DaprWorkflowContext, rec_json: str): - """Top-level workflow. - - Verifies the patient's insurance, then calls PrescribeMedication as a child - workflow with PropagationScope.LINEAGE, giving PrescribeMedication ancestral - history to forward (or not) downstream. - """ rec = PatientRecord.from_json(rec_json) if not ctx.is_replaying: print(f' [PatientIntake] Starting intake for patient {rec.patient_id}', flush=True) From 5e9fa6b1c0aaf3e5ae3161cb525da2dc99101322 Mon Sep 17 00:00:00 2001 From: "nelson.parente" Date: Fri, 29 May 2026 16:36:36 +0100 Subject: [PATCH 5/5] docs(python): trim redundant README sections in history-propagation - Merge 'Key demonstration' into 'Scenarios' so the LINEAGE/OWN_HISTORY behavior is explained once instead of three times. - Cut the unsigned-history warning note from a paragraph to one line. --- .../python/history-propagation/README.md | 38 ++++++------------- 1 file changed, 11 insertions(+), 27 deletions(-) diff --git a/tutorials/workflow/python/history-propagation/README.md b/tutorials/workflow/python/history-propagation/README.md index 46a52c6ee..a4f83f074 100644 --- a/tutorials/workflow/python/history-propagation/README.md +++ b/tutorials/workflow/python/history-propagation/README.md @@ -33,32 +33,21 @@ PatientIntake (workflow) | `PropagationScope.LINEAGE` | Caller's own events + any ancestor events it received | Full chain-of-custody verification (compliance audits) | | `PropagationScope.OWN_HISTORY` | Caller's own events only (no ancestor chain) | Trust boundary — downstream only sees the immediate caller (pharmacy dispense) | -### Key demonstration - -- **ComplianceAudit** receives the full lineage via `PropagationScope.LINEAGE` — - it verifies that `VerifyInsurance` ran in the grandparent workflow - (PatientIntake), plus `CheckAllergies` and `ScreenDrugInteractions` - ran in PrescribeMedication. - -- **DispenseMedication** receives only PrescribeMedication's history via - `PropagationScope.OWN_HISTORY`. The PatientIntake ancestral history is - excluded — the pharmacy system doesn't need (or get to see) the upstream - chain. Before dispensing, the pharmacy verifies that `CheckAllergies` and - `ScreenDrugInteractions` completed in the propagated history. - ### Scenarios -The demo runs two scenarios back-to-back to show both the happy path and -the pharmacy's safety check: +`ComplianceAudit` always runs with `PropagationScope.LINEAGE`, so it sees the +full ancestor chain — `VerifyInsurance` from PatientIntake plus `CheckAllergies` +and `ScreenDrugInteractions` from PrescribeMedication — and approves only when +every upstream check completed. The demo then runs `DispenseMedication` twice +to show the `OWN_HISTORY` trust boundary in action: 1. **Lineage forwarded → pharmacy dispenses.** `PrescribeMedication` calls - `DispenseMedication` with `PropagationScope.OWN_HISTORY`. The pharmacy - sees the completed allergy and interaction screens in the propagated - history and fills the prescription. + `DispenseMedication` with `PropagationScope.OWN_HISTORY`. The pharmacy sees + PrescribeMedication's screening events — but not the PatientIntake chain — + and fills the prescription. 2. **Lineage withheld → pharmacy refuses.** `PrescribeMedication` calls - `DispenseMedication` **without** history propagation (simulating an - upstream system that fails to forward its lineage). With no propagated + `DispenseMedication` **without** history propagation. With no propagated history to prove the prescription was screened, the pharmacy refuses to dispense and returns a `refused` result explaining what was missing. @@ -167,10 +156,5 @@ In scenario 2 (lineage withheld) the pharmacy refuses: [PrescribeMedication] Step 4 BLOCKED: pharmacy refused to dispense (missing lineage: no propagated history received from prescriber) ``` -In standalone mode the sidecar will log -`propagating unsigned workflow history to ...` warnings — these are -expected. Without `WorkflowHistorySigning` enabled, propagated history -chunks aren't cryptographically signed, which is fine for a local -`dapr run` demo. Signing the chunks within an mTLS trust boundary is a -production concern handled at the cluster/control-plane level and is out -of scope for this quickstart. +In standalone mode the sidecar logs `propagating unsigned workflow history to ...` +warnings — these are expected and harmless for a local `dapr run` demo.