APPENDIX H: RESUME PORTFOLIO BUILDING CAPSTONE PROJECTS

MICRO-STEPS Part A: Start the folder. 1. Create a new folder named project5_support_desk and open a terminal inside it. Part B: Build the budget. 2. Create a file named budget.py. 3. Type or paste the following code into budget.py and save it. File: budget.py [START OF PYTHON CODE] # Import the time module to measure elapsed seconds import time # Define an exception class raised whenever a budget limit is exceeded class BudgetExceeded(Exception): pass # Define a class that tracks and enforces limits on tokens, calls, and elapsed time class Budget: # Define the constructor that sets up the limits and starts the clock def __init__(self, max_tokens, max_calls, max_seconds): self.max_tokens = max_tokens self.max_calls = max_calls self.max_seconds = max_seconds self.tokens = 0 self.calls = 0 self.started = time.time() # Define a method that charges a number of tokens against the budget and checks every limit def charge(self, tokens): # Count this as one more call and add its tokens to the running total self.calls += 1 self.tokens += tokens # If too many calls have been made, raise the budget-exceeded error if self.calls > self.max_calls: raise BudgetExceeded("too many calls") # If too many tokens have been used, raise the budget-exceeded error if self.tokens > self.max_tokens: raise BudgetExceeded("too many tokens") # If too much time has passed since the budget started, raise the budget-exceeded error if time.time() - self.started > self.max_seconds: raise BudgetExceeded("too much time") [END OF PYTHON CODE] A Budget is created fresh for each ticket with three limits: tokens, calls, and seconds. Every time a worker does work, it calls charge, and charge raises BudgetExceeded the moment any limit is crossed, like a meter that cuts the power at your limit. 4. Type python -c "import budget as b; x = b.Budget(100, 2, 5); x.charge(10); x.charge(10); x.charge(10)" and press Enter. You should see an error ending in BudgetExceeded: too many calls, because the budget allowed only two calls and you made three. Part C: Build the retry helper. 5. Create a file named retry.py. 6. Type or paste the following code into retry.py and save it. File: retry.py [START OF PYTHON CODE] # Import the time module to pause between retries import time # Define an exception class for errors that are expected to go away if retried class TransientError(Exception): pass # Define a function that retries a piece of work with doubling delays until it succeeds or runs out of attempts def with_retry(work, attempts=3, base_delay=0.5, sleeper=time.sleep, on_retry=None): # Start the delay at the given base delay delay = base_delay # Try the work up to the given number of attempts for attempt in range(1, attempts + 1): try: # Try running the work and return its result if it succeeds return work() except TransientError as problem: # If this was the last allowed attempt, give up and let the error propagate if attempt == attempts: raise # If a retry callback was given, report this attempt before waiting if on_retry is not None: on_retry(attempt, delay, str(problem)) # Wait for the current delay before trying again sleeper(delay) # Double the delay for the next attempt delay = delay + delay [END OF PYTHON CODE] The function with_retry tries some work, and if the work raises a TransientError, which means a temporary problem that may go away, it waits and tries again. The wait doubles each time by adding the delay to itself, so the waits are half a second, then one second, and so on. After the last attempt, it lets the error out so that the supervisor can decide what to do. The parameter named sleeper lets a test replace the real waiting with a fake, so tests run instantly. 7. Type python -c "import retry as r; print(r.with_retry(lambda: 'fine'))" and press Enter. You should see fine. Part D: Build the workers. 8. Create a file named workers.py. 9. Type or paste the following code into workers.py and save it. File: workers.py [START OF PYTHON CODE] # Import the transient-error class to raise when simulating a flaky service from retry import TransientError # Create a dictionary variable called FAILS_LEFT tracking how many times each ticket should still fail FAILS_LEFT = {} # Define a function that resets the simulated failures so ticket T3 will fail exactly once def reset_failures(): FAILS_LEFT.clear() FAILS_LEFT["T3"] = 1 # Define a function that classifies a ticket into a category, charging the budget and simulating failures or loops def classify(ticket, budget): # Charge the budget for this classification call budget.charge(100) # If this ticket still has a scripted failure left, use it up and raise a transient error if FAILS_LEFT.get(ticket["id"], 0) > 0: FAILS_LEFT[ticket["id"]] -= 1 raise TransientError("classifier service timed out") # Read out the ticket's text, lowercased for easy matching text = ticket["text"].lower() # If the ticket's text mentions "loop", keep charging the budget forever to simulate a runaway loop if "loop" in text: while True: budget.charge(300) # If the ticket mentions something broken or damaged, classify it as a damaged item if "broken" in text or "damaged" in text: return "damaged_item" # If the ticket asks where something is or says it is late, classify it as a shipping issue if "where" in text or "late" in text: return "shipping" # If the ticket mentions a refund, classify it as a refund request if "refund" in text: return "refund" # Otherwise, classify it as a general ticket return "general" # Define a function that drafts a reply for a ticket, charging the budget and tailoring the wording to the category def draft_reply(ticket, category, budget, attempt): # Charge the budget for this drafting call budget.charge(200) # Build the base reply text naming the ticket's category in plain words text = "Thank you for contacting us about " + category.replace("_", " ") + ". We will look into it and reply soon." # On the first attempt at a refund reply, add an overly strong guarantee (planted problem for the reviewer to catch) if category == "refund" and attempt == 1: text = text + " We guarantee a full refund." # Return the finished draft text return text # Define a function that reviews a draft reply for banned words before it can be sent def review(draft): # Create a list variable called banned naming words the draft must never contain banned = ["guarantee", "promise"] # Check the draft for each banned word, lowercased for easy matching for word in banned: if word in draft.lower(): return False, "contains the banned word " + word # If no banned words were found, the draft is approved return True, "ok" [END OF PYTHON CODE] The function classify sorts a ticket into a category by looking for key words. It is wired with two planted problems: ticket T3 fails once with a TransientError, and any ticket that contains the word loop sends the worker into an endless loop of charges. The function draft_reply writes a polite reply, but on its first attempt for a refund ticket it adds a promise that the company guarantees a refund. The function review rejects any draft containing the words guarantee or promise. Part E: Build the supervisor. 10. Create a file named supervisor.py. 11. Type or paste the following code into supervisor.py and save it. File: supervisor.py [START OF PYTHON CODE] # Import the os module to work with file paths and environment variables import os # Import the json module to write structured data import json # Import the time module to measure and record when events happen import time # Import the workers module as a whole, to use its functions by name import workers # Import the budget class and its exception from the budget module from budget import Budget, BudgetExceeded # Import the retry helper and its transient-error class from the retry module from retry import with_retry, TransientError # Create a text variable called HERE holding the folder that contains this file HERE = os.path.dirname(os.path.abspath(__file__)) # Create a text variable called AUDIT_PATH pointing to the audit log file AUDIT_PATH = os.path.join(HERE, "audit_log.jsonl") # Create a list variable called TICKETS holding the sample support tickets to process TICKETS = [ {"id": "T1", "text": "My garden hose arrived broken."}, {"id": "T2", "text": "Where is my order? It is late."}, {"id": "T3", "text": "I was charged twice, please help."}, {"id": "T4", "text": "I want a refund for the wrong size."}, {"id": "T5", "text": "Please loop forever on this strange request."}, ] # Define a function that appends one structured event to the audit log file def audit(event, data): record = {"time": round(time.time(), 2), "event": event, "data": data} with open(AUDIT_PATH, "a", encoding="utf-8") as f: f.write(json.dumps(record) + "\n") # Define a function that handles one ticket end to end: classify, draft, review, and redraft if needed def handle_ticket(ticket, sleeper=time.sleep): # Create a fresh budget for this one ticket budget = Budget(max_tokens=2000, max_calls=10, max_seconds=5) # Define a small helper that logs each retry attempt to the audit log def note_retry(attempt, delay, why): audit("retry", {"ticket": ticket["id"], "attempt": attempt, "delay": delay, "why": why}) # Try the full ticket-handling sequence, catching budget and retry failures try: # Classify the ticket, retrying automatically if the classifier raises a transient error category = with_retry(lambda: workers.classify(ticket, budget), sleeper=sleeper, on_retry=note_retry) # Draft a first reply for this category draft = workers.draft_reply(ticket, category, budget, 1) # Review the first draft for banned words ok, why = workers.review(draft) # If the first draft failed review, log it and draft a second attempt, then review that one too if not ok: audit("redraft", {"ticket": ticket["id"], "why": why}) draft = workers.draft_reply(ticket, category, budget, 2) ok, why = workers.review(draft) # If the draft still failed review after the second attempt, escalate the ticket to a human if not ok: return {"id": ticket["id"], "status": "escalated", "why": "review failed twice"} # Otherwise, the ticket is done, so return its category, final draft, and tokens used return {"id": ticket["id"], "status": "done", "category": category, "draft": draft, "tokens": budget.tokens} # If the budget ran out at any point, escalate the ticket with that reason except BudgetExceeded as problem: return {"id": ticket["id"], "status": "escalated", "why": "budget: " + str(problem)} # If all retry attempts were exhausted, escalate the ticket with that reason except TransientError as problem: return {"id": ticket["id"], "status": "escalated", "why": "retries exhausted: " + str(problem)} # Define a function that runs every ticket through the handler and reports overall metrics def run_all(tickets, sleeper=time.sleep): # Reset the scripted classifier failures so each run starts the same way workers.reset_failures() # Create an empty list to hold the outcome of each ticket results = [] # Process every ticket in order for ticket in tickets: # If the kill switch environment variable is set, escalate the ticket without processing it if os.environ.get("HARNESS_KILL_SWITCH") == "1": outcome = {"id": ticket["id"], "status": "escalated", "why": "kill switch is on"} # Otherwise, handle the ticket normally else: outcome = handle_ticket(ticket, sleeper) # Log this ticket's outcome and add it to the results list audit("ticket_outcome", outcome) results.append(outcome) # Build the summary metrics: total tickets, how many finished, and how many were escalated metrics = { "total": len(results), "done": sum(1 for r in results if r["status"] == "done"), "escalated": sum(1 for r in results if r["status"] == "escalated"), } # Log the run's overall metrics audit("run_metrics", metrics) # Return both the individual results and the overall metrics return results, metrics # If this file is run directly (not imported), process every sample ticket and print the outcomes if __name__ == "__main__": results, metrics = run_all(TICKETS) for r in results: print(r["id"], r["status"], r.get("why", r.get("category", ""))) print(metrics) [END OF PYTHON CODE] Read the function handle_ticket from the top. It makes a new budget, classifies with retries, drafts, reviews, redrafts once if the review fails, and returns a result dictionary saying done or escalated and why. Two kinds of failure are caught by name: running out of budget and running out of retries, and both lead to an escalation, which means a human takes over. The function run_all checks the kill switch before every ticket, writes an audit record for each outcome, and returns simple metrics. 12. Type python supervisor.py and press Enter. You should see five lines, one per ticket, ending with a dictionary of totals. You should see T1 through T4 marked done and T5 marked escalated with the reason budget: too many tokens, and the totals should read 5 total, 4 done, and 1 escalated. 13. Open audit_log.jsonl. You should see a retry event for T3 with a delay of 0.5, a redraft event for T4 with a reason mentioning the banned word, and one ticket_outcome event per ticket. Part F: Prove it with tests. 14. Create a file named test_project5.py. 15. Type or paste the following code into test_project5.py and save it. File: test_project5.py [START OF PYTHON CODE] # Import the os module to work with file paths and environment variables import os # Import the json module to read the audit log import json # Import the supervisor module being tested import supervisor # Define a function that deletes the audit log so each test starts clean def reset(): if os.path.exists(supervisor.AUDIT_PATH): os.remove(supervisor.AUDIT_PATH) # Define a function that reads the audit log and returns every record matching a given event kind def events(kind): with open(supervisor.AUDIT_PATH, "r", encoding="utf-8") as f: return [json.loads(line) for line in f if json.loads(line)["event"] == kind] # Define a test that runs the full ticket set and checks retries, redrafts, and escalation all behave correctly def test_normal_day(): reset() # Create a list to record every delay the supervisor would have slept for slept = [] # Run every ticket, capturing sleep delays instead of actually waiting results, metrics = supervisor.run_all(supervisor.TICKETS, sleeper=slept.append) # Build a lookup from ticket id to its final status status = {r["id"]: r["status"] for r in results} # Check that the two straightforward tickets finished normally assert status["T1"] == "done" and status["T2"] == "done" # Check that the ticket with a scripted failure still finished, after one 0.5-second retry delay assert status["T3"] == "done" and slept == [0.5] # Check that the refund ticket finished after exactly one redraft was needed assert status["T4"] == "done" and len(events("redraft")) == 1 # Check that the looping ticket was escalated instead of running forever assert status["T5"] == "escalated" # Check that the overall metrics match: five tickets, four done, one escalated assert metrics == {"total": 5, "done": 4, "escalated": 1} print("PASS retries, redraft, and budget escalation all work") # Define a test that checks the kill switch escalates every ticket without processing any of them def test_kill_switch(): reset() # Turn the kill switch on via the environment variable os.environ["HARNESS_KILL_SWITCH"] = "1" # Make sure the kill switch is turned back off no matter how the run ends try: results, metrics = supervisor.run_all(supervisor.TICKETS, sleeper=lambda s: None) finally: del os.environ["HARNESS_KILL_SWITCH"] # Check that nothing finished and every ticket was escalated assert metrics["done"] == 0 and metrics["escalated"] == 5 print("PASS kill switch stops all work") # If this file is run directly (not imported), run both tests in order if __name__ == "__main__": test_normal_day() test_kill_switch() print("ALL TESTS PASSED") [END OF PYTHON CODE] 16. Type python test_project5.py and press Enter. You should see PASS retries, redraft, and budget escalation all work, then PASS kill switch stops all work, and then ALL TESTS PASSED. 17. In supervisor.py, find the line that creates the Budget and change max_tokens to 1000000000000 and max_calls to 1000000000000, which is a one followed by twelve zeros, then run python supervisor.py. You should see the program pause for about five seconds and then mark T5 as escalated with the reason budget: too much time, which shows that the time limit is the last line of defense when the other limits are set too loosely. 18. Change both numbers back to 2000 and 10 and rerun the tests. Part G: Try the kill switch by hand. 19. In PowerShell, type $env:HARNESS_KILL_SWITCH = "1" and press Enter, or in the Windows command prompt type set HARNESS_KILL_SWITCH=1. 20. Type python supervisor.py and press Enter. You should see all five tickets escalated with the reason kill switch is on. 21. Turn the switch off with Remove-Item Env:HARNESS_KILL_SWITCH in PowerShell, or set HARNESS_KILL_SWITCH= in the command prompt, and run python supervisor.py again to confirm normal behavior returns. Part H: Write the incident runbook. 22. Create a text file named runbook.txt. 23. Write a heading called Purpose and one sentence saying who uses the runbook and when. 24. Write a heading called How To Stop Everything and list the exact steps for the kill switch from step 19, including how to confirm it worked by reading the audit log. 25. Write a heading called Severity Levels with three levels in your own words, for example Level 1 for wrong or harmful replies reaching customers, Level 2 for a spike in escalations, and Level 3 for slowness only. 26. Write a heading called First Five Minutes with five numbered actions, such as: stop new work, save the audit log, note the time, tell the team lead, and decide on the severity level. 27. Write a heading called Common Symptoms with at least four entries, each stating the symptom, the likely cause, and the first thing to check, for example "Many tickets escalated with budget reasons; likely a loop or a slow model; check the average tokens per ticket in the audit log." 28. Write a heading called Recovery that explains how to turn work back on gradually, starting with a few tickets and watching the escalation count. 29. Write a heading called After The Incident that asks for a short written review without blame, a list of what to change, and a new test that would have caught the problem. 30. Run a practice drill: pretend the classifier is failing for every ticket, follow your runbook line by line, and note any step that was unclear. 31. Fix each unclear step in the runbook. Part I: Write the deployment plan. 32. Create a text file named deployment_plan.txt. 33. Write a heading called Environments and list a development environment, a staging environment that mirrors production, and a production environment, saying what each is for. 34. Write a heading called Configuration and Secrets stating that settings such as limits come from environment variables or a configuration file, and that secret keys are kept in a secrets manager or the platform's protected settings and never in the code. 35. Write a heading called Release Steps with numbered steps: run all tests, run the evaluation suite from Project 4 and require the quality gate to pass, deploy to staging, run a small batch of invented tickets, then release to a small share of real traffic, which is called a canary release, before releasing to everyone. 36. Write a heading called Rollback describing how to return to the previous version quickly and who decides. 37. Write a heading called Monitoring listing what you would watch, such as the escalation rate, tokens per ticket, retry count, average duration, and the count of reviewer rejections, with an alert rule for each. 38. Write a heading called Risks and Limits and state plainly that this design was tested only with scripted workers and invented tickets.