engineering/opencompletion.com
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #14 from russellballestrini/user-experience-day-1

Browse source 
User experience day 1
This commit is contained in:
Russell 2025-08-11 17:21:50 -04:00 • committed by GitHub
commit 77203e4bda
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
18 changed files with 3394 additions and 144 deletions
Download patch file Download diff file Expand all files Collapse all files

1
CLAUDE.md
View file

@ -2,6 +2,7 @@
## Commit Messages
- NEVER add Claude attributions like "🤖 Generated with Claude Code" to commit messages
- NEVER add "Co-Authored-By: Claude <noreply@anthropic.com>" to commit messages
- Keep commit messages focused on the actual changes and their purpose
- Use conventional commit format when appropriate
- Be concise but descriptive about what was changed and why

13
Makefile
View file

@ -201,6 +201,19 @@ clean:
find . -name "*.pyc" -delete 2>/dev/null || true
find . -name "*.pyo" -delete 2>/dev/null || true
find . -name "*~" -delete 2>/dev/null || true
.PHONY: init-db
init-db:
@echo "🗄️ Initializing database tables..."
@if [ -f vars.sh ]; then \
. ./vars.sh && python init_db.py; \
echo "✅ Database tables created successfully"; \
else \
echo "❌ Error: vars.sh not found. Please create it from vars.sh.sample"; \
exit 1; \
fi
clean-cache:
rm -rf .pytest_cache/ 2>/dev/null || true
rm -rf htmlcov/ 2>/dev/null || true
rm -rf .coverage 2>/dev/null || true

75
activity_yaml_validator.py
View file

@ -242,6 +242,12 @@ class ActivityYAMLValidator:
f"Section {section_id}, step {step_id}: 'feedback_tokens_for_ai' must be a string"
)
# Validate feedback_prompts (new multi-prompt system)
if "feedback_prompts" in step:
self._validate_feedback_prompts(
step["feedback_prompts"], section_id, step_id
)
# Validate buckets and transitions
if "buckets" in step:
self._validate_buckets(step["buckets"], section_id, step_id)
@ -251,6 +257,75 @@ class ActivityYAMLValidator:
step["transitions"], step.get("buckets", []), section_id, step_id
)
def _validate_feedback_prompts(
self, feedback_prompts: List[Dict[str, Any]], section_id: str, step_id: str
):
"""Validate feedback_prompts structure"""
if not isinstance(feedback_prompts, list):
self.errors.append(
f"Section {section_id}, step {step_id}: 'feedback_prompts' must be a list"
)
return
if len(feedback_prompts) == 0:
self.errors.append(
f"Section {section_id}, step {step_id}: 'feedback_prompts' cannot be empty"
)
return
prompt_names = set()
for i, prompt in enumerate(feedback_prompts):
if not isinstance(prompt, dict):
self.errors.append(
f"Section {section_id}, step {step_id}: feedback_prompts[{i}] must be a dictionary"
)
continue
# Required fields for each prompt
required_fields = ["name", "tokens_for_ai"]
for field in required_fields:
if field not in prompt:
self.errors.append(
f"Section {section_id}, step {step_id}: feedback_prompts[{i}] missing required field '{field}'"
)
# Validate name uniqueness
if "name" in prompt:
if not isinstance(prompt["name"], str):
self.errors.append(
f"Section {section_id}, step {step_id}: feedback_prompts[{i}].name must be a string"
)
else:
if prompt["name"] in prompt_names:
self.errors.append(
f"Section {section_id}, step {step_id}: duplicate feedback prompt name '{prompt['name']}'"
)
prompt_names.add(prompt["name"])
# Validate tokens_for_ai
if "tokens_for_ai" in prompt:
if not isinstance(prompt["tokens_for_ai"], str):
self.errors.append(
f"Section {section_id}, step {step_id}: feedback_prompts[{i}].tokens_for_ai must be a string"
)
# Check for STFU token usage (informational)
elif "STFU" in prompt["tokens_for_ai"]:
# This is valid - STFU token is used to suppress empty feedback messages
pass
# Validate metadata_filter (optional)
if "metadata_filter" in prompt:
if not isinstance(prompt["metadata_filter"], list):
self.errors.append(
f"Section {section_id}, step {step_id}: feedback_prompts[{i}].metadata_filter must be a list"
)
else:
for j, filter_key in enumerate(prompt["metadata_filter"]):
if not isinstance(filter_key, str):
self.errors.append(
f"Section {section_id}, step {step_id}: feedback_prompts[{i}].metadata_filter[{j}] must be a string"
)
def _validate_buckets(self, buckets: List[str], section_id: str, step_id: str):
"""Validate buckets list"""
if not isinstance(buckets, list):

313
app.py
View file

@ -22,6 +22,8 @@ from flask import (
send_from_directory,
jsonify,
Response,
redirect,
url_for,
)
from flask_socketio import SocketIO, emit, join_room, leave_room
@ -31,10 +33,12 @@ from sqlalchemy.exc import InvalidRequestError
from models import db, Room, UserSession, Message, ActivityState
app = Flask(__name__)
app = Flask(__name__, instance_relative_config=True)
app.config["SECRET_KEY"] = os.environ.get("SECRET_KEY", "dev-key-change-in-production")
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///chat.db"
app.config["SQLALCHEMY_DATABASE_URI"] = (
f"sqlite:///{os.path.join(app.instance_path, 'chat.db')}"
)
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
db.init_app(app)
@ -233,6 +237,28 @@ def get_models():
return jsonify({"models": list(MODEL_CLIENT_MAP.keys())})
@app.route("/api/activities", methods=["GET"])
def get_activities():
"""Return the list of available activities."""
activities = []
if app.config.get("LOCAL_ACTIVITIES"):
# List local activity files from research directory
import os
research_dir = "research"
if os.path.exists(research_dir):
for filename in sorted(os.listdir(research_dir)):
if filename.endswith((".yaml", ".yml")):
activities.append(f"research/{filename}")
else:
# For S3 activities, you would list from S3
# This is a placeholder - you'd need to implement S3 listing
pass
return jsonify({"activities": activities})
@app.route("/chat/<room_name>")
def chat(room_name):
# Query all rooms so that newest is first.
@ -341,6 +367,25 @@ def search_page():
# Call the function to search messages
search_results = search_messages(keywords)
# If there's exactly one search result, redirect directly to that room
if len(search_results) == 1:
room_result = search_results[0]
room_name = room_result["room_name"]
# Build the redirect URL with current parameters
redirect_params = {}
if username and username != "guest":
redirect_params["username"] = username
# Preserve other URL parameters like model, voice, etc.
for param in ["model", "voice"]:
value = request.args.get(param)
if value:
redirect_params[param] = value
redirect_url = url_for("chat", room_name=room_name, **redirect_params)
return redirect(redirect_url)
return render_template(
"search.html",
rooms=rooms,
@ -651,6 +696,30 @@ def handle_update_message(data):
)
@socketio.on("get_activity_status")
def handle_get_activity_status(data):
"""Get the current activity status for a room."""
room_name = data["room_name"]
room = get_room(room_name)
if room:
activity_state = ActivityState.query.filter_by(room_id=room.id).first()
if activity_state:
emit(
"activity_status",
{
"active": True,
"activity_name": activity_state.s3_file_path,
"section_id": activity_state.section_id,
"step_id": activity_state.step_id,
},
room=request.sid,
)
else:
emit("activity_status", {"active": False}, room=request.sid)
def group_consecutive_roles(messages):
if not messages:
return []
@ -763,7 +832,10 @@ def chat_claude(
"message_chunk",
{
"id": msg_id,
"content": f"**{username} ({model_name}):**\n\n{content}",
"content": content,
"username": username,
"model_name": model_name,
"is_first_chunk": True,
},
room=room.name,
)
@ -921,7 +993,10 @@ def chat_gpt(username, room_name, model_name="gpt-4o-mini"):
"message_chunk",
{
"id": msg_id,
"content": f"**{username} ({model_name}):**\n\n{content}",
"content": content,
"username": username,
"model_name": model_name,
"is_first_chunk": True,
},
room=room.name,
)
@ -1035,7 +1110,10 @@ def chat_llama(username, room_name, model_name="mistral-7b-instruct-v0.2.Q3_K_L.
"message_chunk",
{
"id": msg_id,
"content": f"**{username} ({model_name}):**\n\n{content}",
"content": content,
"username": username,
"model_name": model_name,
"is_first_chunk": True,
},
room=room.name,
)
@ -1542,12 +1620,14 @@ def loop_through_steps_until_question(
# Check if the current step has a question
if "question" in step:
question_content = f"Question: {step['question']}"
question_content = step["question"]
translated_question_content = translate_text(
question_content, user_language
)
new_message = Message(
username="System", content=translated_question_content, room_id=room.id
username="System (Question)",
content=translated_question_content,
room_id=room.id,
)
db.session.add(new_message)
db.session.commit()
@ -1596,6 +1676,8 @@ def loop_through_steps_until_question(
},
room=room_name,
)
# Return to activity chooser
socketio.emit("activity_status", {"active": False}, room=room_name)
break
@ -1623,6 +1705,18 @@ def start_activity(room_name, s3_file_path, username):
activity_content, activity_state, room_name, username
)
# Emit activity status update
socketio.emit(
"activity_status",
{
"active": True,
"activity_name": s3_file_path,
"section_id": initial_section["section_id"],
"step_id": initial_step["step_id"],
},
room=room_name,
)
def cancel_activity(room_name, username):
with app.app_context():
@ -1656,6 +1750,9 @@ def cancel_activity(room_name, username):
room=room_name,
)
# Emit activity status update
socketio.emit("activity_status", {"active": False}, room=room_name)
def display_activity_metadata(room_name, username):
with app.app_context():
@ -2088,33 +2185,57 @@ def handle_activity_response(room_name, user_response, username):
# if "correct" or max_attempts reached.
# Provide feedback based on the category
# Filter metadata for feedback if metadata_feedback_filter is specified
feedback_metadata = activity_state.dict_metadata
if "metadata_feedback_filter" in transition:
filter_keys = transition["metadata_feedback_filter"]
feedback_metadata = {
k: v
for k, v in activity_state.dict_metadata.items()
if k in filter_keys
}
# Handle feedback systems
feedback_messages = []
feedback = provide_feedback(
transition,
category,
step["question"],
feedback_tokens_for_ai,
user_response,
user_language,
username,
json.dumps(feedback_metadata),
json.dumps(new_metadata),
)
if "feedback_prompts" in step:
# New multi-prompt system - pass full metadata, let each prompt filter
multi_feedback_messages = provide_feedback_prompts(
transition,
category,
step["question"],
step["feedback_prompts"],
user_response,
user_language,
username,
json.dumps(activity_state.dict_metadata), # Pass full metadata
json.dumps(new_metadata),
feedback_tokens_for_ai, # Pass legacy tokens to be combined
)
feedback_messages.extend(multi_feedback_messages)
elif feedback_tokens_for_ai:
# Legacy single feedback system - use transition-level filtering
feedback_metadata = activity_state.dict_metadata
if "metadata_feedback_filter" in transition:
filter_keys = transition["metadata_feedback_filter"]
feedback_metadata = {
k: v
for k, v in activity_state.dict_metadata.items()
if k in filter_keys
}
# Store and emit the feedback
if feedback:
# feedback is metadata language aware, doesn't need to be translated.
feedback = provide_feedback(
transition,
category,
step["question"],
feedback_tokens_for_ai,
user_response,
user_language,
username,
json.dumps(feedback_metadata),
json.dumps(new_metadata),
)
if feedback and feedback.strip():
feedback_messages.append(
{"name": "Feedback", "content": feedback}
)
# Store and emit all feedback messages
for feedback_msg in feedback_messages:
new_message = Message(
username="System", content=feedback, room_id=room.id
username=f"System ({feedback_msg['name'].title()})",
content=feedback_msg["content"],
room_id=room.id,
)
db.session.add(new_message)
db.session.commit()
@ -2123,8 +2244,8 @@ def handle_activity_response(room_name, user_response, username):
"chat_message",
{
"id": new_message.id,
"username": "System",
"content": feedback,
"username": f"System ({feedback_msg['name'].title()})",
"content": feedback_msg["content"],
},
room=room_name,
)
@ -2199,12 +2320,12 @@ def handle_activity_response(room_name, user_response, username):
db.session.commit()
# Emit the question again
question_content = f"Question: {step['question']}"
question_content = step["question"]
translated_question_content = translate_text(
question_content, user_language
)
new_message = Message(
username="System",
username="System (Question)",
content=translated_question_content,
room_id=room.id,
)
@ -2517,11 +2638,131 @@ def provide_feedback(
json_metadata,
json_new_metadata,
)
feedback += f"\n\nAI Feedback: {ai_feedback}"
feedback += f"\n\n{ai_feedback}"
return feedback
def provide_feedback_prompts(
transition,
category,
question,
feedback_prompts,
user_response,
user_language,
username,
json_metadata,
json_new_metadata,
legacy_tokens_for_ai="",
):
"""Generate feedback from multiple prompts"""
feedback_messages = []
# Parse full metadata once for filtering
full_metadata = json.loads(json_metadata)
# Add user_response to metadata for filtering purposes
full_metadata["user_response"] = user_response
for prompt in feedback_prompts:
prompt_name = prompt.get("name", "unnamed")
tokens_for_ai = prompt.get("tokens_for_ai", "")
# Apply per-prompt metadata filtering if specified
prompt_metadata = full_metadata
if "metadata_filter" in prompt:
filter_keys = prompt["metadata_filter"]
prompt_metadata = {
k: v for k, v in full_metadata.items() if k in filter_keys
}
# Check skip condition if specified
skip_condition = prompt.get("skip_condition")
if skip_condition:
should_skip = False
values = list(prompt_metadata.values())
if skip_condition == "all_null":
should_skip = all(value is None or value == "" or value == "None" for value in values)
elif skip_condition == "all_false":
should_skip = all(value is False or value == "False" for value in values)
elif skip_condition == "all_true":
should_skip = all(value is True or value == "True" for value in values)
if should_skip:
print(f"DEBUG: Skipping prompt '{prompt_name}' - skip_condition '{skip_condition}' met")
continue
# Special debug for Ship Status and Game Over
if prompt_name == "Ship Status":
print(f"DEBUG SHIP STATUS - filter_keys: {filter_keys}")
print(f"DEBUG SHIP STATUS - filtered metadata: {prompt_metadata}")
print(
f"DEBUG SHIP STATUS - user_sunk_ship_this_round = '{prompt_metadata.get('user_sunk_ship_this_round')}'"
)
print(
f"DEBUG SHIP STATUS - ai_sunk_ship_this_round = '{prompt_metadata.get('ai_sunk_ship_this_round')}'"
)
elif prompt_name == "Game Over":
print(f"DEBUG GAME OVER - filter_keys: {filter_keys}")
print(f"DEBUG GAME OVER - filtered metadata: {prompt_metadata}")
print(
f"DEBUG GAME OVER - game_over = '{prompt_metadata.get('game_over')}'"
)
print(
f"DEBUG GAME OVER - user_wins = '{prompt_metadata.get('user_wins')}'"
)
print(f"DEBUG GAME OVER - ai_wins = '{prompt_metadata.get('ai_wins')}'")
else:
if prompt_name == "Ship Status":
print(
f"DEBUG SHIP STATUS - NO metadata_filter, full metadata: {prompt_metadata}"
)
elif prompt_name == "Game Over":
print(
f"DEBUG GAME OVER - NO metadata_filter, full metadata: {prompt_metadata}"
)
# Combine legacy tokens with prompt-specific tokens
if legacy_tokens_for_ai:
tokens_for_ai = legacy_tokens_for_ai + " " + tokens_for_ai
# Add language instruction
tokens_for_ai += (
f" You must provide the feedback in the user's language: {user_language}."
)
# Add transition-specific AI feedback if present
if "ai_feedback" in transition:
tokens_for_ai += f" {transition['ai_feedback'].get('tokens_for_ai', '')}"
# Determine user_response for this prompt based on metadata filtering
filtered_user_response = user_response
if (
"metadata_filter" in prompt
and "user_response" not in prompt["metadata_filter"]
):
filtered_user_response = "" # Remove user response if not in filter
ai_feedback = generate_ai_feedback(
category,
question,
filtered_user_response,
tokens_for_ai,
username,
json.dumps(prompt_metadata), # Use filtered metadata for this prompt
json_new_metadata,
)
# Only add feedback if it has content
if ai_feedback and ai_feedback.strip():
feedback_messages.append(
{"name": prompt_name, "content": ai_feedback.strip()}
)
return feedback_messages
def translate_text(text, target_language):
# Guard clause for default language
target_language = target_language.lower().split()

78
research/activity29-battleship.yaml
View file

@ -187,21 +187,57 @@ sections:
If the user wants to exit, categorize as 'exit'.
Otherwise, categorize as 'invalid_move'.
feedback_tokens_for_ai: |
Write battleship feedback from the game's perspective that covers:
1. User's shot result - check user_hit_result in metadata:
- If "hit": Describe the impact and explosion
- If "miss": Describe the splash and fog of war
2. AI's shot result - report where the AI fired:
- If hit: Describe the damage to the player's ship
- If miss: Describe the near miss and ocean spray
3. CRITICAL: If ai_sunk_ship_this_round contains a ship name, express dismay that the AI destroyed the player's ship in 2 sentences describing the carnage at sea
4. CRITICAL: If user_sunk_ship_this_round contains a ship name, celebrate the player destroying the AI ship in 2 sentences describing the carnage at sea
5. CRITICAL: If game_over is true, announce the victory:
- If user_wins is true: Celebrate the player's total victory with excitement!
- If ai_wins is true: Express dismay at the player's defeat!
Describe the sights and sounds of naval warfare! You are the game system rooting for the player!
You are a battleship narrator. Each prompt has its own specific role - follow the individual prompt instructions precisely.
feedback_prompts:
- name: "Shot Report"
tokens_for_ai: |
🎯 Report ONLY the hit/miss results for both shots this turn. DO NOT report ship sinking or game over.
Check metadata:
- user_shot: Player's target position
- user_hit_result: "hit" or "miss"
- ai_shot: AI's target position
- ai_hit_result: "hit" or "miss"
Format: "🎯 Your shot at position [user_shot]: [user_hit_result]! 🤖 Enemy shot at position [ai_shot]: [ai_hit_result]!"
metadata_filter:
- user_shot
- ai_shot
- user_hit_result
- ai_hit_result
- user_response
- name: "Ship Status"
tokens_for_ai: |
A ship has been destroyed! Generate a dramatic 2-3 sentence description.
Metadata tells you which ship(s) were sunk:
- user_sunk_ship_this_round: The ship YOU destroyed (your victory)
- ai_sunk_ship_this_round: The ship that was destroyed by the enemy (your loss)
Format:
- If user destroyed a ship: "💥 You've sunk their [ship]! [2-3 dramatic sentences imagining how this warship meets its doom]"
- If AI destroyed your ship: "🔥 Your [ship] has been destroyed! [2-3 dramatic sentences about its destruction]"
- If both: Include both messages
metadata_filter:
- user_sunk_ship_this_round
- ai_sunk_ship_this_round
skip_condition: "all_null"
- name: "Game Over"
tokens_for_ai: |
The naval battle has ended! Generate an epic conclusion.
Based on the metadata:
- If user_wins is true: "🎉 TOTAL VICTORY! You have destroyed all enemy ships! The seas belong to you, Admiral. Your tactical brilliance has secured complete naval supremacy."
- If ai_wins is true: "💀 DEFEAT! All your ships have been destroyed. Your fleet lies scattered across the ocean floor. The enemy's superior strategy has prevailed."
Make it dramatic and final - this is the end of the battle!
metadata_filter:
- game_over
- user_wins
- ai_wins
skip_condition: "all_false"
processing_script: |
import random
@ -846,16 +882,6 @@ sections:
The user shot seems valid.
metadata_tmp_add:
user_shot: "the-users-response"
metadata_feedback_filter:
- user_hit_result
- ai_hit_result
- ai_shot
- user_shot
- user_sunk_ship_this_round
- ai_sunk_ship_this_round
- game_over
- user_wins
- ai_wins
next_section_and_step: "section_1:step_2"
invalid_move:
content_blocks:
@ -877,8 +903,6 @@ sections:
tokens_for_ai: |
If the user wants to restart or play again, categorize as 'restart'.
If the user wants to exit, categorize as 'exit'.
feedback_tokens_for_ai: |
Acknowledge the user's choice appropriately.
buckets:
- restart
- exit

78
research/activity29-testship.yaml
View file

@ -165,21 +165,57 @@ sections:
If the user wants to exit, categorize as 'exit'.
Otherwise, categorize as 'invalid_move'.
feedback_tokens_for_ai: |
Write battleship feedback from the game's perspective that covers:
1. User's shot result - check user_hit_result in metadata:
- If "hit": Describe the impact and explosion
- If "miss": Describe the splash and fog of war
2. AI's shot result - report where the AI fired:
- If hit: Describe the damage to the player's ship
- If miss: Describe the near miss and ocean spray
3. CRITICAL: If ai_sunk_ship_this_round contains a ship name, express dismay that the AI destroyed the player's ship in 2 sentences describing the carnage at sea
4. CRITICAL: If user_sunk_ship_this_round contains a ship name, celebrate the player destroying the AI ship in 2 sentences describing the carnage at sea
5. CRITICAL: If game_over is true, announce the victory:
- If user_wins is true: Celebrate the player's total victory with excitement!
- If ai_wins is true: Express dismay at the player's defeat!
Describe the sights and sounds of naval warfare! You are the game system rooting for the player!
You are a battleship narrator. Each prompt has its own specific role - follow the individual prompt instructions precisely.
feedback_prompts:
- name: "Shot Report"
tokens_for_ai: |
🎯 Report ONLY the hit/miss results for both shots this turn. DO NOT report ship sinking or game over.
Check metadata:
- user_shot: Player's target position
- user_hit_result: "hit" or "miss"
- ai_shot: AI's target position
- ai_hit_result: "hit" or "miss"
Format: "🎯 Your shot at position [user_shot]: [user_hit_result]! 🤖 Enemy shot at position [ai_shot]: [ai_hit_result]!"
metadata_filter:
- user_shot
- ai_shot
- user_hit_result
- ai_hit_result
- user_response
- name: "Ship Status"
tokens_for_ai: |
A ship has been destroyed! Generate a dramatic 2-3 sentence description.
Metadata tells you which ship(s) were sunk:
- user_sunk_ship_this_round: The ship YOU destroyed (your victory)
- ai_sunk_ship_this_round: The ship that was destroyed by the enemy (your loss)
Format:
- If user destroyed a ship: "💥 You've sunk their [ship]! [2-3 dramatic sentences imagining how this warship meets its doom]"
- If AI destroyed your ship: "🔥 Your [ship] has been destroyed! [2-3 dramatic sentences about its destruction]"
- If both: Include both messages
metadata_filter:
- user_sunk_ship_this_round
- ai_sunk_ship_this_round
skip_condition: "all_null"
- name: "Game Over"
tokens_for_ai: |
The naval battle has ended! Generate an epic conclusion.
Based on the metadata:
- If user_wins is true: "🎉 TOTAL VICTORY! You have destroyed all enemy ships! The seas belong to you, Admiral. Your tactical brilliance has secured complete naval supremacy."
- If ai_wins is true: "💀 DEFEAT! All your ships have been destroyed. Your fleet lies scattered across the ocean floor. The enemy's superior strategy has prevailed."
Make it dramatic and final - this is the end of the battle!
metadata_filter:
- game_over
- user_wins
- ai_wins
skip_condition: "all_false"
processing_script: |
import random
@ -814,16 +850,6 @@ sections:
The user shot seems valid.
metadata_tmp_add:
user_shot: "the-users-response"
metadata_feedback_filter:
- user_hit_result
- ai_hit_result
- ai_shot
- user_shot
- user_sunk_ship_this_round
- ai_sunk_ship_this_round
- game_over
- user_wins
- ai_wins
next_section_and_step: "section_1:step_2"
invalid_move:
content_blocks:
@ -845,8 +871,6 @@ sections:
tokens_for_ai: |
If the user wants to restart or play again, categorize as 'restart'.
If the user wants to exit, categorize as 'exit'.
feedback_tokens_for_ai: |
Acknowledge the user's choice appropriately.
buckets:
- restart
- exit

108
research/guarded_ai.py
View file

@ -122,7 +122,7 @@ def generate_ai_feedback(category, question, user_response, tokens_for_ai, metad
return f"Error: {e}"
# Provide feedback based on the category
# Provide feedback based on the category (legacy single feedback system)
def provide_feedback(
transition,
category,
@ -150,6 +150,68 @@ def provide_feedback(
return feedback
# Provide feedback using multiple prompts (new system)
def provide_feedback_prompts(
transition,
category,
question,
feedback_prompts,
user_response,
user_language,
metadata,
legacy_tokens_for_ai="",
):
"""Generate feedback from multiple prompts"""
feedback_messages = []
# Add user_response to metadata for filtering purposes
full_metadata = metadata.copy()
full_metadata["user_response"] = user_response
for prompt in feedback_prompts:
prompt_name = prompt.get("name", "unnamed")
tokens_for_ai = prompt.get("tokens_for_ai", "")
# Apply per-prompt metadata filtering if specified
prompt_metadata = full_metadata
if "metadata_filter" in prompt:
filter_keys = prompt["metadata_filter"]
prompt_metadata = {
k: v for k, v in full_metadata.items() if k in filter_keys
}
# Combine legacy tokens with prompt-specific tokens
if legacy_tokens_for_ai:
tokens_for_ai = legacy_tokens_for_ai + " " + tokens_for_ai
# Add language instruction
tokens_for_ai += f" Provide the feedback in {user_language}."
# Add transition-specific AI feedback if present
if "ai_feedback" in transition:
tokens_for_ai += f" {transition['ai_feedback'].get('tokens_for_ai', '')}"
# Determine user_response for this prompt based on metadata filtering
filtered_user_response = user_response
if (
"metadata_filter" in prompt
and "user_response" not in prompt["metadata_filter"]
):
filtered_user_response = "" # Remove user response if not in filter
ai_feedback = generate_ai_feedback(
category, question, filtered_user_response, tokens_for_ai, prompt_metadata
)
# Only add feedback if it has content and isn't exactly the STFU token
if ai_feedback and ai_feedback.strip() and ai_feedback.strip() != "STFU":
feedback_messages.append(
{"name": prompt_name, "content": ai_feedback.strip()}
)
return feedback_messages
def execute_processing_script(metadata, script):
# Prepare the local environment for the script
local_env = {"metadata": metadata, "script_result": None}
@ -413,16 +475,40 @@ def simulate_activity(yaml_file_path):
print(f"\nMetadata: {json.dumps(metadata, indent=2)}")
# Provide feedback based on the category
feedback = provide_feedback(
transition,
category,
question,
user_response,
user_language,
step.get("feedback_tokens_for_ai", ""),
metadata,
)
print(f"\nFeedback: {feedback}")
feedback_messages = []
if "feedback_prompts" in step:
# New multi-prompt system - legacy tokens get combined with each prompt
multi_feedback_messages = provide_feedback_prompts(
transition,
category,
question,
step["feedback_prompts"],
user_response,
user_language,
metadata,
step.get(
"feedback_tokens_for_ai", ""
), # Pass legacy tokens to be combined
)
feedback_messages.extend(multi_feedback_messages)
elif step.get("feedback_tokens_for_ai"):
# Legacy single feedback system - only if no feedback_prompts
feedback = provide_feedback(
transition,
category,
question,
user_response,
user_language,
step.get("feedback_tokens_for_ai", ""),
metadata,
)
if feedback and feedback.strip():
feedback_messages.append({"name": "Feedback", "content": feedback})
# Display all feedback messages
for feedback_msg in feedback_messages:
print(f"\n{feedback_msg['name']}: {feedback_msg['content']}")
if category not in [
"partial_understanding",

12
static/js/utils.js Normal file
View file

@ -0,0 +1,12 @@
/**
* Utility functions for the OpenCompletion application
*/
/**
* Convert a string to a URL-friendly slug
* @param {string} str - The string to slugify
* @returns {string} - The slugified string
*/
function slugify(str) {
return str.toLowerCase().replace(/\s+/g, '-').replace(/[^\w-]+/g, '');
}

284
templates/base.html
View file

@ -22,6 +22,9 @@
<!-- Include DOMPurify to sanitize HTML and prevent XSS attacks -->
<script src="https://cdn.jsdelivr.net/npm/dompurify@2/dist/purify.min.js"></script>
<!-- Include utility functions -->
<script src="{{ url_for('static', filename='js/utils.js') }}"></script>
<script>
// Connect to the server using socket.io
@ -76,6 +79,19 @@
display: block;
}
/* Styling for the message body wrapper that contains header and content */
.message-body {
width: 100%;
display: flex;
flex-direction: column;
}
/* Styling for the message header (username/model) */
.message-header {
width: 100%;
margin-bottom: 0;
}
/* Styling for the message div holding html/markdown content */
.message-content {
width: 100%;
@ -121,6 +137,85 @@
#rooms-list {
border-right: 1px solid #e1e1e1;
overflow-y: auto;
padding: 10px;
}
/* Styling for site header */
#site-header {
margin-bottom: 20px;
text-align: center;
}
#opencompletion-btn {
width: 100%;
padding: 10px;
background-color: #007bff;
color: white;
border: none;
border-radius: 5px;
font-size: 14px;
font-weight: bold;
cursor: pointer;
transition: background-color 0.3s;
}
#opencompletion-btn:hover {
background-color: #0056b3;
}
/* Styling for new room creation section */
#new-room-section {
margin-bottom: 20px;
padding: 10px;
border: 1px solid #e1e1e1;
border-radius: 5px;
background-color: #f8f9fa;
}
#new-room-section h4 {
margin: 0 0 10px 0;
font-size: 14px;
color: #495057;
}
#new-room-name {
width: 100%;
padding: 8px;
border: 1px solid #ced4da;
border-radius: 3px;
font-size: 12px;
resize: vertical;
margin-bottom: 10px;
box-sizing: border-box;
}
#create-room-btn {
width: 100%;
padding: 8px;
background-color: #28a745;
color: white;
border: none;
border-radius: 3px;
font-size: 12px;
cursor: pointer;
transition: background-color 0.3s;
}
#create-room-btn:hover {
background-color: #218838;
}
/* Styling for public rooms header */
#public-rooms-header {
margin-bottom: 10px;
}
#public-rooms-header h4 {
margin: 0;
font-size: 14px;
color: #495057;
border-bottom: 1px solid #e1e1e1;
padding-bottom: 5px;
}
/* Styling for the unordered list in the rooms list */
@ -243,6 +338,78 @@
.utility-belt {
padding: 10px;
}
/* Activity controls styling */
#activity-controls {
margin-top: 20px;
padding: 10px;
border-top: 1px solid #e1e1e1;
}
#activity-controls h3 {
margin-top: 0;
margin-bottom: 10px;
}
#current-activity-info {
background-color: #f0f0f0;
padding: 10px;
border-radius: 5px;
margin-bottom: 10px;
}
#current-activity-info p {
margin: 0 0 10px 0;
}
#activity-controls button {
background-color: #4CAF50;
color: white;
border: none;
padding: 8px 16px;
text-align: center;
text-decoration: none;
display: inline-block;
font-size: 14px;
margin: 4px 2px;
cursor: pointer;
border-radius: 4px;
}
#cancel-activity-btn {
background-color: #f44336;
}
#activity-controls button:hover {
opacity: 0.8;
}
#activity-select {
width: 100%;
max-width: 100%;
box-sizing: border-box;
padding: 5px;
border: 1px solid #e1e1e1;
border-radius: 4px;
font-size: 14px;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
/* Model and voice select dropdowns styling */
#model-select, #voice-select, #model-select-mobile, #voice-select-mobile {
width: 100%;
max-width: 100%;
box-sizing: border-box;
padding: 5px;
border: 1px solid #e1e1e1;
border-radius: 4px;
font-size: 14px;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
/* Media query for mobile devices */
@media (max-width: 768px) {
@ -262,7 +429,6 @@
</style>
</head>
<body>
<a href="https://github.com/russellballestrini/opencompletion#interacting-with-language-models" target="_blank">🚀 docs for models & other commands, also try /help</a>
<!-- Search form -->
<div id="search-form" style="width: 90%;">
<form action="/search" method="get">
@ -279,6 +445,10 @@
<div id="room-list-modal-content">
<button id="close-modal-button" onclick="closeModal()">×</button>
<div id="utility-belt-mobile">
<div id="username-chooser-mobile">
<label for="username-input-mobile">Username:</label>
<input type="text" id="username-input-mobile" placeholder="guest" maxlength="50" style="width: 100%; padding: 4px; margin-top: 2px; border: 1px solid #ccc; border-radius: 3px;">
</div>
<div id="model-chooser-mobile">
<label for="model-select-mobile">Choose Model:</label>
<select id="model-select-mobile">
@ -296,6 +466,28 @@
<option value="shimmer">Shimmer</option>
</select>
</div>
<div id="auto-play-tts-mobile">
<button id="auto-play-tts-btn-mobile" onclick="toggleAutoPlayTTS()" style="width: 100%; margin-top: 10px; background-color: #f44336; color: white; border: none; padding: 8px; border-radius: 4px; cursor: pointer;">
Auto-Play TTS: OFF
</button>
</div>
<div id="activity-controls-mobile">
<h3>Activities</h3>
<div id="current-activity-info-mobile" style="display: none;">
<p>Current Activity: <span id="current-activity-name-mobile"></span></p>
<button id="cancel-activity-btn-mobile" onclick="cancelActivity()">Cancel Activity</button>
</div>
<div id="activity-list-section-mobile">
<div style="display: grid; grid-template-columns: minmax(0, 1fr) auto; align-items: center; gap: 5px; margin-bottom: 5px;">
<select id="activity-select-mobile" style="width: 100%; max-width: 100%; box-sizing: border-box;">
<option value="">-- Select an Activity --</option>
</select>
<button id="refresh-activities-btn-mobile" onclick="refreshActivityList()">🔄</button>
</div>
<button id="load-activity-btn-mobile" onclick="loadSelectedActivityMobile()" style="margin-top: 5px;">Load Activity</button>
</div>
</div>
<div id="user-lists-mobile">
<div id="active-users-list">
<h3>Active Users</h3>
@ -321,6 +513,24 @@
<!-- Chatroom list -->
<div class="main-container">
<div id="rooms-list">
<!-- opencompletion.com button -->
<div id="site-header">
<button id="opencompletion-btn" onclick="window.open('https://opencompletion.com', '_blank')">
opencompletion.com
</button>
</div>
<!-- New room creation section -->
<div id="new-room-section">
<textarea id="new-room-name" placeholder="Enter room name..." rows="2" maxlength="100"></textarea>
<button id="create-room-btn" onclick="createNewRoom()">Create Room</button>
</div>
<!-- Public rooms header -->
<div id="public-rooms-header">
<h4>Public Rooms</h4>
</div>
<ul id="rooms-list-ul">
<!-- Loop through rooms and create list items for each room -->
{% for room in rooms %}
@ -370,9 +580,20 @@
const modal = document.getElementById("room-list-modal");
const modalContent = document.getElementById("room-list-modal-content");
const closeButton = document.getElementById("close-modal-button");
// Clone the site header, new room section, public rooms header, and rooms list
const siteHeader = document.getElementById("site-header").cloneNode(true);
const newRoomSection = document.getElementById("new-room-section").cloneNode(true);
const publicRoomsHeader = document.getElementById("public-rooms-header").cloneNode(true);
const roomsList = document.getElementById("rooms-list-ul").cloneNode(true);
document.getElementById("rooms-list-modal-content").innerHTML = ''; // Clear previous content
// Clear previous content and add all sections
document.getElementById("rooms-list-modal-content").innerHTML = '';
document.getElementById("rooms-list-modal-content").appendChild(siteHeader);
document.getElementById("rooms-list-modal-content").appendChild(newRoomSection);
document.getElementById("rooms-list-modal-content").appendChild(publicRoomsHeader);
document.getElementById("rooms-list-modal-content").appendChild(roomsList);
modal.style.display = "flex";
modalContent.style.display = "block";
closeButton.style.display = "block";
@ -387,6 +608,65 @@
modalContent.style.display = "none";
closeButton.style.display = "none";
}
// Function to update all room links with current URL parameters
function updateRoomLinksWithCurrentParams() {
const currentParams = new URLSearchParams(window.location.search).toString();
// Update desktop room links
const roomLinks = document.querySelectorAll('#rooms-list-ul a[href*="/chat/"]');
roomLinks.forEach(link => {
const url = new URL(link.href, window.location.origin);
const roomPath = url.pathname; // e.g., "/chat/room-name"
link.href = roomPath + (currentParams ? '?' + currentParams : '');
});
// Update mobile room links (if they exist)
const mobileRoomLinks = document.querySelectorAll('#rooms-list-modal-content a[href*="/chat/"]');
mobileRoomLinks.forEach(link => {
const url = new URL(link.href, window.location.origin);
const roomPath = url.pathname; // e.g., "/chat/room-name"
link.href = roomPath + (currentParams ? '?' + currentParams : '');
});
}
// Function to create a new room
function createNewRoom() {
const roomName = document.getElementById("new-room-name").value.trim();
if (!roomName) {
alert("Please enter a room name.");
return;
}
// Slugify the room name to make it URL-friendly
const slugifiedRoomName = slugify(roomName);
if (!slugifiedRoomName) {
alert("Please enter a valid room name with at least some letters or numbers.");
return;
}
// Get current URL parameters to maintain username, model, voice, etc.
const urlParams = new URLSearchParams(window.location.search);
const currentParams = urlParams.toString();
// Navigate to the new room using the slugified name
window.location.href = `/chat/${slugifiedRoomName}${currentParams ? '?' + currentParams : ''}`;
}
// Add event listener for Enter key in the new room textarea
document.addEventListener('DOMContentLoaded', function() {
const newRoomTextarea = document.getElementById("new-room-name");
if (newRoomTextarea) {
newRoomTextarea.addEventListener("keydown", function(event) {
if (event.key === "Enter" && !event.shiftKey) {
event.preventDefault();
createNewRoom();
}
});
}
});
// Add event listener to the hamburger button
document.getElementById("hamburger-button").addEventListener("click", openModal);

500
templates/chat.html
View file

@ -20,6 +20,10 @@
<a href="/download_chat_history_md?room_name={{ room_name }}" download="{{ room_name }}.md">Markdown</a>
</div>
<br>
<div>
<label for="username-input">Username</label>
<input type="text" id="username-input" placeholder="guest" maxlength="50" style="width: 100%; padding: 4px; margin-top: 2px; border: 1px solid #ccc; border-radius: 3px;">
</div>
<div>
<label for="model-select">Model</label>
<select id="model-select">
@ -37,6 +41,29 @@
<option value="shimmer">Shimmer</option>
</select>
</div>
<div>
<button id="auto-play-tts-btn" onclick="toggleAutoPlayTTS()" style="width: 100%; margin-top: 10px; background-color: #f44336; color: white; border: none; padding: 8px; border-radius: 4px; cursor: pointer;">
Auto-Play TTS: OFF
</button>
</div>
<div id="activity-controls">
<h3>Activities</h3>
<div id="current-activity-info" style="display: none;">
<p>Current Activity: <span id="current-activity-name"></span></p>
<button id="cancel-activity-btn" onclick="cancelActivity()">Cancel Activity</button>
</div>
<div id="activity-list-section">
<div style="display: grid; grid-template-columns: minmax(0, 1fr) auto; align-items: center; gap: 5px; margin-bottom: 5px;">
<select id="activity-select">
<option value="">-- Select an Activity --</option>
</select>
<button id="refresh-activities-btn" onclick="refreshActivityList()">🔄</button>
</div>
<button id="load-activity-btn" onclick="loadSelectedActivity()" style="margin-top: 5px;">Load Activity</button>
</div>
</div>
<div id="user-lists">
<div id="active-users-list">
<h3>Active Users</h3>
@ -59,7 +86,14 @@
const API_KEY = "dummy-api-key";
const TTS_API_URL = "https://speech.ai.unturf.com/v1/audio/speech";
const urlParams = new URLSearchParams(window.location.search);
const username = urlParams.get("username");
let username = urlParams.get("username") || "guest"; // Default to "guest" if no username in URL
// If no username was in the URL, add it now
if (!urlParams.get("username")) {
const newUrl = new URL(window.location.href);
newUrl.searchParams.set("username", username);
window.history.replaceState({}, '', newUrl);
}
const room_name = "{{ room_name }}";
// Global constants for valid voices
@ -78,11 +112,17 @@ const dompurify_config = {
// keeping track of scrolling to prevent autoscrolling.
let userHasScrolledUp = false;
let currentAudio = null; // To keep track of the currently playing audio
let currentQueuedAudio = null; // To keep track of currently playing queued TTS audio
let audioCache = {}; // Cache to store audio blobs
// Flag to prevent mutual updates on desktop/mobile
let isSyncingDropdowns = false;
// Auto-play TTS state
let autoPlayTTS = localStorage.getItem('autoPlayTTS') === 'true' || false;
let ttsQueue = [];
let isPlayingTTS = false;
// Function to sanitize the username
function sanitizeUsername(username) {
// Split the username on commas and take the first part.
@ -90,29 +130,54 @@ function sanitizeUsername(username) {
return username.split(',')[0].trim();
}
// Function to sync dropdowns and update the query string
function syncDropdownsAndQueryString() {
const sanitizedUsername = sanitizeUsername(username);
// Function to sync all inputs and update the query string
function syncInputsAndQueryString() {
const usernameInputDesktop = document.getElementById("username-input");
const usernameInputMobile = document.getElementById("username-input-mobile");
const modelSelectDesktop = document.getElementById("model-select");
const voiceSelectDesktop = document.getElementById("voice-select");
const modelSelectMobile = document.getElementById("model-select-mobile");
const voiceSelectMobile = document.getElementById("voice-select-mobile");
// Determine the current model and voice from any dropdown
// Get current values
const currentUsername = usernameInputDesktop?.value || username || "guest";
const currentModel = modelSelectDesktop.value;
const currentVoice = VALID_VOICES.includes(voiceSelectDesktop.value) ? voiceSelectDesktop.value : 'onyx';
// Sync both desktop and mobile dropdowns
// Update global username variable
username = currentUsername;
const sanitizedUsername = sanitizeUsername(username);
// Sync username inputs
if (usernameInputDesktop) usernameInputDesktop.value = sanitizedUsername;
if (usernameInputMobile) usernameInputMobile.value = sanitizedUsername;
// Sync dropdowns
modelSelectDesktop.value = currentModel;
voiceSelectDesktop.value = currentVoice;
modelSelectMobile.value = currentModel;
voiceSelectMobile.value = currentVoice;
if (modelSelectMobile) modelSelectMobile.value = currentModel;
if (voiceSelectMobile) voiceSelectMobile.value = currentVoice;
// Save to localStorage for persistence
localStorage.setItem('selectedModel', currentModel);
localStorage.setItem('selectedVoice', currentVoice);
// Update URL
const newUrl = new URL(window.location.href);
newUrl.searchParams.set("username", sanitizedUsername);
newUrl.searchParams.set("model", currentModel);
newUrl.searchParams.set("voice", currentVoice);
window.history.replaceState({}, '', newUrl);
// Update room links with new parameters
if (typeof updateRoomLinksWithCurrentParams === 'function') {
updateRoomLinksWithCurrentParams();
}
}
// Backward compatibility
function syncDropdownsAndQueryString() {
syncInputsAndQueryString();
}
document.addEventListener('DOMContentLoaded', (event) => {
@ -121,6 +186,9 @@ document.addEventListener('DOMContentLoaded', (event) => {
const voiceSelectDesktop = document.getElementById("voice-select");
const modelSelectMobile = document.getElementById("model-select-mobile");
const voiceSelectMobile = document.getElementById("voice-select-mobile");
// Initialize auto-play TTS button state from localStorage
updateAutoPlayTTSDisplay();
// Function to populate the dropdown
function populateModelDropdown(models) {
@ -172,13 +240,24 @@ document.addEventListener('DOMContentLoaded', (event) => {
userHasScrolledUp = distanceFromBottom > 5;
});
// Set initial model and voice from query string
const initialModel = urlParams.get("model") || "None";
const initialVoice = urlParams.get("voice") || "onyx";
// Set initial model, voice, and username from URL, localStorage, or defaults
const initialModel = urlParams.get("model") || localStorage.getItem('selectedModel') || "None";
const initialVoice = urlParams.get("voice") || localStorage.getItem('selectedVoice') || "onyx";
const initialUsername = username; // Already set to URL param or "guest"
modelSelectDesktop.value = initialModel;
voiceSelectDesktop.value = initialVoice;
modelSelectMobile.value = initialModel;
voiceSelectMobile.value = initialVoice;
// Set initial username values
const usernameInputDesktop = document.getElementById("username-input");
const usernameInputMobile = document.getElementById("username-input-mobile");
if (usernameInputDesktop) usernameInputDesktop.value = initialUsername;
if (usernameInputMobile) usernameInputMobile.value = initialUsername;
// Initial sync to ensure localStorage and URL are updated with current values
syncInputsAndQueryString();
// Add event listeners for desktop dropdowns
modelSelectDesktop.addEventListener("change", () => {
@ -213,6 +292,39 @@ document.addEventListener('DOMContentLoaded', (event) => {
syncDropdownsAndQueryString();
isSyncingDropdowns = false;
});
// Add event listeners for username inputs
if (usernameInputDesktop) {
usernameInputDesktop.addEventListener("input", () => {
if (isSyncingDropdowns) return;
isSyncingDropdowns = true;
if (usernameInputMobile) {
usernameInputMobile.value = usernameInputDesktop.value;
}
syncInputsAndQueryString();
isSyncingDropdowns = false;
});
usernameInputDesktop.addEventListener("blur", () => {
syncInputsAndQueryString();
});
}
if (usernameInputMobile) {
usernameInputMobile.addEventListener("input", () => {
if (isSyncingDropdowns) return;
isSyncingDropdowns = true;
if (usernameInputDesktop) {
usernameInputDesktop.value = usernameInputMobile.value;
}
syncInputsAndQueryString();
isSyncingDropdowns = false;
});
usernameInputMobile.addEventListener("blur", () => {
syncInputsAndQueryString();
});
}
});
// Socket event when the user connects
@ -220,8 +332,8 @@ socket.on("connect", () => {
// Sanitize the username before joining
const sanitizedUsername = sanitizeUsername(username);
socket.emit("join", {"username": sanitizedUsername, "room_name": room_name});
// Sync dropdowns and update the query string
syncDropdownsAndQueryString();
// Sync inputs and update the query string
syncInputsAndQueryString();
});
// Function to update the active and inactive user lists in the DOM
@ -322,8 +434,9 @@ socket.on('update_room_list', function(updatedRoom) {
}
});
// Function to read text using TTS
// Function to read text using TTS (for manual button clicks)
async function speakText(text, playButton, messageId) {
console.log("speakText called with:", {text: text, messageId, autoPlay: autoPlayTTS});
const voice = document.getElementById("voice-select").value;
const cacheKey = `${messageId}-${voice}`; // Unique cache key for each message and voice
@ -377,6 +490,152 @@ async function speakText(text, playButton, messageId) {
}
}
// Function to read text using TTS (for queued auto-play)
async function speakTextQueued(text, playButton, messageId) {
return new Promise((resolve, reject) => {
const voice = document.getElementById("voice-select").value;
const cacheKey = `${messageId}-${voice}`;
const cleanText = text.replace(/[^a-zA-Z0-9\s.,!?]/g, '');
const playAudio = (audio) => {
currentQueuedAudio = audio; // Track the currently playing queued audio
audio.onended = () => {
console.log("TTS finished for:", messageId);
currentQueuedAudio = null; // Clear when finished
resolve();
};
audio.onerror = () => {
console.error("TTS audio error for:", messageId);
currentQueuedAudio = null; // Clear on error
reject(new Error("Audio playback failed"));
};
audio.play().catch(reject);
};
// Check if audio is cached
if (audioCache[cacheKey]) {
playAudio(audioCache[cacheKey]);
return;
}
// Fetch new audio
fetch(TTS_API_URL, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${API_KEY}`
},
body: JSON.stringify({
model: 'tts-1',
voice: voice,
input: cleanText
})
})
.then(response => {
if (!response.ok) {
throw new Error(`HTTP error! status: ${response.status}`);
}
return response.blob();
})
.then(audioBlob => {
const audioUrl = URL.createObjectURL(audioBlob);
const audio = new Audio(audioUrl);
audio.playbackRate = 0.9;
audioCache[cacheKey] = audio;
playAudio(audio);
})
.catch(reject);
});
}
// Function to add TTS to queue
function queueTTS(text, playButton, messageId) {
ttsQueue.push({ text, playButton, messageId });
console.log("Added to TTS queue:", messageId, "Queue length:", ttsQueue.length);
processNextTTS();
}
// Function to process the next TTS in queue
function processNextTTS() {
if (isPlayingTTS || ttsQueue.length === 0) {
return;
}
isPlayingTTS = true;
const { text, playButton, messageId } = ttsQueue.shift();
console.log("Processing TTS from queue:", messageId);
// Use non-blocking async processing
speakTextQueued(text, playButton, messageId)
.then(() => {
console.log("TTS completed successfully for:", messageId);
})
.catch((error) => {
console.error("TTS error:", error);
})
.finally(() => {
isPlayingTTS = false;
// Schedule next item with minimal delay to prevent blocking
setTimeout(processNextTTS, 10);
});
}
// Function to update auto-play TTS button display
function updateAutoPlayTTSDisplay() {
const autoPlayBtn = document.getElementById("auto-play-tts-btn");
const autoPlayBtnMobile = document.getElementById("auto-play-tts-btn-mobile");
if (autoPlayTTS) {
autoPlayBtn.textContent = "Auto-Play TTS: ON";
autoPlayBtn.style.backgroundColor = "#4CAF50";
autoPlayBtnMobile.textContent = "Auto-Play TTS: ON";
autoPlayBtnMobile.style.backgroundColor = "#4CAF50";
} else {
autoPlayBtn.textContent = "Auto-Play TTS: OFF";
autoPlayBtn.style.backgroundColor = "#f44336";
autoPlayBtnMobile.textContent = "Auto-Play TTS: OFF";
autoPlayBtnMobile.style.backgroundColor = "#f44336";
// Clear queue when turning off
ttsQueue = [];
isPlayingTTS = false;
}
}
// Function to toggle auto-play TTS
function toggleAutoPlayTTS() {
autoPlayTTS = !autoPlayTTS;
console.log("Auto-play TTS toggled to:", autoPlayTTS);
// Save to localStorage
localStorage.setItem('autoPlayTTS', autoPlayTTS.toString());
// If turning off, clear the queue and stop current audio
if (!autoPlayTTS) {
console.log("Clearing TTS queue, had", ttsQueue.length, "items");
ttsQueue = [];
isPlayingTTS = false;
// Stop any currently playing audio
if (currentAudio) {
currentAudio.pause();
currentAudio.currentTime = 0;
if (currentAudio.playButton) {
currentAudio.playButton.textContent = "Play";
}
currentAudio = null;
}
// Stop any currently playing queued TTS audio
if (currentQueuedAudio) {
currentQueuedAudio.pause();
currentQueuedAudio.currentTime = 0;
currentQueuedAudio = null;
}
}
updateAutoPlayTTSDisplay();
}
// Function to toggle audio playback
function toggleAudioPlayback(audio, playButton) {
if (currentAudio && currentAudio !== audio) {
@ -467,6 +726,20 @@ socket.on("chat_message", (data) => {
// Scroll to the bottom of the chat container to show the new message.
if (data.id) {
document.getElementById("chat").scrollTop = document.getElementById("chat").scrollHeight;
// Auto-play TTS if enabled and message has content - AFTER buttons are created
if (autoPlayTTS && data.content && data.content.trim() !== "") {
setTimeout(() => {
// Find the play button after buttons have been created
const buttons = messageWrapper.querySelectorAll("button");
const playButton = Array.from(buttons).find(btn => btn.textContent === "Play");
console.log("Auto-play TTS: enabled=", autoPlayTTS, "content=", data.content, "playButton=", playButton);
if (playButton) {
console.log("Queueing TTS for message:", data.id);
queueTTS(data.content, playButton, data.id);
}
}, 10); // Very short delay to let buttons be created
}
}
});
@ -543,12 +816,15 @@ socket.on("delete_processing_message", (msg_id) => {
tempMessages.forEach((tempMessage) => {
tempMessage.remove();
});
// Clear the message buffer for the corresponding message ID
// Clear the message buffer and header for the corresponding message ID
delete messageBuffers[msg_id];
delete messageHeaders[msg_id];
});
// A dictionary to hold buffers for each message ID
const messageBuffers = {};
// A dictionary to track message headers (username/model) for each message ID
const messageHeaders = {};
// Socket event for receiving chunks of a message
socket.on("message_chunk", (data) => {
@ -566,9 +842,20 @@ socket.on("message_chunk", (data) => {
// If the message-content div doesn't exist, create it
if (!messageWrapper.querySelector(".message-content")) {
// Create a message body wrapper to contain both header and content
const messageBodyWrapper = document.createElement("div");
messageBodyWrapper.className = "message-body";
messageWrapper.appendChild(messageBodyWrapper);
// Create header element for username/model
const headerElement = document.createElement("div");
headerElement.className = "message-header";
messageBodyWrapper.appendChild(headerElement);
// Create content element for actual message content
targetMessageElement = document.createElement("div");
targetMessageElement.className = "message-content";
messageWrapper.appendChild(targetMessageElement);
messageBodyWrapper.appendChild(targetMessageElement);
} else {
targetMessageElement = messageWrapper.querySelector(".message-content");
}
@ -578,14 +865,29 @@ socket.on("message_chunk", (data) => {
messageBuffers[data.id] = "";
}
// Store header info on first chunk and update header element
if (data.is_first_chunk && data.username && data.model_name) {
messageHeaders[data.id] = {
username: data.username,
model_name: data.model_name
};
// Update header element
const headerElement = messageWrapper.querySelector(".message-header");
if (headerElement) {
const headerContent = `**${data.username} (${data.model_name}):**`;
headerElement.innerHTML = DOMPurify.sanitize(marked.marked(headerContent), dompurify_config);
}
}
// Append the chunk to the buffer
messageBuffers[data.id] += data.content;
// Process the entire buffer with marked and set it as the content of the target element
// Process just the content and set it in the content element
const sanitizedContent = DOMPurify.sanitize(marked.marked(messageBuffers[data.id]), dompurify_config);
targetMessageElement.innerHTML = sanitizedContent;
// Store the raw markdown in a data attribute for later use in editing
// Store the raw markdown in a data attribute for later use in editing (without header for clean editing)
targetMessageElement.dataset.rawMarkdown = messageBuffers[data.id];
// Apply syntax highlighting to code blocks within the content
@ -623,13 +925,35 @@ socket.on("message_chunk", (data) => {
const playButton = document.createElement("button");
playButton.textContent = "Play";
playButton.onclick = () => {
const fullText = targetMessageElement.textContent || targetMessageElement.innerText;
speakText(fullText, playButton, data.id);
// Use content from the content element (clean text without header)
const cleanText = targetMessageElement.textContent || targetMessageElement.innerText || "";
speakText(cleanText, playButton, data.id);
};
buttonContainer.appendChild(playButton);
// Append the button container before the message content
messageWrapper.insertBefore(buttonContainer, targetMessageElement);
// Insert the button container at the beginning of the message wrapper (before header and content)
messageWrapper.insertBefore(buttonContainer, messageWrapper.firstChild);
// Auto-play TTS if enabled and message is complete (only when streaming finishes)
console.log("DEBUG: Streaming complete check:", {
autoPlayTTS: autoPlayTTS,
is_complete: data.is_complete,
hasBuffer: !!messageBuffers[data.id],
bufferContent: messageBuffers[data.id] ? messageBuffers[data.id].substring(0, 50) + "..." : "none",
messageId: data.id
});
if (autoPlayTTS && data.is_complete && messageBuffers[data.id] && messageBuffers[data.id].trim() !== "") {
const playButton = Array.from(buttonContainer.querySelectorAll("button")).find(btn => btn.textContent === "Play");
console.log("DEBUG: Found play button for streaming TTS:", playButton ? "YES" : "NO", playButton?.textContent);
if (playButton) {
setTimeout(() => {
// Use content from the content element (clean text without header)
const cleanText = targetMessageElement.textContent || targetMessageElement.innerText || "";
console.log("DEBUG: Queueing streaming TTS:", data.id, cleanText.substring(0, 50) + "...");
queueTTS(cleanText, playButton, data.id);
}, 50); // Small delay to let the message render
}
}
}
});
@ -846,11 +1170,131 @@ function addLineNumbers(block) {
// Socket event for setting the chat background
socket.on("set_background", (data) => {
const chat = document.getElementById("chat");
chat.style.backgroundImage = `url('data:image/png;base64,${data.image_data}')`;
chat.style.backgroundRepeat = "no-repeat";
chat.style.backgroundPosition = "right center";
chat.style.backgroundSize = "auto"; // Ensures the image is not stretched
// Use setTimeout to ensure background updates don't get blocked by TTS
setTimeout(() => {
const chat = document.getElementById("chat");
chat.style.backgroundImage = `url('data:image/png;base64,${data.image_data}')`;
chat.style.backgroundRepeat = "no-repeat";
chat.style.backgroundPosition = "right center";
chat.style.backgroundSize = "auto"; // Ensures the image is not stretched
console.log("Background image updated");
}, 0);
});
// Activity management functions
function refreshActivityList() {
fetch('/api/activities')
.then(response => response.json())
.then(data => {
const activitySelect = document.getElementById('activity-select');
const activitySelectMobile = document.getElementById('activity-select-mobile');
// Clear existing options except the first one for desktop
while (activitySelect.options.length > 1) {
activitySelect.remove(1);
}
// Clear existing options except the first one for mobile
while (activitySelectMobile.options.length > 1) {
activitySelectMobile.remove(1);
}
// Add activities to both dropdowns
data.activities.forEach(activity => {
const option = document.createElement('option');
option.value = activity;
option.textContent = activity;
activitySelect.appendChild(option);
const optionMobile = document.createElement('option');
optionMobile.value = activity;
optionMobile.textContent = activity;
activitySelectMobile.appendChild(optionMobile);
});
})
.catch(error => {
console.error('Error fetching activities:', error);
alert('Failed to fetch activities');
});
}
function loadSelectedActivity() {
const activitySelect = document.getElementById('activity-select');
const selectedActivity = activitySelect.value;
if (!selectedActivity) {
alert('Please select an activity');
return;
}
// Send command to load activity
socket.emit("chat_message", {
"username": username,
"message": `/activity ${selectedActivity}`,
"model": document.getElementById("model-select").value,
"room_name": room_name
});
}
function loadSelectedActivityMobile() {
const activitySelectMobile = document.getElementById('activity-select-mobile');
const selectedActivity = activitySelectMobile.value;
if (!selectedActivity) {
alert('Please select an activity');
return;
}
// Send command to load activity
socket.emit("chat_message", {
"username": username,
"message": `/activity ${selectedActivity}`,
"model": document.getElementById("model-select").value,
"room_name": room_name
});
}
function cancelActivity() {
if (confirm('Are you sure you want to cancel the current activity?')) {
socket.emit("chat_message", {
"username": username,
"message": "/activity cancel",
"model": document.getElementById("model-select").value,
"room_name": room_name
});
}
}
// Socket event for activity status updates
socket.on("activity_status", (data) => {
const currentActivityInfo = document.getElementById('current-activity-info');
const activityListSection = document.getElementById('activity-list-section');
const currentActivityName = document.getElementById('current-activity-name');
const currentActivityInfoMobile = document.getElementById('current-activity-info-mobile');
const activityListSectionMobile = document.getElementById('activity-list-section-mobile');
const currentActivityNameMobile = document.getElementById('current-activity-name-mobile');
if (data.active) {
currentActivityInfo.style.display = 'block';
activityListSection.style.display = 'none';
currentActivityName.textContent = data.activity_name || 'Unknown';
currentActivityInfoMobile.style.display = 'block';
activityListSectionMobile.style.display = 'none';
currentActivityNameMobile.textContent = data.activity_name || 'Unknown';
} else {
currentActivityInfo.style.display = 'none';
activityListSection.style.display = 'block';
currentActivityInfoMobile.style.display = 'none';
activityListSectionMobile.style.display = 'block';
}
});
// Load activities on page load
document.addEventListener('DOMContentLoaded', () => {
refreshActivityList();
// Request current activity status
socket.emit("get_activity_status", {"room_name": room_name});
});
</script>

5
templates/index.html
View file

@ -6,6 +6,7 @@
<title>Chatroom</title>
<link rel="icon" href="{{ url_for('static', filename='favicon.ico') }}">
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.min.js"></script>
<script src="{{ url_for('static', filename='js/utils.js') }}"></script>
<style>
body {
font-family: Arial, sans-serif;
@ -52,10 +53,6 @@
</div>
<script>
function slugify(str) {
return str.toLowerCase().replace(/\s+/g, '-').replace(/[^\w-]+/g, '');
}
document.getElementById('join-room-form').addEventListener('submit', function(e) {
e.preventDefault();
const username = document.getElementById('username').value;

12
tests/functional/test_activity_flows.py
View file

@ -379,7 +379,9 @@ class TestRealActivityFiles(unittest.TestCase):
mock_get_client.return_value = (self.mock_client, "test-model")
# Load actual activity3.yaml
activity_file = Path(__file__).parent.parent.parent / "research" / "activity3.yaml"
activity_file = (
Path(__file__).parent.parent.parent / "research" / "activity3.yaml"
)
activity = guarded_ai.load_yaml_activity(str(activity_file))
# Should have section_5 as the terminal section
@ -408,7 +410,9 @@ class TestRealActivityFiles(unittest.TestCase):
mock_get_client.return_value = (self.mock_client, "test-model")
activity_file = (
Path(__file__).parent.parent.parent / "research" / "activity17-choose-adventure.yaml"
Path(__file__).parent.parent.parent
/ "research"
/ "activity17-choose-adventure.yaml"
)
activity = guarded_ai.load_yaml_activity(str(activity_file))
@ -450,7 +454,9 @@ class TestRealActivityFiles(unittest.TestCase):
mock_get_client.return_value = (self.mock_client, "test-model")
activity_file = (
Path(__file__).parent.parent.parent / "research" / "activity20-n-plus-1.yaml"
Path(__file__).parent.parent.parent
/ "research"
/ "activity20-n-plus-1.yaml"
)
activity = guarded_ai.load_yaml_activity(str(activity_file))

14
tests/functional/test_battleship_pre_script.py
View file

@ -28,7 +28,9 @@ class TestBattleshipPreScript(unittest.TestCase):
def test_battleship_yaml_has_pre_script(self):
"""Test that battleship YAML loads and has pre_script"""
activity_file = (
Path(__file__).parent.parent.parent / "research" / "activity29-battleship.yaml"
Path(__file__).parent.parent.parent
/ "research"
/ "activity29-battleship.yaml"
)
activity = guarded_ai.load_yaml_activity(str(activity_file))
@ -57,7 +59,9 @@ class TestBattleshipPreScript(unittest.TestCase):
def test_battleship_pre_script_execution_simulation(self):
"""Test simulated battleship pre_script execution"""
activity_file = (
Path(__file__).parent.parent.parent / "research" / "activity29-battleship.yaml"
Path(__file__).parent.parent.parent
/ "research"
/ "activity29-battleship.yaml"
)
activity = guarded_ai.load_yaml_activity(str(activity_file))
@ -96,7 +100,11 @@ class TestBattleshipPreScript(unittest.TestCase):
def test_testship_yaml_has_pre_script(self):
"""Test that testship YAML also has pre_script"""
activity_file = Path(__file__).parent.parent.parent / "research" / "activity29-testship.yaml"
activity_file = (
Path(__file__).parent.parent.parent
/ "research"
/ "activity29-testship.yaml"
)
activity = guarded_ai.load_yaml_activity(str(activity_file))
# Should also have pre_script (same structure as battleship)

12
tests/functional/test_guarded_ai.py
View file

@ -320,7 +320,9 @@ class TestActivityYAMLChanges(unittest.TestCase):
"""Test that activity3's new terminal section loads correctly"""
import guarded_ai as guarded_ai
activity_file = Path(__file__).parent.parent.parent / "research" / "activity3.yaml"
activity_file = (
Path(__file__).parent.parent.parent / "research" / "activity3.yaml"
)
activity = guarded_ai.load_yaml_activity(str(activity_file))
# Should have section_5 now
@ -348,7 +350,9 @@ class TestActivityYAMLChanges(unittest.TestCase):
import guarded_ai as guarded_ai
activity_file = (
Path(__file__).parent.parent.parent / "research" / "activity17-choose-adventure.yaml"
Path(__file__).parent.parent.parent
/ "research"
/ "activity17-choose-adventure.yaml"
)
activity = guarded_ai.load_yaml_activity(str(activity_file))
@ -375,7 +379,9 @@ class TestActivityYAMLChanges(unittest.TestCase):
"activity29-battleship.yaml",
"activity29-testship.yaml",
]:
activity_file = Path(__file__).parent.parent.parent / "research" / battleship_file
activity_file = (
Path(__file__).parent.parent.parent / "research" / battleship_file
)
activity = guarded_ai.load_yaml_activity(str(activity_file))
# Find exit transitions and verify they go to step_4

541
tests/functional/test_streaming_protocol.py Normal file
View file

@ -0,0 +1,541 @@
#!/usr/bin/env python3
"""
Functional tests for streaming message protocol
Tests the critical streaming functionality that sends real-time messages
via websockets, including the new protocol that separates username/model
from content for cleaner TTS processing.
"""
import unittest
import tempfile
import json
import sys
import threading
import time
from unittest.mock import Mock, patch, MagicMock, call
from pathlib import Path
from queue import Queue
# Add parent directory to path to import the app
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
class StreamingProtocolTest(unittest.TestCase):
"""Test streaming message protocol and websocket emissions"""
def setUp(self):
"""Set up test fixtures with mocked dependencies"""
self.username = "testuser"
self.room_name = "test_room"
self.model_name = "test-model-v1"
self.test_content = ["Hello", " world", "!", " How", " are", " you?"]
# Mock external dependencies
self.mock_socketio = MagicMock()
self.mock_db = MagicMock()
self.mock_room = MagicMock()
self.mock_room.name = self.room_name
# Track emitted messages
self.emitted_messages = []
self.mock_socketio.emit.side_effect = self._capture_emit
def _capture_emit(self, event_type, data, **kwargs):
"""Capture socketio.emit calls for verification"""
self.emitted_messages.append(
{"event": event_type, "data": data, "kwargs": kwargs}
)
def test_openai_streaming_protocol(self):
"""Test OpenAI/GPT streaming with new protocol format"""
# Mock OpenAI streaming response
mock_chunks = []
for i, content in enumerate(self.test_content):
chunk = MagicMock()
chunk.choices = [MagicMock()]
chunk.choices[0].delta.content = content
mock_chunks.append(chunk)
mock_client = MagicMock()
mock_completion = MagicMock()
mock_completion.__iter__.return_value = iter(mock_chunks)
mock_client.chat.completions.create.return_value = mock_completion
# Import and patch app with mocks
with patch.dict(
"sys.modules",
{
"gevent": MagicMock(),
"flask_socketio": self.mock_socketio,
"boto3": MagicMock(),
"openai": MagicMock(),
"together": MagicMock(),
"models": MagicMock(),
},
):
# Mock environment variables to avoid startup error
with patch.dict(
"os.environ",
{
"MODEL_ENDPOINT_0": "https://test.api.com",
"MODEL_API_KEY_0": "test-key",
},
):
import app
# Mock message creation
mock_message = MagicMock()
mock_message.id = 123
mock_message.content = ""
# Mock database and room operations
with patch.object(app.db.session, "add"), patch.object(
app.db.session, "commit"
), patch.object(app.db.session, "query") as mock_query, patch.object(
app, "get_room", return_value=self.mock_room
), patch.object(
app,
"get_openai_client_and_model",
return_value=(mock_client, self.model_name),
), patch.object(
app, "socketio", self.mock_socketio
), patch(
"app.Message", return_value=mock_message
):
mock_query.return_value.filter.return_value.one_or_none.return_value = (
mock_message
)
# Execute the streaming function
app.chat_gpt(self.username, self.room_name, self.model_name)
# Update content to simulate accumulation
mock_message.content = "".join(self.test_content)
# Verify the streaming protocol
message_chunks = [
msg for msg in self.emitted_messages if msg["event"] == "message_chunk"
]
# Should have one chunk per content piece plus completion signal
expected_chunks = len(self.test_content) + 1 # +1 for completion
self.assertEqual(len(message_chunks), expected_chunks)
# First chunk should have the new protocol format
first_chunk = message_chunks[0]
self.assertEqual(first_chunk["data"]["content"], self.test_content[0])
self.assertEqual(first_chunk["data"]["username"], self.username)
self.assertEqual(first_chunk["data"]["model_name"], self.model_name)
self.assertTrue(first_chunk["data"]["is_first_chunk"])
self.assertEqual(first_chunk["data"]["id"], 123)
# Subsequent content chunks should be simple format
for i in range(1, len(self.test_content)):
chunk = message_chunks[i]
self.assertEqual(chunk["data"]["content"], self.test_content[i])
self.assertEqual(chunk["data"]["id"], 123)
# Should not have username/model in subsequent chunks
self.assertNotIn("username", chunk["data"])
self.assertNotIn("model_name", chunk["data"])
self.assertNotIn("is_first_chunk", chunk["data"])
# Final chunk should be completion signal
completion_chunk = message_chunks[-1]
self.assertEqual(completion_chunk["data"]["content"], "")
self.assertTrue(completion_chunk["data"]["is_complete"])
self.assertEqual(completion_chunk["data"]["id"], 123)
def test_bedrock_streaming_protocol(self):
"""Test AWS Bedrock/Claude streaming with new protocol"""
# Mock Bedrock streaming response
mock_events = []
for content in self.test_content:
event = {
"chunk": {
"bytes": json.dumps(
{
"type": "content_block_delta",
"delta": {"type": "text_delta", "text": content},
}
).encode()
}
}
mock_events.append(event)
mock_client = MagicMock()
mock_response = {"body": iter(mock_events)}
mock_client.invoke_model_with_response_stream.return_value = mock_response
# Import and test
with patch.dict(
"sys.modules",
{
"gevent": MagicMock(),
"flask_socketio": self.mock_socketio,
"boto3": MagicMock(),
"openai": MagicMock(),
"together": MagicMock(),
"models": MagicMock(),
},
):
with patch.dict(
"os.environ",
{
"MODEL_ENDPOINT_0": "https://test.api.com",
"MODEL_API_KEY_0": "test-key",
},
):
import app
# Mock message creation
mock_message = MagicMock()
mock_message.id = 456
mock_message.content = ""
with patch.object(app.db.session, "add"), patch.object(
app.db.session, "commit"
), patch.object(app.db.session, "query") as mock_query, patch.object(
app, "get_room", return_value=self.mock_room
), patch.object(
app, "get_s3_client", return_value=mock_client
), patch.object(
app, "socketio", self.mock_socketio
), patch(
"app.Message", return_value=mock_message
):
mock_query.return_value.filter.return_value.one_or_none.return_value = (
mock_message
)
# Execute Bedrock streaming
app.chat_claude(self.username, self.room_name, self.model_name)
# Update content to simulate accumulation
mock_message.content = "".join(self.test_content)
# Verify Bedrock streaming protocol
message_chunks = [
msg for msg in self.emitted_messages if msg["event"] == "message_chunk"
]
# Should have content chunks plus completion
expected_chunks = len(self.test_content) + 1
self.assertEqual(len(message_chunks), expected_chunks)
# First chunk verification
first_chunk = message_chunks[0]
self.assertEqual(first_chunk["data"]["username"], self.username)
self.assertEqual(first_chunk["data"]["model_name"], self.model_name)
self.assertTrue(first_chunk["data"]["is_first_chunk"])
def test_llama_streaming_protocol(self):
"""Test Llama.cpp streaming with new protocol"""
# Mock Llama streaming response
mock_chunks = []
for content in self.test_content:
chunk = {"choices": [{"delta": {"content": content}}]}
mock_chunks.append(chunk)
mock_model = MagicMock()
mock_model.create_chat_completion.return_value = iter(mock_chunks)
with patch.dict(
"sys.modules",
{
"gevent": MagicMock(),
"flask_socketio": self.mock_socketio,
"boto3": MagicMock(),
"openai": MagicMock(),
"together": MagicMock(),
"models": MagicMock(),
"llama_cpp": MagicMock(),
},
):
with patch.dict(
"os.environ",
{
"MODEL_ENDPOINT_0": "https://test.api.com",
"MODEL_API_KEY_0": "test-key",
},
):
import app
# Mock message creation
mock_message = MagicMock()
mock_message.id = 789
mock_message.content = ""
with patch.object(app.db.session, "add"), patch.object(
app.db.session, "commit"
), patch.object(app.db.session, "query") as mock_query, patch.object(
app, "get_room", return_value=self.mock_room
), patch.object(
app, "socketio", self.mock_socketio
), patch(
"app.Message", return_value=mock_message
):
mock_query.return_value.filter.return_value.one_or_none.return_value = (
mock_message
)
# Mock llama_cpp model loading
with patch("llama_cpp.Llama", return_value=mock_model):
app.chat_llama(self.username, self.room_name, self.model_name)
# Update content to simulate accumulation
mock_message.content = "".join(self.test_content)
# Verify Llama streaming protocol
message_chunks = [
msg for msg in self.emitted_messages if msg["event"] == "message_chunk"
]
# Verify protocol consistency across all models
self.assertGreater(len(message_chunks), 0)
first_chunk = message_chunks[0]
self.assertEqual(first_chunk["data"]["username"], self.username)
self.assertEqual(first_chunk["data"]["model_name"], self.model_name)
self.assertTrue(first_chunk["data"]["is_first_chunk"])
def test_streaming_protocol_backwards_compatibility(self):
"""Test that the new protocol maintains expected behavior"""
# Mock a simple streaming scenario
content_chunks = ["Hello", " there!"]
mock_chunks = []
for content in content_chunks:
chunk = MagicMock()
chunk.choices = [MagicMock()]
chunk.choices[0].delta.content = content
mock_chunks.append(chunk)
mock_client = MagicMock()
mock_completion = MagicMock()
mock_completion.__iter__.return_value = iter(mock_chunks)
mock_client.chat.completions.create.return_value = mock_completion
with patch.dict(
"sys.modules",
{
"gevent": MagicMock(),
"flask_socketio": self.mock_socketio,
"boto3": MagicMock(),
"openai": MagicMock(),
"together": MagicMock(),
"models": MagicMock(),
},
):
with patch.dict(
"os.environ",
{
"MODEL_ENDPOINT_0": "https://test.api.com",
"MODEL_API_KEY_0": "test-key",
},
):
import app
with patch.object(app.db.session, "add"), patch.object(
app.db.session, "commit"
), patch.object(app.db.session, "query") as mock_query, patch.object(
app, "get_room", return_value=self.mock_room
), patch.object(
app,
"get_openai_client_and_model",
return_value=(mock_client, self.model_name),
), patch.object(
app, "socketio", self.mock_socketio
):
mock_message = MagicMock()
mock_message.id = 999
mock_query.return_value.filter.return_value.one_or_none.return_value = (
None
)
app.Message.return_value = mock_message
app.chat_gpt(self.username, self.room_name, self.model_name)
# Verify key properties of the new protocol
message_chunks = [
msg for msg in self.emitted_messages if msg["event"] == "message_chunk"
]
# All chunks should have an ID
for chunk in message_chunks:
self.assertIn("id", chunk["data"])
self.assertEqual(chunk["data"]["id"], 999)
# First chunk should have metadata fields
first_chunk = message_chunks[0]
required_first_chunk_fields = [
"id",
"content",
"username",
"model_name",
"is_first_chunk",
]
for field in required_first_chunk_fields:
self.assertIn(
field, first_chunk["data"], f"Missing required field: {field}"
)
# Content chunks should be minimal
for i in range(1, len(content_chunks)):
chunk = message_chunks[i]
# Should only have id and content
self.assertEqual(set(chunk["data"].keys()), {"id", "content"})
# Completion chunk should have is_complete
completion_chunk = message_chunks[-1]
self.assertTrue(completion_chunk["data"].get("is_complete", False))
def test_streaming_content_accumulation(self):
"""Test that streaming content is properly accumulated"""
test_chunks = ["The", " quick", " brown", " fox"]
expected_full_content = "".join(test_chunks)
mock_chunks = []
for content in test_chunks:
chunk = MagicMock()
chunk.choices = [MagicMock()]
chunk.choices[0].delta.content = content
mock_chunks.append(chunk)
mock_client = MagicMock()
mock_completion = MagicMock()
mock_completion.__iter__.return_value = iter(mock_chunks)
mock_client.chat.completions.create.return_value = mock_completion
with patch.dict(
"sys.modules",
{
"gevent": MagicMock(),
"flask_socketio": self.mock_socketio,
"boto3": MagicMock(),
"openai": MagicMock(),
"together": MagicMock(),
"models": MagicMock(),
},
):
with patch.dict(
"os.environ",
{
"MODEL_ENDPOINT_0": "https://test.api.com",
"MODEL_API_KEY_0": "test-key",
},
):
import app
mock_message = MagicMock()
mock_message.id = 555
mock_message.content = ""
with patch.object(app.db.session, "add"), patch.object(
app.db.session, "commit"
), patch.object(app.db.session, "query") as mock_query, patch.object(
app, "get_room", return_value=self.mock_room
), patch.object(
app,
"get_openai_client_and_model",
return_value=(mock_client, self.model_name),
), patch.object(
app, "socketio", self.mock_socketio
):
mock_query.return_value.filter.return_value.one_or_none.return_value = (
None
)
app.Message.return_value = mock_message
app.chat_gpt(self.username, self.room_name, self.model_name)
# Verify that content was properly accumulated in the database
# The message content should be the full accumulated text
self.assertEqual(mock_message.content, expected_full_content)
# Verify individual chunks were sent correctly
message_chunks = [
msg
for msg in self.emitted_messages
if msg["event"] == "message_chunk" and msg["data"].get("content")
]
# Each chunk should contain its piece of content
for i, chunk in enumerate(message_chunks[:-1]): # Exclude completion chunk
if i < len(test_chunks):
self.assertEqual(chunk["data"]["content"], test_chunks[i])
def test_error_handling_in_streaming(self):
"""Test error handling during streaming operations"""
mock_client = MagicMock()
mock_client.chat.completions.create.side_effect = Exception("API Error")
with patch.dict(
"sys.modules",
{
"gevent": MagicMock(),
"flask_socketio": self.mock_socketio,
"boto3": MagicMock(),
"openai": MagicMock(),
"together": MagicMock(),
"models": MagicMock(),
},
):
with patch.dict(
"os.environ",
{
"MODEL_ENDPOINT_0": "https://test.api.com",
"MODEL_API_KEY_0": "test-key",
},
):
import app
with patch.object(app.db.session, "add"), patch.object(
app.db.session, "commit"
), patch.object(app.db.session, "query") as mock_query, patch.object(
app, "get_room", return_value=self.mock_room
), patch.object(
app,
"get_openai_client_and_model",
return_value=(mock_client, self.model_name),
), patch.object(
app, "socketio", self.mock_socketio
):
mock_message = MagicMock()
mock_message.id = 444
mock_query.return_value.filter.return_value.one_or_none.return_value = (
None
)
app.Message.return_value = mock_message
# Should not raise exception, should handle gracefully
try:
app.chat_gpt(self.username, self.room_name, self.model_name)
except Exception as e:
self.fail(
f"Streaming should handle errors gracefully, but got: {e}"
)
# Should still send completion signal even after error
completion_chunks = [
msg
for msg in self.emitted_messages
if msg["event"] == "message_chunk"
and msg["data"].get("is_complete")
]
self.assertEqual(len(completion_chunks), 1)
if __name__ == "__main__":
unittest.main(verbosity=2)

141
tests/unit/test_activity_yaml_validator.py
View file

@ -606,6 +606,147 @@ sections:
# Should catch the YAML syntax error we know is in there
self.assertTrue(any("YAML syntax error" in error for error in errors))
def test_feedback_prompts_validation(self):
"""Test validation of feedback_prompts structure"""
valid_feedback_prompts = """
sections:
- section_id: "section_1"
title: "Test"
steps:
- step_id: "step_1"
title: "Test Step"
question: "Test?"
feedback_prompts:
- name: "hit_miss"
tokens_for_ai: "Report hit/miss for both players"
- name: "ship_sinking"
tokens_for_ai: "Report any ship sinking events"
buckets:
- test
transitions:
test:
next_section_and_step: "section_1:step_2"
- step_id: "step_2"
title: "Final"
content_blocks:
- "Done"
"""
temp_file = self.create_temp_yaml(valid_feedback_prompts)
try:
is_valid, errors, warnings = self.validator.validate_file(temp_file)
self.assertTrue(is_valid)
self.assertEqual(len(errors), 0)
finally:
os.unlink(temp_file)
def test_invalid_feedback_prompts(self):
"""Test validation of invalid feedback_prompts structure"""
invalid_feedback_prompts = """
sections:
- section_id: "section_1"
title: "Test"
steps:
- step_id: "step_1"
title: "Test Step"
question: "Test?"
feedback_prompts: "should_be_list"
buckets:
- test
transitions:
test:
next_section_and_step: "section_1:step_2"
- step_id: "step_2"
title: "Test Step 2"
question: "Another test?"
feedback_prompts: [] # Empty list should error
buckets:
- test2
transitions:
test2:
next_section_and_step: "section_1:step_3"
- step_id: "step_3"
title: "Test Step 3"
question: "Third test?"
feedback_prompts:
- "should_be_dict"
- name: "valid_name"
# Missing tokens_for_ai
- name: "duplicate"
tokens_for_ai: "First prompt"
- name: "duplicate" # Duplicate name
tokens_for_ai: "Second prompt"
- name: 123 # Invalid name type
tokens_for_ai: "Valid tokens"
- name: "valid_name2"
tokens_for_ai: 456 # Invalid tokens type
buckets:
- test3
transitions:
test3:
content_blocks: ["Done"]
"""
temp_file = self.create_temp_yaml(invalid_feedback_prompts)
try:
is_valid, errors, warnings = self.validator.validate_file(temp_file)
self.assertFalse(is_valid)
# Check for specific error types
self.assertTrue(
any("feedback_prompts' must be a list" in error for error in errors)
)
self.assertTrue(
any("feedback_prompts' cannot be empty" in error for error in errors)
)
self.assertTrue(any("must be a dictionary" in error for error in errors))
self.assertTrue(any("missing required field" in error for error in errors))
self.assertTrue(
any("duplicate feedback prompt name" in error for error in errors)
)
self.assertTrue(any("name must be a string" in error for error in errors))
self.assertTrue(
any("tokens_for_ai must be a string" in error for error in errors)
)
finally:
os.unlink(temp_file)
def test_both_feedback_systems(self):
"""Test that both feedback_tokens_for_ai and feedback_prompts can be used together"""
both_feedback_systems = """
sections:
- section_id: "section_1"
title: "Test"
steps:
- step_id: "step_1"
title: "Test Step"
question: "Test?"
feedback_tokens_for_ai: "Legacy feedback system"
feedback_prompts:
- name: "new_system_1"
tokens_for_ai: "New system prompt 1"
- name: "new_system_2"
tokens_for_ai: "New system prompt 2"
buckets:
- test
transitions:
test:
next_section_and_step: "section_1:step_2"
- step_id: "step_2"
title: "Final"
content_blocks:
- "Done"
"""
temp_file = self.create_temp_yaml(both_feedback_systems)
try:
is_valid, errors, warnings = self.validator.validate_file(temp_file)
self.assertTrue(is_valid, f"Should be valid but got errors: {errors}")
self.assertEqual(len(errors), 0)
finally:
os.unlink(temp_file)
def test_cli_integration(self):
"""Test the command line interface"""
import subprocess

987
tests/unit/test_app_feedback.py Normal file
View file

@ -0,0 +1,987 @@
#!/usr/bin/env python3
"""
Unit tests for app.py feedback functions.
Tests the feedback generation functions including:
- Legacy provide_feedback function
- New provide_feedback_prompts function
- Both systems integration
- Metadata filtering
- Language handling
"""
import unittest
from unittest.mock import patch, MagicMock, call
import sys
import json
from pathlib import Path
# Add parent directory to path to import app functions
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
class TestAppFeedback(unittest.TestCase):
"""Test cases for app.py feedback functions"""
def setUp(self):
"""Set up test fixtures"""
self.sample_transition = {
"ai_feedback": {"tokens_for_ai": "Additional transition instructions"},
"metadata_feedback_filter": ["shot_location", "hit_result", "ship_sunk"],
}
self.sample_metadata = {
"shot_location": "A5",
"hit_result": "hit",
"ship_sunk": "destroyer",
"private_info": "should_be_filtered",
"player_health": 100,
}
self.sample_new_metadata = {"new_shot": "B3", "new_result": "miss"}
def test_provide_feedback_import(self):
"""Test that we can import the provide_feedback function"""
try:
from app import provide_feedback
self.assertTrue(callable(provide_feedback))
except ImportError as e:
self.fail(f"Could not import provide_feedback: {e}")
def test_provide_feedback_prompts_import(self):
"""Test that we can import the provide_feedback_prompts function"""
try:
from app import provide_feedback_prompts
self.assertTrue(callable(provide_feedback_prompts))
except ImportError as e:
self.fail(f"Could not import provide_feedback_prompts: {e}")
@patch("app.get_openai_client_and_model")
def test_provide_feedback_legacy(self, mock_get_client):
"""Test legacy provide_feedback function"""
# Import here to avoid issues if module is not available
try:
from app import provide_feedback
except ImportError:
self.skipTest("app module not available for testing")
# Setup mock
mock_client = MagicMock()
mock_completion = MagicMock()
mock_completion.choices[0].message.content = "Great shot! You hit the target."
mock_client.chat.completions.create.return_value = mock_completion
mock_get_client.return_value = (mock_client, "test-model")
# Test data
transition = self.sample_transition
category = "hit"
question = "Where do you want to shoot?"
feedback_tokens_for_ai = "Provide battleship feedback"
user_response = "A5"
user_language = "English"
username = "testuser"
json_metadata = json.dumps(self.sample_metadata)
json_new_metadata = json.dumps(self.sample_new_metadata)
# Call function
feedback = provide_feedback(
transition,
category,
question,
feedback_tokens_for_ai,
user_response,
user_language,
username,
json_metadata,
json_new_metadata,
)
# Verify result
self.assertIn("Great shot! You hit the target.", feedback)
# Verify client was called
mock_client.chat.completions.create.assert_called_once()
call_args = mock_client.chat.completions.create.call_args[1]
# Check that system message includes language and transition instructions
system_message = call_args["messages"][0]["content"]
self.assertIn("English", system_message)
self.assertIn("Additional transition instructions", system_message)
@patch("app.get_openai_client_and_model")
def test_provide_feedback_prompts_multi(self, mock_get_client):
"""Test provide_feedback_prompts with multiple prompts"""
try:
from app import provide_feedback_prompts
except ImportError:
self.skipTest("app module not available for testing")
# Setup mock to return different responses for each prompt
mock_client = MagicMock()
mock_completion_1 = MagicMock()
mock_completion_1.choices[0].message.content = (
"Your shot at A5 was a hit! Enemy shot at B3 missed."
)
mock_completion_2 = MagicMock()
mock_completion_2.choices[0].message.content = (
"The enemy's destroyer has been sunk!"
)
mock_client.chat.completions.create.side_effect = [
mock_completion_1,
mock_completion_2,
]
mock_get_client.return_value = (mock_client, "test-model")
# Test data
transition = self.sample_transition
category = "valid_move"
question = "Where do you want to shoot?"
feedback_prompts = [
{
"name": "hit_miss_feedback",
"tokens_for_ai": "Report the hit/miss results for both players this turn",
},
{
"name": "ship_sinking_feedback",
"tokens_for_ai": "Report any ships that were sunk this turn",
},
]
user_response = "A5"
user_language = "English"
username = "testuser"
json_metadata = json.dumps(self.sample_metadata)
json_new_metadata = json.dumps(self.sample_new_metadata)
# Call function
feedback_messages = provide_feedback_prompts(
transition,
category,
question,
feedback_prompts,
user_response,
user_language,
username,
json_metadata,
json_new_metadata,
"",
)
# Verify results
self.assertEqual(len(feedback_messages), 2)
# Check first feedback message
self.assertEqual(feedback_messages[0]["name"], "hit_miss_feedback")
self.assertIn("Your shot at A5 was a hit", feedback_messages[0]["content"])
# Check second feedback message
self.assertEqual(feedback_messages[1]["name"], "ship_sinking_feedback")
self.assertIn("destroyer has been sunk", feedback_messages[1]["content"])
# Verify client was called twice
self.assertEqual(mock_client.chat.completions.create.call_count, 2)
@patch("app.get_openai_client_and_model")
def test_provide_feedback_with_filtered_metadata(self, mock_get_client):
"""Test that provide_feedback works correctly with pre-filtered metadata"""
try:
from app import provide_feedback
except ImportError:
self.skipTest("app module not available for testing")
# Setup mock
mock_client = MagicMock()
mock_completion = MagicMock()
mock_completion.choices[0].message.content = "Filtered feedback"
mock_client.chat.completions.create.return_value = mock_completion
mock_get_client.return_value = (mock_client, "test-model")
# Simulate app.py behavior: filter metadata before calling provide_feedback
filtered_metadata = {
k: v
for k, v in self.sample_metadata.items()
if k in self.sample_transition["metadata_feedback_filter"]
}
provide_feedback(
self.sample_transition,
"test",
"Question?",
"tokens",
"response",
"English",
"user",
json.dumps(filtered_metadata),
json.dumps({}),
)
# Check that user message contains only filtered metadata
call_args = mock_client.chat.completions.create.call_args[1]
user_message = call_args["messages"][1]["content"]
# Should contain filtered fields
self.assertIn("shot_location", user_message)
self.assertIn("hit_result", user_message)
self.assertIn("ship_sunk", user_message)
# Should NOT contain unfiltered fields (because we pre-filtered)
self.assertNotIn("private_info", user_message)
self.assertNotIn("player_health", user_message)
@patch("app.get_openai_client_and_model")
def test_provide_feedback_no_filter(self, mock_get_client):
"""Test feedback when no metadata filter is specified"""
try:
from app import provide_feedback
except ImportError:
self.skipTest("app module not available for testing")
# Setup mock
mock_client = MagicMock()
mock_completion = MagicMock()
mock_completion.choices[0].message.content = "Unfiltered feedback"
mock_client.chat.completions.create.return_value = mock_completion
mock_get_client.return_value = (mock_client, "test-model")
# Call function without metadata filter
transition = {
"ai_feedback": {"tokens_for_ai": "Generate feedback"}
} # No metadata_feedback_filter
provide_feedback(
transition,
"test",
"Question?",
"tokens",
"response",
"English",
"user",
json.dumps(self.sample_metadata),
json.dumps({}),
)
# Check that user message contains all metadata
call_args = mock_client.chat.completions.create.call_args[1]
user_message = call_args["messages"][1]["content"]
# Should contain all metadata fields when no filter is applied
self.assertIn("private_info", user_message)
self.assertIn("player_health", user_message)
@patch("app.get_openai_client_and_model")
def test_provide_feedback_error_handling(self, mock_get_client):
"""Test error handling in feedback functions"""
try:
from app import provide_feedback
except ImportError:
self.skipTest("app module not available for testing")
# Setup mock to raise exception
mock_client = MagicMock()
mock_client.chat.completions.create.side_effect = Exception("API Error")
mock_get_client.return_value = (mock_client, "test-model")
# Call function
feedback = provide_feedback(
{"ai_feedback": {"tokens_for_ai": "Generate feedback"}},
"test",
"Question?",
"tokens",
"response",
"English",
"user",
json.dumps({}),
json.dumps({}),
)
# Should handle error gracefully
self.assertIn("Error", feedback)
@patch("app.get_openai_client_and_model")
def test_provide_feedback_prompts_filter_empty(self, mock_get_client):
"""Test feedback_prompts with empty results filtered out"""
try:
from app import provide_feedback_prompts
except ImportError:
self.skipTest("app module not available for testing")
# Setup mock to return mixed results including empty
mock_client = MagicMock()
mock_completion_1 = MagicMock()
mock_completion_1.choices[0].message.content = "" # Empty result
mock_completion_2 = MagicMock()
mock_completion_2.choices[0].message.content = (
" " # Whitespace only (should be filtered)
)
mock_completion_3 = MagicMock()
mock_completion_3.choices[0].message.content = "Valid feedback" # Valid result
mock_client.chat.completions.create.side_effect = [
mock_completion_1,
mock_completion_2,
mock_completion_3,
]
mock_get_client.return_value = (mock_client, "test-model")
# Test data
feedback_prompts = [
{"name": "empty", "tokens_for_ai": "Empty prompt"},
{"name": "whitespace", "tokens_for_ai": "Whitespace prompt"},
{"name": "valid", "tokens_for_ai": "Valid prompt"},
]
feedback_messages = provide_feedback_prompts(
{},
"test",
"Question?",
feedback_prompts,
"response",
"English",
"user",
json.dumps({}),
json.dumps({}),
"",
)
# Should only return valid feedback (empty and whitespace filtered out)
self.assertEqual(len(feedback_messages), 1)
self.assertEqual(feedback_messages[0]["name"], "valid")
self.assertEqual(feedback_messages[0]["content"], "Valid feedback")
@patch("app.get_openai_client_and_model")
def test_provide_feedback_prompts_per_prompt_metadata_filtering(
self, mock_get_client
):
"""Test that each prompt gets its own filtered metadata"""
try:
from app import provide_feedback_prompts
except ImportError:
self.skipTest("app module not available for testing")
# Setup mock to return different responses
mock_client = MagicMock()
mock_completion_1 = MagicMock()
mock_completion_1.choices[0].message.content = (
"Shot feedback with hit/miss data"
)
mock_completion_2 = MagicMock()
mock_completion_2.choices[0].message.content = "Ship feedback with sinking data"
mock_client.chat.completions.create.side_effect = [
mock_completion_1,
mock_completion_2,
]
mock_get_client.return_value = (mock_client, "test-model")
# Test data with mixed metadata
full_metadata = {
"user_shot": "A5",
"user_hit_result": "hit",
"ai_shot": "B3",
"ai_hit_result": "miss",
"user_sunk_ship_this_round": "Destroyer",
"ai_sunk_ship_this_round": None,
"game_over": False,
"extra_field": "should_not_appear",
}
feedback_prompts = [
{
"name": "shot_report",
"tokens_for_ai": "Report hit/miss",
"metadata_filter": [
"user_shot",
"user_hit_result",
"ai_shot",
"ai_hit_result",
],
},
{
"name": "ship_status",
"tokens_for_ai": "Report ship sinking",
"metadata_filter": [
"user_sunk_ship_this_round",
"ai_sunk_ship_this_round",
],
},
]
feedback_messages = provide_feedback_prompts(
{},
"test",
"Question?",
feedback_prompts,
"response",
"English",
"user",
json.dumps(full_metadata),
json.dumps({}),
"",
)
# Verify both prompts got responses
self.assertEqual(len(feedback_messages), 2)
self.assertEqual(feedback_messages[0]["name"], "shot_report")
self.assertEqual(feedback_messages[1]["name"], "ship_status")
# Verify the first prompt only got shot-related metadata
first_call_args = mock_client.chat.completions.create.call_args_list[0][1]
first_user_message = first_call_args["messages"][1]["content"]
self.assertIn("user_shot", first_user_message)
self.assertIn("user_hit_result", first_user_message)
self.assertIn("ai_shot", first_user_message)
self.assertIn("ai_hit_result", first_user_message)
self.assertNotIn("user_sunk_ship_this_round", first_user_message)
self.assertNotIn("extra_field", first_user_message)
# Verify the second prompt only got ship-related metadata
second_call_args = mock_client.chat.completions.create.call_args_list[1][1]
second_user_message = second_call_args["messages"][1]["content"]
self.assertIn("user_sunk_ship_this_round", second_user_message)
self.assertIn("ai_sunk_ship_this_round", second_user_message)
self.assertNotIn("user_shot", second_user_message)
self.assertNotIn("extra_field", second_user_message)
@patch("app.get_openai_client_and_model")
def test_ship_status_metadata_filtering_debug(self, mock_get_client):
"""Debug test to check if Ship Status is getting only the right metadata"""
try:
from app import provide_feedback_prompts
except ImportError:
self.skipTest("app module not available for testing")
# Setup mock
mock_client = MagicMock()
mock_completion = MagicMock()
mock_completion.choices[0].message.content = "Test response"
mock_client.chat.completions.create.return_value = mock_completion
mock_get_client.return_value = (mock_client, "test-model")
# Test data mimicking the actual battleship scenario
full_metadata = {
"user_shot": "46", # This should NOT appear in Ship Status
"ai_shot": "49", # This should NOT appear in Ship Status
"user_hit_result": "hit",
"ai_hit_result": "miss",
"user_sunk_ship_this_round": "Destroyer", # This SHOULD appear
"ai_sunk_ship_this_round": None, # This SHOULD appear
"game_over": False,
"extra_stuff": "should not appear anywhere",
}
# Exact structure from battleship YAML
feedback_prompts = [
{
"name": "Shot Report",
"tokens_for_ai": "🎯 Report ONLY the hit/miss results",
"metadata_filter": [
"user_shot",
"ai_shot",
"user_hit_result",
"ai_hit_result",
],
},
{
"name": "Ship Status",
"tokens_for_ai": "You are the Ship Destruction Oracle",
"metadata_filter": [
"user_sunk_ship_this_round",
"ai_sunk_ship_this_round",
],
},
]
# Call the function
provide_feedback_prompts(
{},
"valid_move",
"Question?",
feedback_prompts,
"46",
"English",
"user",
json.dumps(full_metadata),
json.dumps({}),
"",
)
# Check what metadata each prompt actually received
self.assertEqual(mock_client.chat.completions.create.call_count, 2)
# First call should be Shot Report
shot_report_call = mock_client.chat.completions.create.call_args_list[0][1]
shot_report_metadata = shot_report_call["messages"][1]["content"]
print("=== SHOT REPORT METADATA ===")
print(shot_report_metadata)
# Shot Report should have shot data but NOT ship destruction data
self.assertIn("user_shot", shot_report_metadata)
self.assertIn("46", shot_report_metadata)
self.assertNotIn("user_sunk_ship_this_round", shot_report_metadata)
self.assertNotIn("Destroyer", shot_report_metadata)
# Second call should be Ship Status
ship_status_call = mock_client.chat.completions.create.call_args_list[1][1]
ship_status_metadata = ship_status_call["messages"][1]["content"]
print("=== SHIP STATUS METADATA ===")
print(ship_status_metadata)
# Ship Status should have ship destruction data but NOT shot data
self.assertIn("user_sunk_ship_this_round", ship_status_metadata)
self.assertIn("Destroyer", ship_status_metadata)
self.assertNotIn("user_shot", ship_status_metadata)
self.assertNotIn("46", ship_status_metadata)
self.assertNotIn("extra_stuff", ship_status_metadata)
def test_provide_feedback_prompts_language_injection(self):
"""Test that language instructions are properly added to prompts"""
try:
from app import provide_feedback_prompts
except ImportError:
self.skipTest("app module not available for testing")
with patch("app.get_openai_client_and_model") as mock_get_client:
mock_client = MagicMock()
mock_completion = MagicMock()
mock_completion.choices[0].message.content = "Feedback in Spanish"
mock_client.chat.completions.create.return_value = mock_completion
mock_get_client.return_value = (mock_client, "test-model")
feedback_prompts = [{"name": "test", "tokens_for_ai": "Base prompt"}]
# Test with Spanish language
provide_feedback_prompts(
{},
"test",
"Question?",
feedback_prompts,
"response",
"Spanish",
"user",
json.dumps({}),
json.dumps({}),
"",
)
# Check that system message includes Spanish language instruction
call_args = mock_client.chat.completions.create.call_args[1]
system_message = call_args["messages"][0]["content"]
self.assertIn("Spanish", system_message)
self.assertIn("Base prompt", system_message)
@patch("app.get_openai_client_and_model")
def test_provide_feedback_transition_tokens(self, mock_get_client):
"""Test that transition ai_feedback tokens are included"""
try:
from app import provide_feedback_prompts
except ImportError:
self.skipTest("app module not available for testing")
mock_client = MagicMock()
mock_completion = MagicMock()
mock_completion.choices[0].message.content = "Enhanced feedback"
mock_client.chat.completions.create.return_value = mock_completion
mock_get_client.return_value = (mock_client, "test-model")
transition = {
"ai_feedback": {"tokens_for_ai": "Be more dramatic in your feedback"}
}
feedback_prompts = [{"name": "test", "tokens_for_ai": "Base prompt"}]
provide_feedback_prompts(
transition,
"test",
"Question?",
feedback_prompts,
"response",
"English",
"user",
json.dumps({}),
json.dumps({}),
"",
)
# Check that system message includes both base and transition tokens
call_args = mock_client.chat.completions.create.call_args[1]
system_message = call_args["messages"][0]["content"]
self.assertIn("Base prompt", system_message)
self.assertIn("Be more dramatic in your feedback", system_message)
@patch("app.get_openai_client_and_model")
def test_user_response_filtering_with_metadata_filter(self, mock_get_client):
"""Test that user_response is filtered correctly using metadata_filter approach"""
try:
from app import provide_feedback_prompts
except ImportError:
self.skipTest("app module not available for testing")
# Setup mock
mock_client = MagicMock()
mock_completion = MagicMock()
mock_completion.choices[0].message.content = "Response for prompt"
mock_client.chat.completions.create.return_value = mock_completion
mock_get_client.return_value = (mock_client, "test-model")
# Test feedback prompts - one that includes user_response, one that doesn't
feedback_prompts = [
{
"name": "Shot Report",
"tokens_for_ai": "Report shot positions",
"metadata_filter": [
"user_shot",
"user_response",
], # Includes user_response
},
{
"name": "Ship Status",
"tokens_for_ai": "Report ship status",
"metadata_filter": ["ship_status"], # Does NOT include user_response
},
]
metadata = {"user_shot": "35", "ship_status": "intact"}
user_response = "I choose position 35"
provide_feedback_prompts(
{},
"valid_move",
"Choose position?",
feedback_prompts,
user_response,
"English",
"user",
json.dumps(metadata),
json.dumps({}),
"",
)
# Should have 2 calls
self.assertEqual(mock_client.chat.completions.create.call_count, 2)
# First call (Shot Report) should have user_response
first_call = mock_client.chat.completions.create.call_args_list[0][1]
first_user_message = first_call["messages"][1]["content"]
self.assertIn(
"I choose position 35", first_user_message
) # user_response should be present
# Second call (Ship Status) should NOT have user_response
second_call = mock_client.chat.completions.create.call_args_list[1][1]
second_user_message = second_call["messages"][1]["content"]
self.assertEqual(
second_user_message.count("I choose position 35"), 0
) # user_response should be empty/filtered
@patch("app.get_openai_client_and_model")
def test_skip_condition_all_null(self, mock_get_client):
"""Test skip_condition 'all_null' skips prompts when all metadata values are null"""
try:
from app import provide_feedback_prompts
except ImportError:
self.skipTest("app module not available for testing")
# Mock client (should not be called for skipped prompts)
mock_client = MagicMock()
mock_get_client.return_value = (mock_client, "test-model")
feedback_prompts = [
{
"name": "Ship Status",
"tokens_for_ai": "Report ship destruction",
"metadata_filter": ["user_sunk_ship", "ai_sunk_ship"],
"skip_condition": "all_null"
}
]
# Test with all null values - should skip
metadata_all_null = {
"user_sunk_ship": None,
"ai_sunk_ship": None
}
feedback_messages = provide_feedback_prompts(
{},
"test",
"Question?",
feedback_prompts,
"response",
"English",
"user",
json.dumps(metadata_all_null),
json.dumps({}),
""
)
# Should be empty (prompt was skipped)
self.assertEqual(len(feedback_messages), 0)
# Client should not have been called
self.assertEqual(mock_client.chat.completions.create.call_count, 0)
@patch("app.get_openai_client_and_model")
def test_skip_condition_all_null_with_values(self, mock_get_client):
"""Test skip_condition 'all_null' does NOT skip when values exist"""
try:
from app import provide_feedback_prompts
except ImportError:
self.skipTest("app module not available for testing")
# Mock client to return valid response
mock_client = MagicMock()
mock_completion = MagicMock()
mock_completion.choices[0].message.content = "Ship destroyed!"
mock_client.chat.completions.create.return_value = mock_completion
mock_get_client.return_value = (mock_client, "test-model")
feedback_prompts = [
{
"name": "Ship Status",
"tokens_for_ai": "Report ship destruction",
"metadata_filter": ["user_sunk_ship", "ai_sunk_ship"],
"skip_condition": "all_null"
}
]
# Test with actual values - should NOT skip
metadata_with_values = {
"user_sunk_ship": "Destroyer",
"ai_sunk_ship": None
}
feedback_messages = provide_feedback_prompts(
{},
"test",
"Question?",
feedback_prompts,
"response",
"English",
"user",
json.dumps(metadata_with_values),
json.dumps({}),
""
)
# Should have feedback (prompt was NOT skipped)
self.assertEqual(len(feedback_messages), 1)
self.assertEqual(feedback_messages[0]["name"], "Ship Status")
self.assertEqual(feedback_messages[0]["content"], "Ship destroyed!")
# Client should have been called
self.assertEqual(mock_client.chat.completions.create.call_count, 1)
@patch("app.get_openai_client_and_model")
def test_skip_condition_all_false(self, mock_get_client):
"""Test skip_condition 'all_false' skips when all metadata values are False"""
try:
from app import provide_feedback_prompts
except ImportError:
self.skipTest("app module not available for testing")
mock_client = MagicMock()
mock_get_client.return_value = (mock_client, "test-model")
feedback_prompts = [
{
"name": "Game Over",
"tokens_for_ai": "Report game over",
"metadata_filter": ["game_over", "user_wins", "ai_wins"],
"skip_condition": "all_false"
}
]
# Test with all false values - should skip
metadata_all_false = {
"game_over": False,
"user_wins": False,
"ai_wins": False
}
feedback_messages = provide_feedback_prompts(
{},
"test",
"Question?",
feedback_prompts,
"response",
"English",
"user",
json.dumps(metadata_all_false),
json.dumps({}),
""
)
# Should be empty (prompt was skipped)
self.assertEqual(len(feedback_messages), 0)
self.assertEqual(mock_client.chat.completions.create.call_count, 0)
@patch("app.get_openai_client_and_model")
def test_skip_condition_all_true(self, mock_get_client):
"""Test skip_condition 'all_true' skips when all metadata values are True"""
try:
from app import provide_feedback_prompts
except ImportError:
self.skipTest("app module not available for testing")
mock_client = MagicMock()
mock_get_client.return_value = (mock_client, "test-model")
feedback_prompts = [
{
"name": "All True Test",
"tokens_for_ai": "Test prompt",
"metadata_filter": ["flag1", "flag2", "flag3"],
"skip_condition": "all_true"
}
]
# Test with all true values - should skip
metadata_all_true = {
"flag1": True,
"flag2": True,
"flag3": True
}
feedback_messages = provide_feedback_prompts(
{},
"test",
"Question?",
feedback_prompts,
"response",
"English",
"user",
json.dumps(metadata_all_true),
json.dumps({}),
""
)
# Should be empty (prompt was skipped)
self.assertEqual(len(feedback_messages), 0)
self.assertEqual(mock_client.chat.completions.create.call_count, 0)
@patch("app.get_openai_client_and_model")
def test_skip_condition_mixed_values(self, mock_get_client):
"""Test skip_condition does NOT skip when values are mixed"""
try:
from app import provide_feedback_prompts
except ImportError:
self.skipTest("app module not available for testing")
# Mock client to return valid response
mock_client = MagicMock()
mock_completion = MagicMock()
mock_completion.choices[0].message.content = "Mixed values feedback"
mock_client.chat.completions.create.return_value = mock_completion
mock_get_client.return_value = (mock_client, "test-model")
feedback_prompts = [
{
"name": "Mixed Test",
"tokens_for_ai": "Mixed test prompt",
"metadata_filter": ["val1", "val2", "val3"],
"skip_condition": "all_false"
}
]
# Test with mixed values - should NOT skip
metadata_mixed = {
"val1": False,
"val2": True, # Mixed with False - should NOT skip
"val3": False
}
feedback_messages = provide_feedback_prompts(
{},
"test",
"Question?",
feedback_prompts,
"response",
"English",
"user",
json.dumps(metadata_mixed),
json.dumps({}),
""
)
# Should have feedback (prompt was NOT skipped due to mixed values)
self.assertEqual(len(feedback_messages), 1)
self.assertEqual(feedback_messages[0]["name"], "Mixed Test")
self.assertEqual(feedback_messages[0]["content"], "Mixed values feedback")
self.assertEqual(mock_client.chat.completions.create.call_count, 1)
@patch("app.get_openai_client_and_model")
def test_skip_condition_battleship_scenario(self, mock_get_client):
"""Test the real battleship scenario that was causing hallucinations"""
try:
from app import provide_feedback_prompts
except ImportError:
self.skipTest("app module not available for testing")
mock_client = MagicMock()
mock_get_client.return_value = (mock_client, "test-model")
feedback_prompts = [
{
"name": "Shot Report",
"tokens_for_ai": "Report shot results",
"metadata_filter": ["user_shot", "ai_shot", "user_hit_result", "ai_hit_result"]
# No skip condition - always runs
},
{
"name": "Ship Status",
"tokens_for_ai": "Report ship destruction",
"metadata_filter": ["user_sunk_ship_this_round", "ai_sunk_ship_this_round"],
"skip_condition": "all_null" # Skip when no ships sunk
},
{
"name": "Game Over",
"tokens_for_ai": "Report game over",
"metadata_filter": ["game_over", "user_wins", "ai_wins"],
"skip_condition": "all_false" # Skip when game not over
}
]
# Real scenario: shots taken, no ships sunk, game continues
real_battleship_metadata = {
"user_shot": 23,
"ai_shot": 46,
"user_hit_result": "hit",
"ai_hit_result": "hit",
"user_sunk_ship_this_round": None, # No ship sunk
"ai_sunk_ship_this_round": None, # No ship sunk
"game_over": False,
"user_wins": False,
"ai_wins": False
}
# Mock only Shot Report response (others should be skipped)
mock_completion = MagicMock()
mock_completion.choices[0].message.content = "🎯 Your shot at 23: hit! AI shot at 46: hit!"
mock_client.chat.completions.create.return_value = mock_completion
feedback_messages = provide_feedback_prompts(
{},
"test",
"Choose position",
feedback_prompts,
"23",
"English",
"user",
json.dumps(real_battleship_metadata),
json.dumps({}),
""
)
# Should only have Shot Report (other two skipped)
self.assertEqual(len(feedback_messages), 1)
self.assertEqual(feedback_messages[0]["name"], "Shot Report")
self.assertIn("🎯", feedback_messages[0]["content"])
# Only one API call should have been made (Ship Status and Game Over skipped)
self.assertEqual(mock_client.chat.completions.create.call_count, 1)
if __name__ == "__main__":
unittest.main(verbosity=2)

364
tests/unit/test_guarded_ai.py Normal file
View file

@ -0,0 +1,364 @@
#!/usr/bin/env python3
"""
Unit tests for the guarded_ai.py module.
Tests the core feedback generation functions including:
- Legacy single feedback system
- New multi-prompt feedback system
- Both systems together
- OpenAI client initialization
- Categorization and feedback generation
"""
import unittest
from unittest.mock import patch, MagicMock, call
import sys
from pathlib import Path
import json
# Add parent directory to path to import guarded_ai
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "research"))
from guarded_ai import (
provide_feedback,
provide_feedback_prompts,
categorize_response,
generate_ai_feedback,
get_openai_client_and_model,
initialize_model_map,
)
class TestGuardedAI(unittest.TestCase):
"""Test cases for guarded_ai functions"""
def setUp(self):
"""Set up test fixtures"""
self.sample_metadata = {
"player_health": 100,
"enemy_health": 80,
"user_shot": "A5",
"ai_shot": "B3",
"user_hit_result": "hit",
"ai_hit_result": "miss",
}
self.sample_transition = {
"ai_feedback": {
"tokens_for_ai": "Additional transition-specific instructions"
},
"metadata_feedback_filter": [
"user_shot",
"ai_shot",
"user_hit_result",
"ai_hit_result",
],
}
@patch("guarded_ai.get_openai_client_and_model")
def test_categorize_response(self, mock_get_client):
"""Test response categorization"""
# Setup mock
mock_client = MagicMock()
mock_completion = MagicMock()
mock_completion.choices[0].message.content = "correct_answer"
mock_client.chat.completions.create.return_value = mock_completion
mock_get_client.return_value = (mock_client, "test-model")
# Test categorization
question = "What is 2+2?"
response = "Four"
buckets = ["correct_answer", "wrong_answer"]
tokens_for_ai = "Categorize math answers"
category = categorize_response(question, response, buckets, tokens_for_ai)
# Verify result
self.assertEqual(category, "correct_answer")
# Verify client was called correctly
mock_client.chat.completions.create.assert_called_once()
call_args = mock_client.chat.completions.create.call_args[1]
self.assertEqual(call_args["model"], "test-model")
self.assertEqual(call_args["max_tokens"], 5)
self.assertEqual(call_args["temperature"], 0)
# Check message content
messages = call_args["messages"]
self.assertEqual(len(messages), 2)
self.assertIn("correct_answer, wrong_answer", messages[0]["content"])
@patch("guarded_ai.get_openai_client_and_model")
def test_generate_ai_feedback(self, mock_get_client):
"""Test AI feedback generation"""
# Setup mock
mock_client = MagicMock()
mock_completion = MagicMock()
mock_completion.choices[0].message.content = "Great job on the math!"
mock_client.chat.completions.create.return_value = mock_completion
mock_get_client.return_value = (mock_client, "test-model")
# Test feedback generation
category = "correct_answer"
question = "What is 2+2?"
user_response = "Four"
tokens_for_ai = "Provide encouraging feedback"
metadata = {"score": 100}
feedback = generate_ai_feedback(
category, question, user_response, tokens_for_ai, metadata
)
# Verify result
self.assertEqual(feedback, "Great job on the math!")
# Verify client was called correctly
mock_client.chat.completions.create.assert_called_once()
call_args = mock_client.chat.completions.create.call_args[1]
self.assertEqual(call_args["model"], "test-model")
self.assertEqual(call_args["max_tokens"], 250)
self.assertEqual(call_args["temperature"], 0.7)
@patch("guarded_ai.generate_ai_feedback")
def test_provide_feedback_legacy(self, mock_generate_feedback):
"""Test legacy single feedback system"""
mock_generate_feedback.return_value = "Good work! Try again."
# Test data
transition = self.sample_transition
category = "partial_understanding"
question = "What is the capital of France?"
user_response = "Paris is nice"
user_language = "English"
tokens_for_ai = "Provide geography feedback"
metadata = {"attempts": 1}
# Call function
feedback = provide_feedback(
transition,
category,
question,
user_response,
user_language,
tokens_for_ai,
metadata,
)
# Verify feedback was generated
self.assertIn("AI Feedback:", feedback)
self.assertIn("Good work! Try again.", feedback)
# Verify generate_ai_feedback was called with filtered metadata
mock_generate_feedback.assert_called_once()
call_args = mock_generate_feedback.call_args[0]
self.assertEqual(call_args[0], category) # category
self.assertEqual(call_args[1], question) # question
self.assertEqual(call_args[2], user_response) # user_response
# Check tokens_for_ai includes language and transition instructions
tokens_arg = call_args[3]
self.assertIn("English", tokens_arg)
self.assertIn("Additional transition-specific instructions", tokens_arg)
# Check metadata was filtered
filtered_metadata = call_args[4]
expected_filtered = {
k: v
for k, v in self.sample_metadata.items()
if k in transition["metadata_feedback_filter"]
}
# Since our test metadata doesn't have the filtered keys, it should be empty or contain only matching keys
# But the function should have passed what it received
@patch("guarded_ai.generate_ai_feedback")
def test_provide_feedback_prompts(self, mock_generate_feedback):
"""Test new multi-prompt feedback system"""
# Setup mock to return different feedback for each prompt
mock_generate_feedback.side_effect = [
"Hit at A5, miss at B3",
"No ships were sunk this round",
]
# Test data
transition = self.sample_transition
category = "valid_move"
question = "Where do you want to shoot?"
feedback_prompts = [
{
"name": "hit_miss",
"tokens_for_ai": "Report the hit/miss results for both players",
},
{
"name": "ship_sinking",
"tokens_for_ai": "Report any ships that were sunk",
},
]
user_response = "A5"
user_language = "English"
metadata = self.sample_metadata
# Call function
feedback_messages = provide_feedback_prompts(
transition,
category,
question,
feedback_prompts,
user_response,
user_language,
metadata,
"",
)
# Verify we got the expected number of feedback messages
self.assertEqual(len(feedback_messages), 2)
# Verify message structure
self.assertEqual(feedback_messages[0]["name"], "hit_miss")
self.assertEqual(feedback_messages[0]["content"], "Hit at A5, miss at B3")
self.assertEqual(feedback_messages[1]["name"], "ship_sinking")
self.assertEqual(
feedback_messages[1]["content"], "No ships were sunk this round"
)
# Verify generate_ai_feedback was called twice
self.assertEqual(mock_generate_feedback.call_count, 2)
@patch("guarded_ai.generate_ai_feedback")
def test_provide_feedback_prompts_empty_responses(self, mock_generate_feedback):
"""Test that empty feedback responses are filtered out"""
# Setup mock to return empty/whitespace responses
mock_generate_feedback.side_effect = [
"", # Empty response
" ", # Whitespace only
"Valid feedback", # Valid response
]
transition = {}
category = "test"
question = "Test?"
feedback_prompts = [
{"name": "empty", "tokens_for_ai": "Empty prompt"},
{"name": "whitespace", "tokens_for_ai": "Whitespace prompt"},
{"name": "valid", "tokens_for_ai": "Valid prompt"},
]
user_response = "Test response"
user_language = "English"
metadata = {}
feedback_messages = provide_feedback_prompts(
transition,
category,
question,
feedback_prompts,
user_response,
user_language,
metadata,
"",
)
# Should only return the valid feedback message
self.assertEqual(len(feedback_messages), 1)
self.assertEqual(feedback_messages[0]["name"], "valid")
self.assertEqual(feedback_messages[0]["content"], "Valid feedback")
def test_provide_feedback_no_ai_feedback_config(self):
"""Test legacy feedback when no ai_feedback config in transition"""
transition = {} # No ai_feedback key
category = "test"
question = "Test?"
user_response = "Response"
user_language = "English"
tokens_for_ai = "Base tokens"
metadata = {}
with patch("guarded_ai.generate_ai_feedback") as mock_generate:
mock_generate.return_value = "" # Should not be called
feedback = provide_feedback(
transition,
category,
question,
user_response,
user_language,
tokens_for_ai,
metadata,
)
# Should NOT call generate_ai_feedback when no ai_feedback in transition
mock_generate.assert_not_called()
self.assertEqual(feedback, "")
@patch.dict(
"os.environ",
{"MODEL_ENDPOINT_0": "http://test.com", "MODEL_API_KEY_0": "test-key"},
)
def test_initialize_model_map(self):
"""Test model map initialization from environment variables"""
with patch("guarded_ai.get_client_for_endpoint") as mock_get_client:
mock_client = MagicMock()
mock_get_client.return_value = mock_client
# Clear and reinitialize
import guarded_ai
guarded_ai.MODEL_CLIENT_MAP = {}
initialize_model_map()
# Verify client was created and stored
mock_get_client.assert_called_with("http://test.com", "test-key")
self.assertIn("endpoint_0", guarded_ai.MODEL_CLIENT_MAP)
self.assertEqual(guarded_ai.MODEL_CLIENT_MAP["endpoint_0"][0], mock_client)
def test_get_openai_client_and_model_default(self):
"""Test getting OpenAI client with default model"""
with patch("guarded_ai.MODEL_CLIENT_MAP", {}):
with patch("guarded_ai.get_client_for_endpoint") as mock_get_client:
mock_client = MagicMock()
mock_get_client.return_value = mock_client
client, model = get_openai_client_and_model()
# Should return default model name
self.assertEqual(model, "adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic")
self.assertEqual(client, mock_client)
def test_get_openai_client_and_model_from_map(self):
"""Test getting OpenAI client from model map"""
mock_client = MagicMock()
test_map = {"endpoint_0": (mock_client, "http://test.com")}
with patch("guarded_ai.MODEL_CLIENT_MAP", test_map):
client, model = get_openai_client_and_model("test-model")
# Should return client from map
self.assertEqual(client, mock_client)
self.assertEqual(model, "test-model")
@patch("guarded_ai.get_openai_client_and_model")
def test_categorize_response_error_handling(self, mock_get_client):
"""Test error handling in categorize_response"""
# Setup mock to raise exception
mock_client = MagicMock()
mock_client.chat.completions.create.side_effect = Exception("API Error")
mock_get_client.return_value = (mock_client, "test-model")
category = categorize_response("Test?", "Answer", ["bucket1"], "tokens")
# Should return error string
self.assertIn("Error:", category)
@patch("guarded_ai.get_openai_client_and_model")
def test_generate_ai_feedback_error_handling(self, mock_get_client):
"""Test error handling in generate_ai_feedback"""
# Setup mock to raise exception
mock_client = MagicMock()
mock_client.chat.completions.create.side_effect = Exception("API Error")
mock_get_client.return_value = (mock_client, "test-model")
feedback = generate_ai_feedback("cat", "Q?", "A", "tokens", {})
# Should return error string
self.assertIn("Error:", feedback)
if __name__ == "__main__":
unittest.main(verbosity=2)