🤖 Build an AI Developer Companion on AWS

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Orbit - four-part developer companion learning lab
Parameters:
Lesson:
Type: Number
Default: 1
AllowedValues: [1, 2, 3, 4]
ModelId:
Type: String
Default: amazon.nova-micro-v1:0
Description: Regional foundation model supporting Converse; inference profiles require IAM changes
Resources:
WebBucket:
Type: AWS::S3::Bucket
Properties:
PublicAccessBlockConfiguration:
BlockPublicAcls: true
BlockPublicPolicy: true
IgnorePublicAcls: true
RestrictPublicBuckets: true
OwnershipControls:
Rules: [{ObjectOwnership: BucketOwnerEnforced}]
BucketEncryption:
ServerSideEncryptionConfiguration:
- ServerSideEncryptionByDefault: {SSEAlgorithm: AES256}
OriginAccess:
Type: AWS::CloudFront::OriginAccessControl
Properties:
OriginAccessControlConfig:
Name: !Sub ${AWS::StackName}-origin
OriginAccessControlOriginType: s3
SigningBehavior: always
SigningProtocol: sigv4
WebHeaders:
Type: AWS::CloudFront::ResponseHeadersPolicy
Properties:
ResponseHeadersPolicyConfig:
Name: !Sub ${AWS::StackName}-headers
SecurityHeadersConfig:
ContentTypeOptions: {Override: true}
FrameOptions: {FrameOption: DENY, Override: true}
ReferrerPolicy: {ReferrerPolicy: no-referrer, Override: true}
StrictTransportSecurity: {AccessControlMaxAgeSec: 31536000, Override: true}
ContentSecurityPolicy:
Override: true
ContentSecurityPolicy: "default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; connect-src 'self' https://*.amazoncognito.com https://*.amazonaws.com; media-src 'self' blob:; object-src 'none'; base-uri 'self'; frame-ancestors 'none'; form-action 'self'"
Web:
Type: AWS::CloudFront::Distribution
Properties:
DistributionConfig:
Enabled: true
DefaultRootObject: index.html
HttpVersion: http2
PriceClass: PriceClass_100
Origins:
- Id: web
DomainName: !GetAtt WebBucket.RegionalDomainName
OriginAccessControlId: !Ref OriginAccess
S3OriginConfig: {OriginAccessIdentity: ''}
DefaultCacheBehavior:
TargetOriginId: web
ViewerProtocolPolicy: redirect-to-https
AllowedMethods: [GET, HEAD]
CachedMethods: [GET, HEAD]
Compress: true
CachePolicyId: 4135ea2d-6df8-44a3-9df3-4b5a84be39ad
ResponseHeadersPolicyId: !Ref WebHeaders
WebPolicy:
Type: AWS::S3::BucketPolicy
Properties:
Bucket: !Ref WebBucket
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal: {Service: cloudfront.amazonaws.com}
Action: s3:GetObject
Resource: !Sub ${WebBucket.Arn}/*
Condition:
StringEquals:
AWS:SourceArn: !Sub arn:${AWS::Partition}:cloudfront::${AWS::AccountId}:distribution/${Web}
Pool:
Type: AWS::Cognito::UserPool
Properties:
AdminCreateUserConfig: {AllowAdminCreateUserOnly: true}
Policies:
PasswordPolicy:
MinimumLength: 12
RequireLowercase: true
RequireUppercase: true
RequireNumbers: true
RequireSymbols: true
UsernameConfiguration: {CaseSensitive: false}
ResourceServer:
Type: AWS::Cognito::UserPoolResourceServer
Properties:
Identifier: orbit
Name: Orbit API
UserPoolId: !Ref Pool
Scopes:
- ScopeName: use
ScopeDescription: Use the Orbit learning lab
Client:
Type: AWS::Cognito::UserPoolClient
DependsOn: ResourceServer
Properties:
UserPoolId: !Ref Pool
GenerateSecret: false
AllowedOAuthFlowsUserPoolClient: true
AllowedOAuthFlows: [code]
AllowedOAuthScopes: [openid, orbit/use]
SupportedIdentityProviders: [COGNITO]
CallbackURLs: [!Sub 'https://${Web.DomainName}/']
LogoutURLs: [!Sub 'https://${Web.DomainName}/']
PreventUserExistenceErrors: ENABLED
AccessTokenValidity: 15
IdTokenValidity: 15
TokenValidityUnits: {AccessToken: minutes, IdToken: minutes}
Domain:
Type: AWS::Cognito::UserPoolDomain
Properties:
UserPoolId: !Ref Pool
Domain: !Sub orbit-${AWS::AccountId}-${Client}
Data:
Type: AWS::DynamoDB::Table
Properties:
BillingMode: PAY_PER_REQUEST
AttributeDefinitions:
- {AttributeName: pk, AttributeType: S}
- {AttributeName: sk, AttributeType: S}
KeySchema:
- {AttributeName: pk, KeyType: HASH}
- {AttributeName: sk, KeyType: RANGE}
TimeToLiveSpecification: {AttributeName: expires, Enabled: true}
Api:
Type: AWS::Serverless::HttpApi
Properties:
CorsConfiguration:
AllowOrigins: [!Sub 'https://${Web.DomainName}']
AllowHeaders: [authorization, content-type]
AllowMethods: [GET, POST, PUT, DELETE, OPTIONS]
DefaultRouteSettings:
ThrottlingBurstLimit: 5
ThrottlingRateLimit: 2
Auth:
DefaultAuthorizer: Cognito
Authorizers:
Cognito:
IdentitySource: $request.header.Authorization
AuthorizationScopes: [orbit/use]
JwtConfiguration:
issuer: !Sub https://cognito-idp.${AWS::Region}.amazonaws.com/${Pool}
audience: [!Ref Client]
App:
Type: AWS::Serverless::Function
Properties:
CodeUri: backend/
Handler: aws_adapter.handler
Runtime: python3.12
Architectures: [arm64]
Timeout: 28
MemorySize: 256
ReservedConcurrentExecutions: 2
Environment:
Variables:
TABLE_NAME: !Ref Data
MODEL_ID: !Ref ModelId
LESSON: !Ref Lesson
Policies:
- Statement:
- Effect: Allow
Action: [dynamodb:GetItem, dynamodb:PutItem, dynamodb:UpdateItem]
Resource: !GetAtt Data.Arn
- Effect: Allow
Action: bedrock:InvokeModel
Resource: !Sub arn:${AWS::Partition}:bedrock:${AWS::Region}::foundation-model/${ModelId}
- Effect: Allow
Action: polly:SynthesizeSpeech
Resource: '*'
Events:
Routes:
Type: HttpApi
Properties:
ApiId: !Ref Api
Path: /{proxy+}
Method: ANY
AppLogs:
Type: AWS::Logs::LogGroup
Properties:
LogGroupName: !Sub /aws/lambda/${App}
RetentionInDays: 7
Outputs:
WebUrl: {Value: !Sub 'https://${Web.DomainName}/'}
WebBucket: {Value: !Ref WebBucket}
DistributionId: {Value: !Ref Web}
ApiUrl: {Value: !Sub 'https://${Api}.execute-api.${AWS::Region}.${AWS::URLSuffix}'}
ClientId: {Value: !Ref Client}
UserPoolId: {Value: !Ref Pool}
AuthDomain: {Value: !Sub 'https://${Domain}.auth.${AWS::Region}.amazoncognito.com'}
Lesson: {Value: !Ref Lesson}

"""Orbit learning lab. Shared domain logic; no AWS imports or credentials here."""
import json
from datetime import datetime, timezone
class Problem(Exception):
def __init__(self, status, message):
self.status, self.message = status, message
super().__init__(message)
DEFAULT = {"goal": "Build small things consistently", "topic": "JavaScript", "level": "beginner", "tone": "encouraging"}
TOPICS = ("JavaScript", "Python", "AWS")
LEVELS = ("beginner", "intermediate", "advanced")
BANK = {
"JavaScript": [
("Keep the unique values", "Write unique(values) without mutating its input. Try [1, 1, 2], [], and repeated strings.", "Compare the array length with a Set. Which values use identity equality?"),
("Retry with a limit", "Write an async retry(fn, attempts) helper. Stop after success; throw the last error after the limit. Test one failure then success, and permanent failure.", "Keep attempts explicit. Never retry an irreversible side effect without an idempotency contract."),
("Limit concurrent work", "Implement mapLimit(items, limit, fn) preserving result order. Test empty input, rejection, and a limit of two.", "Use a fixed number of workers that claim the next index; decide what rejection means for running tasks.")],
"Python": [
("Count words", "Write word_counts(text). Normalize case and whitespace. Test empty input and repeated words. State how punctuation is handled.", "Start with split() and a dictionary; document punctuation rather than guessing."),
("Stream a large file", "Write a generator that yields valid JSON records from a JSONL file. Report invalid line numbers without loading the entire file.", "Iterate the file inside a context manager and keep parsing errors separate from I/O errors."),
("Design a safe cache", "Build a bounded in-memory cache with expiry. Inject a clock. Test expiry boundaries, eviction, and concurrent access assumptions.", "Separate expiry from capacity and document which operations are atomic.")],
"AWS": [
("Draw a request journey", "Sketch browser → API Gateway → Lambda. Mark where authentication happens and where credentials must never be stored.", "Separate user authentication from the execution role used by Lambda."),
("Make a retry safe", "Design an SQS worker that can receive the same request twice. Specify a business key and the atomic storage operation that protects its effect.", "A queue message ID is not always the same as a business operation ID."),
("Threat-model shared memory", "Design per-user DynamoDB memory. Explain trusted identity, conditional updates, deletion races, and how you would test two users.", "A user ID in JSON is untrusted. Start at the validated identity boundary.")]
}
def initial():
return {"profile": dict(DEFAULT), "completed": [], "total": 0, "difficulty": 0, "assignment": None}
def only(body, keys):
if not isinstance(body, dict) or set(body) != set(keys):
raise Problem(400, "Unexpected or missing fields")
def text(value, maximum):
if not isinstance(value, str) or not 1 <= len(value.strip()) <= maximum:
raise Problem(400, f"Text must contain 1–{maximum} characters")
return value.strip()
class Service:
def __init__(self, store, model, voice, lesson=4, today=None):
self.store, self.model, self.voice = store, model, voice
self.lesson = lesson
self.today = today or (lambda: datetime.now(timezone.utc).date().isoformat())
def require(self, stage):
if self.lesson < stage:
raise Problem(403, f"This feature starts in part {stage}")
def call(self, user, method, path, body):
if not user:
raise Problem(401, "Sign in to continue")
if not isinstance(body, dict):
raise Problem(400, "Expected a JSON object")
state, version = self.store.load(user) if self.lesson >= 2 else (initial(), 0)
if (method, path) == ("GET", "/state"):
return {"state": state, "version": version, "lesson": self.lesson}
if (method, path) == ("PUT", "/profile"):
self.require(2)
only(body, ["profile", "version"])
profile = body["profile"]
only(profile, DEFAULT.keys())
profile = dict(profile)
profile["goal"] = text(profile["goal"], 180)
if profile["topic"] not in TOPICS or profile["level"] not in LEVELS or profile["tone"] not in ("encouraging", "direct", "playful"):
raise Problem(400, "Unsupported preference")
if type(body["version"]) is not int or body["version"] != version:
raise Problem(409, "Memory changed. Refresh before saving again.")
state["profile"] = profile
state["difficulty"] = LEVELS.index(profile["level"])
return {"version": self.store.save(user, state, version), "state": state}
if (method, path) == ("DELETE", "/state"):
self.require(2)
only(body, ["version"])
if type(body["version"]) is not int or body["version"] != version:
raise Problem(409, "Memory changed. Refresh before deleting.")
fresh = initial()
return {"version": self.store.save(user, fresh, version), "state": fresh}
if (method, path) == ("POST", "/chat"):
only(body, ["message", "history", "tone"])
message = text(body["message"], 1200)
if body["tone"] not in ("encouraging", "direct", "playful"):
raise Problem(400, "Unsupported tone")
history = body["history"]
if not isinstance(history, list) or len(history) > 8 or len(history) % 2:
raise Problem(400, "Send at most four complete conversation turns")
messages = []
for i, turn in enumerate(history):
only(turn, ["role", "text"])
if turn["role"] != ("user" if i % 2 == 0 else "assistant"):
raise Problem(400, "Invalid conversation order")
messages.append({"role": turn["role"], "content": [{"text": text(turn["text"], 3000)}]})
context = {"preferences": state["profile"] if self.lesson >= 2 else {"tone": body["tone"]}}
if self.lesson >= 3:
context["completedCount"] = state["total"]
context["assignment"] = state["assignment"]
messages.append({"role": "user", "content": [{"text": "Untrusted preference/context data: " + json.dumps(context) + "\nCurrent question: " + message}]})
self.store.consume(user, self.today(), "chat", 40)
reply, usage = self.model(messages)
return {"reply": reply[:3000], "usage": usage}
if (method, path) == ("POST", "/challenge"):
self.require(3)
only(body, [])
day = self.today()
if state["assignment"] and state["assignment"]["day"] == day:
return {"state": state, "version": version}
profile = state["profile"]
level = state["difficulty"]
title, task, hint = BANK[profile["topic"]][level]
state["assignment"] = {"id": f"{day}:{profile['topic']}:{level}", "day": day,
"title": title, "task": task, "hint": hint, "level": LEVELS[level], "done": False}
return {"state": state, "version": self.store.save(user, state, version)}
if (method, path) == ("POST", "/complete"):
self.require(3)
only(body, ["id", "rating"])
if body["rating"] not in ("hard", "right", "easy"):
raise Problem(400, "Choose a difficulty rating")
assignment = state["assignment"]
if not assignment or body["id"] != assignment["id"]:
raise Problem(409, "Load your assigned challenge first")
if assignment["done"]:
return {"state": state, "version": version}
assignment["done"] = True
state["total"] += 1
state["completed"] = (state["completed"] + [{"id": assignment["id"], "day": self.today(), "title": assignment["title"], "rating": body["rating"]}])[-30:]
delta = {"hard": -1, "right": 0, "easy": 1}[body["rating"]]
state["difficulty"] = max(0, min(2, state["difficulty"] + delta))
return {"state": state, "version": self.store.save(user, state, version)}
if (method, path) == ("POST", "/voice"):
self.require(4)
only(body, ["text"])
speech = text(body["text"], 600)
self.store.consume(user, self.today(), "voice", 20)
return self.voice(speech)
raise Problem(404, "Route not found")