#!/usr/bin/env python3 """LockedIn Labs: turn a model response into an application decision. Python 3.11+ standard library. Synthetic, selected response excerpts only. No cloud calls, credentials, tool execution, package installs or file writes. Companion: https://training.lockedinlabs.ai/training/guides/cloud-responses """ import argparse from copy import deepcopy import json import sys import unittest PROVIDERS = ("gcp", "aws", "azure") DECISIONS = ("validate", "incomplete", "authorize", "investigate") MAX_INPUT = 8192 DRAFT = "Ticket T-104 is waiting for a review." PARTIAL = "Ticket T-104 is waiting for" NEXT = { "validate": "Check the draft against authorized evidence and the output contract.", "incomplete": "Hold the unfinished output; choose a bounded recovery policy.", "authorize": "Validate and authorize the proposed function before any execution.", "investigate": "Hold the result and inspect the unsupported or invalid contract.", } class Unsupported(ValueError): """The small teaching contract does not explain this input.""" def require(condition, reason): if not condition: raise Unsupported(reason) def shape(value, keys, path): # Exact excerpt shapes are intentional. Real responses include more fields. require(isinstance(value, dict) and set(value) == set(keys), f"{path}: expected exactly the documented teaching fields.") def one(value, path): require(isinstance(value, list) and len(value) == 1, f"{path}: this exercise supports one item; mixed, empty or multiple items need an explicit policy.") return value[0] def nonempty(value, path): require(isinstance(value, str) and bool(value.strip()), f"{path}: expected a nonempty string.") def strict_json(raw): def object_pairs(pairs): result = {} for key, value in pairs: require(key not in result, "JSON contains duplicate keys; do not choose a value silently.") result[key] = value return result def constant(_): raise Unsupported("JSON contains a non-standard numeric constant.") require(isinstance(raw, str) and len(raw) <= MAX_INPUT, "Input exceeds the small lab's character limit.") try: return json.loads(raw, object_pairs_hook=object_pairs, parse_constant=constant) except (json.JSONDecodeError, RecursionError, ValueError) as error: if isinstance(error, Unsupported): raise raise Unsupported("JSON could not be parsed; no repair or execution was attempted.") from None def tool(name, arguments): # This checks only the fictional function's shape, never caller permission. require(name == "get_ticket", "The proposed function is outside this exercise's contract.") shape(arguments, ("ticket_id",), "function arguments") nonempty(arguments["ticket_id"], "ticket_id") return "tool" def content_kind(provider, item): require(isinstance(item, dict), "Content item is not an object.") if provider == "gcp": if set(item) == {"text"}: nonempty(item["text"], "text") return "text" shape(item, ("functionCall",), "part") call = item["functionCall"] shape(call, ("name", "args"), "functionCall") return tool(call["name"], call["args"]) if provider == "aws": if set(item) == {"text"}: nonempty(item["text"], "text") return "text" shape(item, ("toolUse",), "content") call = item["toolUse"] shape(call, ("toolUseId", "name", "input"), "toolUse") nonempty(call["toolUseId"], "toolUseId") return tool(call["name"], call["input"]) if item.get("type") == "message": shape(item, ("type", "role", "content"), "message") require(item["role"] == "assistant", "Expected an assistant message.") part = one(item["content"], "message.content") shape(part, ("type", "text"), "output_text") require(part["type"] == "output_text", "This exercise expects output_text.") nonempty(part["text"], "text") return "text" shape(item, ("type", "call_id", "name", "arguments"), "function_call") require(item["type"] == "function_call", "Unsupported output type.") nonempty(item["call_id"], "call_id") require(isinstance(item["arguments"], str), "Responses function arguments must be a JSON string.") return tool(item["name"], strict_json(item["arguments"])) def classify(provider, raw): """An adapter translates known vendor fields into our four teaching decisions. 'authorize' means authorization is required, NOT that access is allowed. Do not use this exact-excerpt adapter as a production API parser. """ try: require(provider in PROVIDERS, "Choose gcp, aws or azure.") packet = strict_json(raw) if provider == "gcp": shape(packet, ("candidates",), "response") candidate = one(packet["candidates"], "candidates") shape(candidate, ("content", "finishReason"), "candidate") content = candidate["content"] shape(content, ("role", "parts"), "content") require(content["role"] == "model", "Expected model content.") kind = content_kind(provider, one(content["parts"], "parts")) signal = candidate["finishReason"] field, normal, limited, proposed = "finishReason", "STOP", "MAX_TOKENS", "STOP" elif provider == "aws": shape(packet, ("output", "stopReason"), "response") shape(packet["output"], ("message",), "output") message = packet["output"]["message"] shape(message, ("role", "content"), "message") require(message["role"] == "assistant", "Expected an assistant message.") kind = content_kind(provider, one(message["content"], "content")) signal = packet["stopReason"] field, normal, limited, proposed = "stopReason", "end_turn", "max_tokens", "tool_use" else: require(isinstance(packet, dict), "Expected a response object.") incomplete = packet.get("status") == "incomplete" shape(packet, ("status", "output", "incomplete_details") if incomplete else ("status", "output"), "response") kind = content_kind(provider, one(packet["output"], "output")) signal = packet["status"] field, normal, limited, proposed = "status", "completed", "incomplete", "completed" if incomplete: shape(packet["incomplete_details"], ("reason",), "incomplete_details") require(packet["incomplete_details"]["reason"] == "max_output_tokens", "This exercise only handles incomplete max_output_tokens responses.") require(isinstance(signal, str), "Completion state must be a string.") if kind == "text" and signal == limited: decision = "incomplete" elif kind == "text" and signal == normal: decision = "validate" elif kind == "tool" and signal == proposed: decision = "authorize" else: raise Unsupported("The content and completion state do not match a supported pair.") return {"decision": decision, "evidence": f"{field}={signal}; content={kind}", "next": NEXT[decision]} except Unsupported as error: return {"decision": "investigate", "evidence": str(error), "next": NEXT["investigate"]} def specimens(provider): """Original synthetic fragments. Edit a copy to test your own counterexample.""" if provider == "gcp": def wrap(part, finish="STOP"): return {"candidates": [{"content": {"role": "model", "parts": [part]}, "finishReason": finish}]} text = wrap({"text": DRAFT}) limit = wrap({"text": PARTIAL}, "MAX_TOKENS") proposed = wrap({"functionCall": {"name": "get_ticket", "args": {"ticket_id": "T-104"}}}) mixed = deepcopy(text) mixed["candidates"][0]["content"]["parts"].append(deepcopy(proposed["candidates"][0]["content"]["parts"][0])) bad = wrap({"functionCall": {"name": "get_ticket", "args": {"ticket_id": 104}}}) unknown_stop = wrap({"text": DRAFT}, "UNKNOWN_FINISH") elif provider == "aws": def wrap(part, finish="end_turn"): return {"output": {"message": {"role": "assistant", "content": [part]}}, "stopReason": finish} text = wrap({"text": DRAFT}) limit = wrap({"text": PARTIAL}, "max_tokens") proposed = wrap({"toolUse": {"toolUseId": "synthetic-call-1", "name": "get_ticket", "input": {"ticket_id": "T-104"}}}, "tool_use") mixed = deepcopy(proposed) mixed["output"]["message"]["content"].insert(0, {"text": DRAFT}) bad = deepcopy(proposed) bad["output"]["message"]["content"][0]["toolUse"]["input"]["ticket_id"] = 104 unknown_stop = wrap({"text": DRAFT}, "UNKNOWN_FINISH") else: def wrap(value, status="completed"): packet = {"status": status, "output": [{"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": value}]}]} if status == "incomplete": packet["incomplete_details"] = {"reason": "max_output_tokens"} return packet text = wrap(DRAFT) limit = wrap(PARTIAL, "incomplete") proposed = {"status": "completed", "output": [{"type": "function_call", "call_id": "synthetic-call-1", "name": "get_ticket", "arguments": '{"ticket_id":"T-104"}'}]} mixed = deepcopy(text) mixed["output"].extend(deepcopy(proposed["output"])) bad = deepcopy(proposed) bad["output"][0]["arguments"] = '{"ticket_id":104}' unknown_stop = wrap(DRAFT, "UNKNOWN_FINISH") packets = {"text": (text, "validate"), "limit": (limit, "incomplete"), "tool": (proposed, "authorize"), "unknown": ({}, "investigate"), "mixed": (mixed, "investigate"), "bad-arguments": (bad, "investigate"), "unknown-stop": (unknown_stop, "investigate")} result = {name: (json.dumps(packet, indent=2), expected) for name, (packet, expected) in packets.items()} result["malformed"] = ('{"output":', "investigate") return result class AdapterChecks(unittest.TestCase): def test_provider_cases(self): for provider in PROVIDERS: for name, (raw, expected) in specimens(provider).items(): with self.subTest(provider=provider, case=name): self.assertEqual(classify(provider, raw)["decision"], expected) def test_complete_looking_text_is_still_incomplete(self): for provider in PROVIDERS: raw = specimens(provider)["limit"][0].replace(PARTIAL, DRAFT) self.assertEqual(classify(provider, raw)["decision"], "incomplete") def test_invalid_or_ambiguous_json_stays_visible(self): for provider in PROVIDERS: for raw in ("null", "[]", '"text"', '{"x":1,"x":2}', '{"x":NaN}', "{" * (MAX_INPUT + 1)): with self.subTest(provider=provider, raw=raw[:20]): self.assertEqual(classify(provider, raw)["decision"], "investigate") def test_valid_json_arguments_are_required(self): packet = json.loads(specimens("azure")["tool"][0]) packet["output"][0]["arguments"] = '{"ticket_id":' self.assertEqual(classify("azure", json.dumps(packet))["decision"], "investigate") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--provider", choices=PROVIDERS, default="gcp") source = parser.add_mutually_exclusive_group() source.add_argument("--case", choices=tuple(specimens("gcp")), help="Inspect one supplied response (default: text)") source.add_argument("--stdin", action="store_true", help="Read one synthetic JSON excerpt from standard input") source.add_argument("--check", action="store_true", help="Check the supplied cases and boundary mutations") parser.add_argument("--predict", choices=DECISIONS, help="Compare your prediction with the adapter's decision") parser.add_argument("--json", action="store_true", help="Print the decision and explanation as JSON") args = parser.parse_args() if args.check: result = unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(AdapterChecks)) return 0 if result.wasSuccessful() else 1 raw = sys.stdin.read(MAX_INPUT + 1) if args.stdin else specimens(args.provider)[args.case or "text"][0] report = classify(args.provider, raw) matches = args.predict is None or args.predict == report["decision"] if args.json: print(json.dumps({**report, "prediction_matches": None if args.predict is None else matches}, indent=2)) else: print(f"OFFLINE RESPONSE LAB / {args.provider.upper()} / {args.case or ('custom excerpt' if args.stdin else 'text')}") print("1. Selected synthetic response excerpt:") # Escape terminal controls in student-supplied input. No eval or tool dispatch. print(ascii(raw) if args.stdin else raw) print(f"2. Observed: {report['evidence']}") print(f"3. Application decision: {report['decision']}") print(f"4. Next: {report['next']}") if args.predict: print("Prediction matches." if matches else f"Reconsider: you predicted {args.predict}.") print("No model or function ran. This checks response handling, not cloud access or answer accuracy.") return 0 if matches else 1 if __name__ == "__main__": raise SystemExit(main())