#!/usr/bin/env python3 """Apply five Embedded Semantics recipes with only the Python standard library. This example never performs the downstream side effect. It returns an explicit plan showing the semantic outcome and the separate authorization result. """ from __future__ import annotations import argparse import json import sys from typing import Any from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen DEFAULT_URL = "https://embeddedsemantics.com/api/v1/resolve" RECIPE_IDS = ( "multilingual-memory", "workflow-routing", "api-event-interoperability", "clarification", "audit-provenance", ) RECIPE_ACTIONS: dict[str, dict[str, str]] = { "multilingual-memory": { "resolved": "prepare a memory record with exact governed identity metadata", "unknown_expression": "prepare an unresolved memory record with no ConceptCode", "ambiguous_expression": "request context or preserve an ambiguous memory record with no ConceptCode", "request_or_service_error": "preserve the input and error evidence with no ConceptCode", }, "workflow-routing": { "resolved": "offer the exact ConceptCode to the separately authorized routing policy", "unknown_expression": "use a documented non-semantic fallback or human review path", "ambiguous_expression": "request the smallest distinguishing clarification before routing", "request_or_service_error": "block semantic routing or use a documented safe default", }, "api-event-interoperability": { "resolved": "attach exact code and registry version as semantic metadata", "unknown_expression": "emit explicit unknown semantic metadata with no ConceptCode", "ambiguous_expression": "emit explicit ambiguous semantic metadata with no ConceptCode", "request_or_service_error": "emit or log explicit resolver-error metadata with no ConceptCode", }, "clarification": { "resolved": "explain the governed identity and continue under a separate action policy", "unknown_expression": "explain that no reviewed identity was found and continue without a code", "ambiguous_expression": "ask one targeted question for the missing distinguishing context", "request_or_service_error": "explain that no trustworthy semantic result is available", }, "audit-provenance": { "resolved": "retain the complete envelope, exact code, registry version, and separate authorization evidence", "unknown_expression": "retain explicit unknown evidence with no ConceptCode", "ambiguous_expression": "retain explicit ambiguity evidence with no ConceptCode", "request_or_service_error": "retain the request or service failure as a separate evidence event", }, } def call_resolver(expression: str, language: str | None, url: str = DEFAULT_URL) -> dict[str, Any]: """Send one minimum-data resolver request and decode its JSON envelope.""" payload: dict[str, str] = {"expression": expression} if language: payload["language"] = language request = Request( url, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), method="POST", headers={ "Accept": "application/json", "Content-Type": "application/json; charset=utf-8", "User-Agent": "EmbeddedSemantics-UseCaseRecipes/1.0", }, ) try: with urlopen(request, timeout=15) as response: body = response.read().decode("utf-8") value = json.loads(body) if not isinstance(value, dict): raise RuntimeError("resolver returned a non-object JSON value") return value except HTTPError as error: body = error.read().decode("utf-8", errors="replace") try: detail: object = json.loads(body) except json.JSONDecodeError: detail = {"rawBody": body} return { "data": None, "error": { "code": "http_error", "message": f"resolver HTTP {error.code}", "detail": detail, }, } except (URLError, TimeoutError, json.JSONDecodeError) as error: return { "data": None, "error": { "code": "service_error", "message": str(error), }, } def build_semantic_record( expression: str, language: str | None, envelope: dict[str, Any], ) -> dict[str, Any]: """Normalize one envelope into the fail-closed recipe record shape.""" record: dict[str, Any] = { "originalExpression": expression, "language": language, "semanticOutcome": "request_or_service_error", "resolverEvidence": envelope, } if envelope.get("error") is not None: record["error"] = envelope.get("error") return record data = envelope.get("data") if not isinstance(data, dict): record["error"] = {"code": "invalid_envelope", "message": "missing data object"} return record status = data.get("status") if status == "resolved": concept = data.get("concept") code = concept.get("code") if isinstance(concept, dict) else None version = concept.get("registryVersion") if isinstance(concept, dict) else None if not isinstance(code, str) or not code or not isinstance(version, str) or not version: record["error"] = { "code": "invalid_resolved_contract", "message": "resolved response requires a non-empty code and registryVersion", } return record record.update( { "semanticOutcome": "resolved", "conceptCode": code, "registryVersion": version, } ) return record if status == "abstained": reason = data.get("reason") if reason == "unknown_expression": record["semanticOutcome"] = "unknown_expression" elif reason == "ambiguous_expression": record["semanticOutcome"] = "ambiguous_expression" else: record["semanticOutcome"] = "other_abstention" record["abstentionReason"] = reason return record record["error"] = {"code": "unsupported_status", "message": f"unsupported status: {status!r}"} return record def apply_recipe( recipe_id: str, semantic_record: dict[str, Any], *, authorization_allowed: bool, authorization_policy_version: str, ) -> dict[str, Any]: """Return a side-effect-free plan for one recipe and separate policy result.""" if recipe_id not in RECIPE_IDS: raise ValueError(f"unsupported recipe: {recipe_id}") outcome = str(semantic_record.get("semanticOutcome", "request_or_service_error")) action_key = outcome if outcome in RECIPE_ACTIONS[recipe_id] else "request_or_service_error" authorization_outcome = "allowed" if authorization_allowed else "denied" return { "recipe": recipe_id, "semanticRecord": semantic_record, "semanticPlan": RECIPE_ACTIONS[recipe_id][action_key], "authorization": { "outcome": authorization_outcome, "policyVersion": authorization_policy_version, "evaluatedSeparately": True, }, "sideEffectAllowed": authorization_allowed, "sideEffectPerformed": False, "rule": ( "This example prepares a decision plan only. The application must implement the actual side effect " "and may never infer authorization from a ConceptCode." ), } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("recipe", choices=RECIPE_IDS) parser.add_argument("expression") parser.add_argument("--language", help="Optional known BCP 47 language tag.") parser.add_argument("--url", default=DEFAULT_URL) parser.add_argument( "--authorized", action="store_true", help="Simulate an independent policy allowing the downstream side effect.", ) parser.add_argument("--policy-version", default="example-policy-v1") args = parser.parse_args() envelope = call_resolver(args.expression, args.language, args.url) record = build_semantic_record(args.expression, args.language, envelope) plan = apply_recipe( args.recipe, record, authorization_allowed=args.authorized, authorization_policy_version=args.policy_version, ) print(json.dumps(plan, ensure_ascii=False, indent=2, sort_keys=True)) return 0 if record["semanticOutcome"] == "resolved" else 3 if __name__ == "__main__": raise SystemExit(main())