#!/usr/bin/env python3 """Quattrino - the seven differentiation journeys (A-G), runnable, TEST/SANDBOX only. INTENT -> AUTHORITY -> ACTION -> PROOF Every step prints EXPECTED vs ACTUAL so you can judge the product, not our claims. Nothing here moves real money: agents are sandbox agents, wallets hold test money, execution is "external" (you would execute with your own provider). Prereqs 1. python bootstrap_sandbox.py (creates ./.quattrino_sandbox.json) 2. the owner password in QUATTRINO_PASSWORD (the journeys need the OWNER side for amendments / delegation issuance / disclosure postures - exactly the actions an agent must never be able to do for itself) 3. quattrino.py next to this file (auto-downloaded from the runtime if missing) Run: python differentiation_journeys.py # all seven python differentiation_journeys.py A C G # a subset Journeys A task authority narrows standing authority ($50 standing, "under $20" task -> $20 effective) B vague intent + latest safe question ("Find someone who can fix this machine.") C hidden commitment ($9.99/month is not a $20 one-time purchase) D purpose-bound disclosure (shipping address: once, to the fulfilling merchant only) E conserved multi-agent delegation ($300 -> 175 + 40 + 60; $25 left; $50 refused) F conversational authority amendment (laptop <= $1,000; <= $1,200 only if RAM >= 32GB; no refurbished) G external execution + outcome compliance (authorize, you execute, record outcome, explain) """ import getpass import json import os import sys import time import urllib.error import urllib.request import uuid HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) CFG_PATH = os.environ.get("QUATTRINO_SANDBOX_FILE", os.path.join(os.getcwd(), ".quattrino_sandbox.json")) CFG = json.load(open(CFG_PATH)) if os.path.exists(CFG_PATH) else {} BASE = (os.environ.get("QUATTRINO_BASE") or CFG.get("base") or "https://quattrino.io").rstrip("/") OWNER_EMAIL = os.environ.get("QUATTRINO_EMAIL") or CFG.get("owner_email") OWNER_PASSWORD = os.environ.get("QUATTRINO_PASSWORD") RUN = uuid.uuid4().hex[:6] # ----------------------------------------------------------------------------- sdk def _ensure_sdk(): try: from quattrino import Quattrino # noqa: F401 return except ImportError: pass print("downloading the single-file SDK from", BASE + "/api/v1/public/sdk/python") with urllib.request.urlopen(BASE + "/api/v1/public/sdk/python", timeout=30) as r: open(os.path.join(HERE, "quattrino.py"), "wb").write(r.read()) _ensure_sdk() from quattrino import Quattrino, QuattrinoError # noqa: E402 # --------------------------------------------------------------------------- owner class Owner: """The human/owner side (session token). Agents can never do these things.""" def __init__(self, email, password): st, j = self.call("POST", "/auth/login", {"email": email, "password": password}) if st != 200 or not j.get("access_token"): if j.get("mfa_required"): sys.exit("owner account uses two-factor auth: run the journeys with an account without MFA, " "or perform the owner steps in the web app") sys.exit("owner login failed: " + json.dumps(j)[:300]) self.token = j["access_token"] def call(self, method, path, body=None): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(BASE + "/api/v1" + path, data=data, method=method) req.add_header("Content-Type", "application/json") if getattr(self, "token", None): req.add_header("Authorization", "Bearer " + self.token) try: with urllib.request.urlopen(req, timeout=30) as r: return r.status, json.loads(r.read() or b"{}") except urllib.error.HTTPError as e: try: return e.code, json.loads(e.read() or b"{}") except ValueError: return e.code, {} def make_agent(self, name, *, standing_cents=5000, wallet=True): """agent + ACTIVE standing policy + sandbox wallet + scoped key -> (agent, Quattrino client)""" st, j = self.call("POST", "/agents", {"name": name, "description": "differentiation journeys", "environment": "sandbox"}) agent = j["agent"] st, p = self.call("POST", "/policies", {"name": name + " standing $%.2f" % (standing_cents / 100), "currency": "USD", "transaction_limit": standing_cents, "transaction_auto_limit": standing_cents, "daily_limit": standing_cents * 20, "monthly_limit": standing_cents * 200}) pid = p["policy"]["policy_id"] self.call("POST", "/policies/" + pid + "/activate", {"confirmed": True}) if wallet: self.call("POST", "/wallets", {"display_name": name + " wallet", "wallet_type": "agent", "currency": "USD", "authority_minor": max(standing_cents * 20, 100000), "agent_id": agent["agent_id"], "default_policy_id": pid, "environment": "sandbox"}) self.call("PATCH", "/agents/" + agent["agent_id"], {"default_policy_id": pid}) st, c = self.call("POST", "/agents/" + agent["agent_id"] + "/credentials", {"name": "journeys"}) return agent, Quattrino(api_key=c["credential"]["api_key"], base_url=BASE) def amend_envelope(self, envelope_id, dimensions, note): return self.call("POST", "/org/envelopes/" + envelope_id + "/amend", {"dimensions": dimensions, "note": note}) def grant_root(self, agent_id, ceiling_cents, purpose): from datetime import datetime, timedelta, timezone return self.call("POST", "/delegations", {"agent_id": agent_id, "amount_ceiling_minor": ceiling_cents, "currency": "USD", "purpose": purpose, "expires_at": (datetime.now(timezone.utc) + timedelta(hours=2)).isoformat()}) def delegate(self, parent_authority_id, child_agent_id, ceiling_cents): return self.call("POST", "/delegations/" + parent_authority_id + "/delegate", {"child_agent_id": child_agent_id, "amount_ceiling_minor": ceiling_cents}) # ------------------------------------------------------------------------- report RESULTS = [] def check(journey, what, expected, actual, ok=None): ok = (expected == actual) if ok is None else bool(ok) RESULTS.append((journey, what, ok)) print(" [%s] %s\n expected: %s\n actual: %s" % ("PASS" if ok else "FAIL", what, expected, actual)) return ok def money(minor): return "${:,.2f}".format((minor or 0) / 100) def codes(result, key): return [d.get("code") for d in (result.get(key) or [])] def title(letter, name): print("\n" + "=" * 78 + "\n %s. %s\n" % (letter, name) + "=" * 78) # ------------------------------------------------------------------------ journeys def journey_A(owner): title("A", "TASK AUTHORITY NARROWS STANDING AUTHORITY ($50 standing; task 'under $20')") agent, q = owner.make_agent("J-A cable buyer " + RUN, standing_cents=5000) task = {"max_amount_minor": 2000, "instruction": "Find the best USB-C cable under $20 and buy it."} last = None for amt, want in ((1999, "AUTHORIZED"), (2000, "AUTHORIZED"), (2001, "DENIED")): r = q.authorize(amount_minor=amt, merchant="Amazon", purpose="USB-C cable", commitment="ONE_TIME", task=task) last = r check("A", money(amt) + " under the $20 task", want, r.decision) ex = q.explain(last.authorization_id) check("A", "explain: standing ceiling", "$50.00", money((ex.get("standing_authority") or {}).get("ceiling_minor"))) check("A", "explain: task ceiling", "$20.00", money(((ex.get("narrowing") or {}).get("task") or {}).get("ceiling_minor"))) check("A", "explain: effective ceiling (binding layer)", "$20.00 / task", money(ex.get("effective_maximum_minor")) + " / " + str(ex.get("binding_layer"))) print("\n plain language:", ex.get("plain_language", "")[:400]) # the same narrowing from plain language instead of a structured task c = q.compile_intent("Find the best USB-C cable under $20 and buy it.") check("A", "plain-language task compiles to a $20 envelope ceiling (SAFE_TO_INFER = narrowing only)", 2000, c["envelope"]["dimensions"]["money"]["per_action_ceiling_minor"]) r = q.authorize(envelope_id=c["envelope"]["envelope_id"], amount_minor=2001, merchant="Amazon", purpose="cable", commitment="ONE_TIME") check("A", "$20.01 under the compiled envelope -> DENIED with the precise gap (+$0.01, owner_amendment)", ("DENIED", 1, "owner_amendment"), (r.decision, (r.question or {}).get("requested_delta_minor"), (r.question or {}).get("how_to_obtain"))) def journey_B(owner): title("B", "VAGUE INTENT + LATEST SAFE QUESTION ('Find someone who can fix this machine.')") agent, q = owner.make_agent("J-B fixer " + RUN, standing_cents=50000) c = q.compile_intent("Find someone who can fix this machine.") env = c["envelope"]["envelope_id"] check("B", "research + contacting suppliers authorized NOW", True, {"research_and_compare", "contact_counterparties"} <= set(c["authorized_now"]), ok=True if {"research_and_compare", "contact_counterparties"} <= set(c["authorized_now"]) else False) check("B", "hiring / paying is NOT yet authorized (nothing inferred from 'fix')", True, "hire_supplier_or_agent" in c["not_yet_authorized"]) check("B", "no money ceiling invented", None, c["envelope"]["dimensions"]["money"]["per_action_ceiling_minor"]) r = q.authorize(envelope_id=env, action_class="NON_CONSEQUENTIAL_RESEARCH", purpose="research suppliers") check("B", "research suppliers", ("AUTHORIZED", False), (r.decision, bool(r.question_required))) r = q.authorize(envelope_id=env, action_class="NON_CONSEQUENTIAL_RESEARCH", purpose="compare three quotes") check("B", "compare quotes (still no question asked)", ("AUTHORIZED", False), (r.decision, bool(r.question_required))) r = q.authorize(envelope_id=env, action_type="hire_agent", merchant="Repair Co", amount_minor=32700, purpose="hire the repair supplier", commitment="CONTRACT") check("B", "hire Repair Co for $327 -> NEEDS_INFORMATION (never DENIED, never AUTHORIZED)", "NEEDS_INFORMATION", r.decision) check("B", "missing authority names the hiring class + how to obtain it", True, any(m["dimension"] == "purpose.allowed_action_classes" and m["how_to_obtain"] == "owner_amendment" for m in r.missing_authority)) check("B", "exactly one precise question, only now (consequence point)", True, bool(r.question_required and r.question)) print("\n the question:", (r.question or {}).get("question")) print(" missing_authority:", json.dumps(r.missing_authority)[:600]) print(" safe actions remaining:", (r.question or {}).get("safe_actions_remaining")) def journey_C(owner): title("C", "HIDDEN COMMITMENT ('Buy software for no more than $20 one time.' vs a $9.99/month subscription)") agent, q = owner.make_agent("J-C software " + RUN, standing_cents=5000) c = q.compile_intent("Buy software for no more than $20 one time.") env = c["envelope"]["envelope_id"] check("C", "'one time' compiles to prohibited recurring classes", True, {"SUBSCRIPTION", "RECURRING", "AUTO_RENEWAL"} <= set(c["envelope"]["dimensions"]["commitment"]["prohibited"] or [])) r = q.authorize(envelope_id=env, merchant="SaaSCo", amount_minor=999, purpose="software", commitment="SUBSCRIPTION") check("C", "$9.99/month SUBSCRIPTION (cheaper than $20!)", ("DENIED", True), (r.decision, "commitment_prohibited" in codes(r, "denial_reasons"))) r = q.authorize(envelope_id=env, merchant="SaaSCo", amount_minor=999, purpose="software") check("C", "commitment undeclared -> NEEDS_INFORMATION (UNKNOWN never becomes ONE_TIME)", ("NEEDS_INFORMATION", "UNKNOWN_COMMITMENT"), (r.decision, r.get("commitment"))) auth = q.authorize_binding(envelope_id=env, merchant="SaaSCo", amount_minor=999, purpose="software", commitment="ONE_TIME", execution="external") check("C", "declared ONE_TIME $9.99 -> AUTHORIZED (binding)", "AUTHORIZED", auth.decision) # the executor comes back and admits it started a subscription out = q.record_outcome(auth.authorization_id, success=True, provider="saasco", actual_amount_minor=999, commitment="SUBSCRIPTION", recurring=True) check("C", "outcome says SUBSCRIPTION under a ONE_TIME authorization -> commitment mismatch is a finding", ("NONCOMPLIANT", True), (out["compliance"]["status"], "UNAUTHORIZED_COMMITMENT" in out["compliance"]["findings"])) check("C", "the authorization itself is never rewritten", "ONE_TIME", out.get("commitment")) print("\n explain:", q.explain(auth.authorization_id)["plain_language"][-260:]) def journey_D(owner): title("D", "PURPOSE-BOUND DISCLOSURE (shipping address: once, to the fulfilling merchant only)") agent, q = owner.make_agent("J-D shopper " + RUN, standing_cents=5000) st, pr = owner.call("POST", "/commerce/profiles", { "name": "journey D profile " + RUN, "display_name": "Sandbox Owner", "email": "owner@sandbox-owner.com", "phone": "+15550001234", "shipping_addresses": [{"label": "office", "line1": "1 Test Street", "city": "Testville", "region": "CA", "postal_code": "94000", "country": "US"}], "allowed_agents": [agent["agent_id"]]}) check("D", "owner keeps the address in a Commerce Profile (existing mechanism; no vault)", 200, st) c = q.compile_intent("Order a USB-C cable under $20 and buy it.") env = c["envelope"]["envelope_id"] r = q.authorize(envelope_id=env, merchant="Amazon", amount_minor=1500, purpose="cable", commitment="ONE_TIME", fulfillment="physical", disclosures=["shipping_address"]) check("D", "before the owner sets a posture: shipping_address", "ASK_OWNER", r["disclosures"][0]["decision"]) dims = c["envelope"]["dimensions"] dims["data"]["fields"] = {"shipping_address": "RELEASE_ONCE"} st, am = owner.amend_envelope(env, dims, "the seller may have the address once, to ship the cable") check("D", "owner: shipping_address = RELEASE_ONCE (envelope version 2)", (200, 2), (st, am.get("current_version"))) auth = q.authorize_binding(envelope_id=env, merchant="Amazon", amount_minor=1500, purpose="cable", commitment="ONE_TIME", fulfillment="physical", disclosures=["shipping_address", "card_number"], execution="external") disc = {d["scope"]: d for d in auth["disclosures"]} check("D", "physical purchase at Amazon: shipping_address", ("RELEASE_ONCE", "amazon", True), (disc["shipping_address"]["decision"], disc["shipping_address"]["recipient"], disc["shipping_address"]["single_use"])) check("D", "card_number is never disclosable", "DENY", disc["card_number"]["decision"]) check("D", "no value appears in the authorize response", False, "Test Street" in json.dumps(auth)) r2 = q.authorize(envelope_id=env, merchant="AnalyticsCo", amount_minor=1500, purpose="usage analytics", commitment="ONE_TIME", fulfillment="digital", disclosures=["shipping_address"]) check("D", "same address requested by an unrelated analytics service", ("DENY", "not_necessary_for_purpose"), (r2["disclosures"][0]["decision"], r2["disclosures"][0]["reason"])) try: q.release_disclosure(auth.authorization_id, "shipping_address", recipient="AnalyticsCo") check("D", "release to the WRONG recipient", "403 disclosure_recipient_mismatch", "released (!)") except QuattrinoError as e: check("D", "release to the WRONG recipient", "403 disclosure_recipient_mismatch", "%s %s" % (e.status, e.code)) rel = q.release_disclosure(auth.authorization_id, "shipping_address", recipient="Amazon") check("D", "release to Amazon (the fulfilling merchant) -> the value, once", "Testville", (rel.get("value") or {}).get("city")) try: q.release_disclosure(auth.authorization_id, "shipping_address", recipient="Amazon") check("D", "second release (replay)", "409 disclosure_already_released", "released again (!)") except QuattrinoError as e: check("D", "second release (replay)", "409 disclosure_already_released", "%s %s" % (e.status, e.code)) def journey_E(owner): title("E", "CONSERVED MULTI-AGENT DELEGATION ($300 root -> B $175, C $40, D $60; $25 left; $50 refused)") root_agent, qR = owner.make_agent("J-E project lead " + RUN, standing_cents=50000) kids = [owner.make_agent("J-E worker %s %s" % (n, RUN), standing_cents=50000, wallet=False) for n in ("B", "C", "D", "X")] st, g = owner.grant_root(root_agent["agent_id"], 30000, "logo project") root = g.get("authority", g) check("E", "owner grants $300 root authority to the project lead (hard / conserved)", (200, "hard"), (st, root.get("allocation"))) issued = [] for (kid, _), cents in zip(kids[:3], (17500, 4000, 6000)): st, d = owner.delegate(root["authority_id"], kid["agent_id"], cents) issued.append((kid["name"][-8:-7], cents, st)) check("E", "issue $175 / $40 / $60 to B / C / D", [200, 200, 200], [s for _, _, s in issued]) st, view = owner.call("GET", "/delegations/" + root["authority_id"]) a = view.get("authority", view) check("E", "root: delegated / delegatable / spend-remaining", ("$275.00", "$25.00", "$300.00"), (money(a.get("amount_delegated_minor")), money(a.get("amount_delegatable_minor")), money(a.get("amount_remaining_minor")))) st, d = owner.delegate(root["authority_id"], kids[3][0]["agent_id"], 5000) err = (d.get("error") or {}) check("E", "another $50 child", ("422", "delegation_exceeds_parent", "$25.00"), (str(st), err.get("code"), money((err.get("details") or {}).get("parent_delegatable_minor")))) print("\n refusal message:", err.get("message")) held = kids[0][1].discover() # worker B sees only its own authority st, mine = owner.call("GET", "/delegations?agent_id=" + kids[0][0]["agent_id"]) b_auth = next((x for x in mine.get("authorities", []) if x.get("parent_authority_id") == root["authority_id"]), {}) check("E", "worker B holds exactly $175 (child <= parent, lineage depth 1)", ("$175.00", 1), (money(b_auth.get("amount_ceiling_minor")), b_auth.get("depth"))) check("E", "worker B's discovery view never exposes policy internals", True, "policy_id" not in json.dumps(held) and "risk" not in json.dumps(held).lower()) def journey_F(owner): title("F", "CONVERSATIONAL AUTHORITY AMENDMENT (laptop <= $1,000 -> <= $1,200 only if RAM >= 32GB -> no refurbished)") agent, q = owner.make_agent("J-F laptop " + RUN, standing_cents=200000) c = q.compile_intent("Find me a laptop under $1,000 and buy it.") env = c["envelope"]["envelope_id"] d1 = c["envelope"]["dimensions"] check("F", "v1: 'under $1,000' -> $1,000 per-action ceiling", 100000, d1["money"]["per_action_ceiling_minor"]) d2 = {**d1, "money": {**d1["money"], "conditional_ceilings": [{"ceiling_minor": 120000, "when": {"attribute": "ram_gb", "op": ">=", "value": 32}}]}} st, v2 = owner.amend_envelope(env, d2, "Go to $1,200 only if it has 32GB RAM") check("F", "v2: owner widens conditionally (provenance owner_confirmation)", (200, 2, "owner_confirmation"), (st, v2.get("current_version"), ((v2.get("versions") or [{}])[-1].get("provenance") or {}).get("source"))) d3 = {**v2["dimensions"], "counterparty": {**v2["dimensions"]["counterparty"], "attribute_prohibitions": [{"attribute": "refurbished", "op": "==", "value": True}]}} st, v3 = owner.amend_envelope(env, d3, "No refurbished models") check("F", "v3: owner narrows (no refurbished); prev_version chain 2 -> 3", (200, 3, 2), (st, v3.get("current_version"), (v3.get("versions") or [{}])[-1].get("prev_version"))) r = q.authorize(envelope_id=env, merchant="Dell", amount_minor=115000, purpose="laptop", commitment="ONE_TIME", attributes={"ram_gb": 32, "refurbished": False}) check("F", "$1,150 / 32GB / new", ("AUTHORIZED", "$1,200.00"), (r.decision, money(r.effective_maximum_minor))) ok_auth = r r = q.authorize(envelope_id=env, merchant="Dell", amount_minor=115000, purpose="laptop", commitment="ONE_TIME", attributes={"ram_gb": 16, "refurbished": False}) check("F", "$1,150 / 16GB", ("DENIED", "$1,000.00"), (r.decision, money(r.effective_maximum_minor))) r = q.authorize(envelope_id=env, merchant="Dell", amount_minor=90000, purpose="laptop", commitment="ONE_TIME", attributes={"ram_gb": 32, "refurbished": True}) check("F", "$900 / refurbished", ("DENIED", True), (r.decision, "attribute_prohibited" in codes(r, "denial_reasons"))) ex = q.explain(ok_auth.authorization_id) task = (ex.get("provenance") or {}).get("TASK_AUTHORITY") or {} envp = task.get("envelope") or {} check("F", "explain: authorization pinned to envelope version 3 with the amendment history", (3, 3), (envp.get("pinned_version") or envp.get("version") or envp.get("current_version"), len(envp.get("amendments") or []))) print("\n amendments:", json.dumps([{k: a_.get(k) for k in ("version", "source", "widened", "narrowed")} for a_ in envp.get("amendments") or []])[:700]) print(" envelope in words:", envp.get("plain_language")) def journey_G(owner): title("G", "EXTERNAL EXECUTION + OUTCOME COMPLIANCE (Quattrino authorizes; YOU execute; record the outcome)") agent, q = owner.make_agent("J-G external " + RUN, standing_cents=5000) auth = q.authorize_binding(amount_minor=1999, merchant="Amazon", purpose="USB-C cable", commitment="ONE_TIME", task={"max_amount_minor": 2000}, execution="external") check("G", "binding authorization, external execution (no Quattrino payment)", ("AUTHORIZED", "external", True), (auth.decision, auth.get("execution_mode"), bool(auth.get("material_terms_hash") and auth.get("expires_at")))) print(" authorized terms:", json.dumps({k: auth.get(k) for k in ("amount_minor", "merchant", "commitment", "effective_maximum_minor", "expires_at")})) # ... you buy the cable with your own provider ... out = q.record_outcome(auth.authorization_id, success=True, provider="amazon", provider_reference="114-2233", actual_amount_minor=1999, merchant="Amazon", commitment="ONE_TIME", disclosed_fields=[], delegations_performed=[]) check("G", "outcome matches: status / compliance", ("CONSUMED", "COMPLIANT"), (out.get("status"), out["compliance"]["status"])) check("G", "evidence hash present; authorization immutable", True, bool(out.get("evidence_hash")) and out.get("amount_minor") == 1999) ex = q.explain(auth.authorization_id) print("\n what happened:", json.dumps(ex.get("what_happened"))[:300]) print(" plain language:", ex.get("plain_language", "")[-220:]) # the deviating executor auth2 = q.authorize_binding(amount_minor=1999, merchant="Amazon", purpose="USB-C cable", commitment="ONE_TIME", task={"max_amount_minor": 2000}, execution="external") out2 = q.record_outcome(auth2.authorization_id, success=True, provider="amazon", provider_reference="114-2299", actual_amount_minor=2599, merchant="BestBuy", commitment="ONE_TIME") check("G", "executor paid $25.99 at BestBuy under a $19.99 Amazon authorization", ("NONCOMPLIANT", True, True), (out2["compliance"]["status"], "AMOUNT_EXCEEDED" in out2["compliance"]["findings"], "COUNTERPARTY_MISMATCH" in out2["compliance"]["findings"])) try: q.record_outcome(auth2.authorization_id, success=True, provider="amazon", actual_amount_minor=1999) check("G", "recording an outcome twice", "409 (single consumption)", "accepted (!)") except QuattrinoError as e: check("G", "recording an outcome twice", "409 (single consumption)", "%s %s" % (e.status, e.code), ok=(e.status == 409)) JOURNEYS = {"A": journey_A, "B": journey_B, "C": journey_C, "D": journey_D, "E": journey_E, "F": journey_F, "G": journey_G} def main(): wanted = [a.upper() for a in sys.argv[1:] if a.upper() in JOURNEYS] or list(JOURNEYS) email = OWNER_EMAIL or input("owner email: ").strip() password = OWNER_PASSWORD or getpass.getpass("owner password (never stored): ") print("runtime:", BASE, "| owner:", email, "| TEST/SANDBOX only") owner = Owner(email, password) t0 = time.time() for letter in wanted: try: JOURNEYS[letter](owner) except Exception as exc: # noqa: BLE001 check(letter, "journey crashed", "no exception", "%s: %s" % (type(exc).__name__, str(exc)[:300]), ok=False) passed = sum(1 for _, _, ok in RESULTS if ok) print("\n" + "=" * 78) print(" DIFFERENTIATION JOURNEYS: %d/%d checks passed in %.1fs" % (passed, len(RESULTS), time.time() - t0)) for letter in wanted: rows = [ok for j, _, ok in RESULTS if j == letter] print(" %s %s (%d/%d)" % (letter, "PASS" if all(rows) else "FAIL", sum(rows), len(rows))) for j, what, ok in RESULTS: if not ok: print(" FAIL", j, "|", what) sys.exit(0 if passed == len(RESULTS) else 1) if __name__ == "__main__": main()