diff --git a/tutorials/workflow/python/history-propagation/README.md b/tutorials/workflow/python/history-propagation/README.md new file mode 100644 index 000000000..a4f83f074 --- /dev/null +++ b/tutorials/workflow/python/history-propagation/README.md @@ -0,0 +1,160 @@ +# 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. + +## 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) | + +### Scenarios + +`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 + PrescribeMedication's screening events — but not the PatientIntake chain — + and fills the prescription. + +2. **Lineage withheld → pharmacy refuses.** `PrescribeMedication` calls + `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. + +## 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+`. + +Install the Python dependencies: + + + +```bash +pip3 install -r requirements.txt && echo "patient-intake deps OK" +``` + + + +Run the demo: + + + +```bash +dapr run -f . +``` + + + +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 logs `propagating unsigned workflow history to ...` +warnings — these are expected and harmless for a local `dapr run` demo. 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..15826a722 --- /dev/null +++ b/tutorials/workflow/python/history-propagation/workflow.py @@ -0,0 +1,429 @@ +# -*- 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: + 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) + 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: + 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: + """Refuses to dispense unless the propagated history proves screening ran.""" + 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): + """Approves only if the propagated history shows insurance + screening completed.""" + 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): + 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, + ) + + 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) + + 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) + + 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 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( + ' -> propagation=PropagationScope.OWN_HISTORY', + flush=True, + ) + else: + 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): + 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