File size: 4,156 Bytes
ce8c08a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | """
Example: API Workflow Verification
Demonstrates verifying a complex API orchestration plan with
resource conflicts, dependency chains, and mandatory steps.
"""
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from qsvaps import (
Plan,
PlanAction,
PlanConstraint,
ResourceConstraint,
ConstraintType,
PlanVerifier,
QSVAPSVisualizer,
)
def main():
"""Build and verify an API orchestration plan."""
# ── Define the plan ──────────────────────────────────────────────────
plan = Plan(
name="Multi-API Data Aggregation",
actions=[
PlanAction(
name="auth_service_a",
description="Authenticate with Service A",
resources=["auth_tokens"],
can_fail=True,
),
PlanAction(
name="auth_service_b",
description="Authenticate with Service B",
resources=["auth_tokens"],
can_fail=True,
),
PlanAction(
name="fetch_users",
description="Fetch user records from Service A",
resources=["bandwidth"],
can_fail=True,
),
PlanAction(
name="fetch_analytics",
description="Fetch analytics from Service B",
resources=["bandwidth"],
can_fail=True,
),
PlanAction(
name="transform",
description="Transform and join datasets",
resources=["compute"],
can_fail=True,
),
PlanAction(
name="validate",
description="Validate output schema",
can_fail=False, # Must succeed
),
PlanAction(
name="store_results",
description="Store results in database",
resources=["db_connection"],
can_fail=False, # Must succeed
),
],
dependencies=[
("auth_service_a", "fetch_users"),
("auth_service_b", "fetch_analytics"),
("fetch_users", "transform"),
("fetch_analytics", "transform"),
("transform", "validate"),
("validate", "store_results"),
],
resource_constraints=[
ResourceConstraint("auth_tokens", max_concurrent=1),
ResourceConstraint("bandwidth", max_concurrent=1),
],
custom_constraints=[
PlanConstraint(
expression=f"x0 or x1", # At least one auth must work
description="At least one authentication must succeed",
constraint_type=ConstraintType.CUSTOM,
variables_involved=["s_auth_service_a", "s_auth_service_b"],
),
],
)
# ── Verify ───────────────────────────────────────────────────────────
viz = QSVAPSVisualizer()
print(viz.draw_plan(plan))
verifier = PlanVerifier(shots=4096)
result = verifier.verify(plan, verbose=True)
print(viz.format_result(result))
# ── Print summary ────────────────────────────────────────────────────
if not result.is_safe:
print(
f"\n⚠️ This plan has {result.num_violations} potential "
f"failure modes out of {result.total_states:,} possible "
f"execution states."
)
print(
f" Grover's algorithm found them in "
f"{result.grover_iterations} iteration(s) "
f"(vs ~{result.total_states} classical checks)."
)
else:
print("\n✅ Plan is safe under all modeled scenarios.")
if __name__ == "__main__":
main()
|