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

Merge pull request #29 from russellballestrini/claude/yaml-spec-immersive-activities-011CUzPAMF9VZh8Po4GNrM6W

Browse source 
Identify YAML limitations for interactive activities
This commit is contained in:
Russell 2025-11-10 12:08:25 -05:00 • committed by GitHub
commit 5ad1279f72
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 2126 additions and 96 deletions
Download patch file Download diff file Expand all files Collapse all files

88
.github/workflows/test.yml vendored Normal file
View file

@ -0,0 +1,88 @@
name: Run Tests
on:
push:
branches: [ main, master, develop, claude/** ]
pull_request:
branches: [ main, master, develop ]
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.11']
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
cache: 'pip'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install -r requirements-test.txt
- name: Run unit tests
run: |
pytest tests/unit/ -v --tb=short --cov=. --cov-report=term-missing
env:
SQLALCHEMY_DATABASE_URI: sqlite:///:memory:
TESTING: 1
- name: Run functional tests
run: |
pytest tests/functional/ -v --tb=short
env:
SQLALCHEMY_DATABASE_URI: sqlite:///:memory:
TESTING: 1
- name: Run integration tests
run: |
pytest tests/integration/ -v --tb=short
env:
SQLALCHEMY_DATABASE_URI: sqlite:///:memory:
TESTING: 1
- name: Validate activity YAML files
run: |
python activity_yaml_validator.py research/SPEC.yaml
python activity_yaml_validator.py research/activity*.yaml
continue-on-error: true
lint:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
cache: 'pip'
- name: Install linting dependencies
run: |
python -m pip install --upgrade pip
pip install black flake8
- name: Check code formatting with black
run: |
black --check --diff .
continue-on-error: true
- name: Lint with flake8
run: |
# Stop the build if there are Python syntax errors or undefined names
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
# Exit-zero treats all errors as warnings
flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
continue-on-error: true

30
CLAUDE.md
View file

@ -78,6 +78,36 @@
## Activity YAML Schema
### Session Persistence & Multi-User Model ("Twitch Plays Pokemon")
**How OpenCompletion Activities Work:**
- **Single Shared Game State**: One activity instance per room/channel
- **Multiple Players**: Zero or more users can participate from different devices
- **Collaborative Control**: Any user can provide input to advance the shared game
- **Persistent Metadata**: State is stored in the database per-room, survives browser refreshes
- **Like "Twitch Plays Pokemon"**: Everyone sees the same state, anyone can control
**Key Implications:**
- `metadata` is **shared** across all users in the room - it's the game state, not player-specific
- When user "Alice" adds metadata, user "Bob" sees it too (same activity instance)
- Use metadata for: scores, progress, choices, inventory, flags - anything that's part of the game
- All users see the same content_blocks, questions, and transitions
- Multiple users can answer the same question - first valid answer advances the game
- Activities can be canceled, which deletes the room's activity state
**Session Lifecycle:**
1. Activity starts → Initial state saved to database (room_id, section_id, step_id, metadata)
2. Users interact → Metadata updates, state progresses through sections/steps
3. Activity completes → State deleted from database
4. Activity canceled → State deleted from database
**Use Cases:**
- Classroom activities where teacher projects screen, students call out answers
- Collaborative puzzles where multiple people work together
- Public challenges where community collectively progresses
- Educational games where everyone learns from same shared experience
### Model Configuration (New Feature)
Activities can specify separate models for classification and feedback generation:

241
activity.py
View file

@ -25,6 +25,18 @@ get_openai_client_and_model = None
# Import SYSTEM_USERS from app.py
SYSTEM_USERS = None
# Import activity utilities for v2.0 features
from activity_utils import (
render_template,
evaluate_condition,
check_conditions,
filter_content_blocks,
resolve_conditional_navigation,
select_weighted_random,
get_progressive_hint,
create_template_context
)
def handle_get_activity_status(data):
"""Get the current activity status for a room."""
@ -121,28 +133,57 @@ def loop_through_steps_until_question(
# Emit the current step content blocks
if "content_blocks" in step:
content = "\n\n".join(step["content_blocks"])
translated_content = translate_text(content, user_language, feedback_model)
new_message = Message(
username="System", content=translated_content, room_id=room.id
# Create template context
context = create_template_context(
metadata=activity_state.dict_metadata,
current_attempt=activity_state.attempts,
max_attempts=activity_state.max_attempts,
current_section=current_section_id,
current_step=current_step_id,
username=username
)
db.session.add(new_message)
db.session.commit()
socketio.emit(
"chat_message",
{
"id": new_message.id,
"username": "System",
"content": translated_content,
},
room=room_name,
# Filter and render content blocks (supports conditional blocks and templates)
filtered_blocks = filter_content_blocks(
step["content_blocks"],
activity_state.dict_metadata,
context
)
socketio.sleep(0.1)
if filtered_blocks:
content = "\n\n".join(filtered_blocks)
translated_content = translate_text(content, user_language, feedback_model)
new_message = Message(
username="System", content=translated_content, room_id=room.id
)
db.session.add(new_message)
db.session.commit()
socketio.emit(
"chat_message",
{
"id": new_message.id,
"username": "System",
"content": translated_content,
},
room=room_name,
)
socketio.sleep(0.1)
# Check if the current step has a question
if "question" in step:
question_content = step["question"]
# Create template context
context = create_template_context(
metadata=activity_state.dict_metadata,
current_attempt=activity_state.attempts,
max_attempts=activity_state.max_attempts,
current_section=current_section_id,
current_step=current_step_id,
username=username
)
# Render template variables in question
question_content = render_template(step["question"], context)
translated_question_content = translate_text(
question_content, user_language, feedback_model
)
@ -486,11 +527,11 @@ def handle_activity_response(room_name, user_response, username, model="MODEL_0"
)
socketio.sleep(0.05)
# Check metadata conditions for the current step
# Check metadata conditions for the current step (v2.0 advanced conditions)
if "metadata_conditions" in transition:
conditions_met = all(
activity_state.dict_metadata.get(key) == value
for key, value in transition["metadata_conditions"].items()
conditions_met = check_conditions(
activity_state.dict_metadata,
transition["metadata_conditions"]
)
if not conditions_met:
# Skip this transition if conditions not met
@ -685,6 +726,21 @@ def handle_activity_response(room_name, user_response, username, model="MODEL_0"
metadata_tmp_keys.append(random_key)
activity_state.add_metadata(random_key, random_value)
# Handle metadata_weighted_random (v2.0)
if "metadata_weighted_random" in transition:
for key, weighted_options in transition["metadata_weighted_random"].items():
selected_value = select_weighted_random(weighted_options)
new_metadata[key] = selected_value
activity_state.add_metadata(key, selected_value)
# Handle metadata_tmp_weighted_random (v2.0)
if "metadata_tmp_weighted_random" in transition:
for key, weighted_options in transition["metadata_tmp_weighted_random"].items():
selected_value = select_weighted_random(weighted_options)
new_metadata[key] = selected_value
metadata_tmp_keys.append(key)
activity_state.add_metadata(key, selected_value)
# Execute the post-script if it exists (supports both old and new naming)
post_script = step.get("post_script") or step.get("processing_script")
if post_script and (
@ -763,30 +819,48 @@ def handle_activity_response(room_name, user_response, username, model="MODEL_0"
user_language = activity_state.dict_metadata.get("language", "English")
# Emit the transition content blocks if they exist
# Emit the transition content blocks if they exist (v2.0 with templates & conditions)
if "content_blocks" in transition:
transition_content = "\n\n".join(transition["content_blocks"])
translated_transition_content = translate_text(
transition_content, user_language, feedback_model
# Create template context
context = create_template_context(
metadata=activity_state.dict_metadata,
current_attempt=activity_state.attempts,
max_attempts=activity_state.max_attempts,
current_section=activity_state.section_id,
current_step=activity_state.step_id,
username=username
)
new_message = Message(
username="System",
content=translated_transition_content,
room_id=room.id,
)
db.session.add(new_message)
db.session.commit()
socketio.emit(
"chat_message",
{
"id": new_message.id,
"username": "System",
"content": translated_transition_content,
},
room=room_name,
# Filter and render content blocks (supports conditional blocks and templates)
filtered_blocks = filter_content_blocks(
transition["content_blocks"],
activity_state.dict_metadata,
context
)
socketio.sleep(0.1)
if filtered_blocks:
transition_content = "\n\n".join(filtered_blocks)
translated_transition_content = translate_text(
transition_content, user_language, feedback_model
)
new_message = Message(
username="System",
content=translated_transition_content,
room_id=room.id,
)
db.session.add(new_message)
db.session.commit()
socketio.emit(
"chat_message",
{
"id": new_message.id,
"username": "System",
"content": translated_transition_content,
},
room=room_name,
)
socketio.sleep(0.1)
# if "correct" or max_attempts reached.
# Provide feedback based on the category
@ -885,6 +959,43 @@ def handle_activity_response(room_name, user_response, username, model="MODEL_0"
# End of multi-bucket processing loop
# Check for progressive hints (v2.0)
if "hints" in step:
context = create_template_context(
metadata=activity_state.dict_metadata,
current_attempt=activity_state.attempts + 1, # Next attempt
max_attempts=activity_state.max_attempts,
current_section=activity_state.section_id,
current_step=activity_state.step_id,
username=username
)
hint = get_progressive_hint(step["hints"], activity_state.attempts + 1, context)
if hint:
# Display hint
translated_hint = translate_text(hint['text'], user_language, feedback_model)
new_message = Message(
username="System (Hint)",
content=translated_hint,
room_id=room.id
)
db.session.add(new_message)
db.session.commit()
socketio.emit(
"chat_message",
{
"id": new_message.id,
"username": "System (Hint)",
"content": translated_hint,
},
room=room_name,
)
socketio.sleep(0.1)
# If hint doesn't count as attempt, don't increment
if not hint['counts_as_attempt']:
any_counts_as_attempt = False
if (
category
not in [
@ -898,20 +1009,32 @@ def handle_activity_response(room_name, user_response, username, model="MODEL_0"
or final_next_section_and_step # Use final navigation from last transition
):
if final_next_section_and_step:
(
current_section_id,
current_step_id,
) = final_next_section_and_step.split(":")
next_section = next(
s
for s in activity_content["sections"]
if s["section_id"] == current_section_id
)
next_step = next(
s
for s in next_section["steps"]
if s["step_id"] == current_step_id
# Resolve conditional navigation (v2.0)
resolved_navigation = resolve_conditional_navigation(
final_next_section_and_step,
activity_state.dict_metadata
)
if resolved_navigation:
(
current_section_id,
current_step_id,
) = resolved_navigation.split(":")
next_section = next(
s
for s in activity_content["sections"]
if s["section_id"] == current_section_id
)
next_step = next(
s
for s in next_section["steps"]
if s["step_id"] == current_step_id
)
else:
# No navigation resolved, move to next step
next_section, next_step = get_next_step(
activity_content, section["section_id"], step["step_id"]
)
else:
# Move to the next step or section
next_section, next_step = get_next_step(
@ -939,8 +1062,16 @@ def handle_activity_response(room_name, user_response, username, model="MODEL_0"
db.session.add(activity_state)
db.session.commit()
# Emit the question again
question_content = step["question"]
# Emit the question again (v2.0 with templates)
context = create_template_context(
metadata=activity_state.dict_metadata,
current_attempt=activity_state.attempts,
max_attempts=activity_state.max_attempts,
current_section=activity_state.section_id,
current_step=activity_state.step_id,
username=username
)
question_content = render_template(step["question"], context)
translated_question_content = translate_text(
question_content, user_language, feedback_model
)

352
activity_utils.py Normal file
View file

@ -0,0 +1,352 @@
"""
Utility functions for OpenCompletion Activity System v2.0
Features:
- Template variable rendering ({{metadata.key}}, {{current_attempt}}, etc.)
- Advanced metadata conditions (gte, lt, contains, regex, etc.)
- Conditional content blocks (show_if)
- Conditional navigation (if/elif/else)
- Weighted random selection
- Progressive hints
"""
import re
import random
from typing import Any, Dict, List, Optional, Union
def render_template(text: str, context: Dict[str, Any]) -> str:
"""
Render template variables in text using {{variable}} syntax.
Supports:
- {{metadata.key}} - Access metadata values
- {{current_attempt}} - Current attempt number
- {{max_attempts}} - Maximum attempts
- {{attempts_remaining}} - Remaining attempts
- {{current_section}} - Current section ID
- {{current_step}} - Current step ID
- {{username}} - Last responding username
Args:
text: Text containing {{variable}} templates
context: Dictionary with metadata, attempts, section/step info
Returns:
Text with variables replaced
"""
if not isinstance(text, str):
return text
# Find all {{variable}} patterns
pattern = r'\{\{([^}]+)\}\}'
def replace_variable(match):
var_name = match.group(1).strip()
# Handle metadata.key syntax
if var_name.startswith('metadata.'):
key = var_name[9:] # Remove 'metadata.' prefix
metadata = context.get('metadata', {})
value = metadata.get(key, f'{{{{metadata.{key}}}}}') # Keep original if not found
return str(value) if value is not None else ''
# Handle built-in variables
value = context.get(var_name, f'{{{{{var_name}}}}}') # Keep original if not found
return str(value) if value is not None else ''
return re.sub(pattern, replace_variable, text)
def evaluate_condition(metadata: Dict[str, Any], condition_key: str, condition_value: Any) -> bool:
"""
Evaluate a single condition against metadata.
Supports operators:
- key: value - Equality
- key_ne: value - Not equal
- key_gt: value - Greater than
- key_gte: value - Greater than or equal
- key_lt: value - Less than
- key_lte: value - Less than or equal
- key_between: [min, max] - Between (inclusive)
- key_contains: value - Comma-separated list contains value
- key_not_contains: value - List does NOT contain value
- key_matches: pattern - Regex match
- key_exists: true/false - Key existence check
- key_not_exists: true/false - Key non-existence check
Args:
metadata: Metadata dictionary to check
condition_key: Condition key (may have operator suffix)
condition_value: Expected value
Returns:
True if condition met, False otherwise
"""
# Check for operator suffixes
if condition_key.endswith('_ne'):
key = condition_key[:-3]
return metadata.get(key) != condition_value
elif condition_key.endswith('_gt'):
key = condition_key[:-3]
try:
return float(metadata.get(key, 0)) > float(condition_value)
except (ValueError, TypeError):
return False
elif condition_key.endswith('_gte'):
key = condition_key[:-4]
try:
return float(metadata.get(key, 0)) >= float(condition_value)
except (ValueError, TypeError):
return False
elif condition_key.endswith('_lt'):
key = condition_key[:-3]
try:
return float(metadata.get(key, 0)) < float(condition_value)
except (ValueError, TypeError):
return False
elif condition_key.endswith('_lte'):
key = condition_key[:-4]
try:
return float(metadata.get(key, 0)) <= float(condition_value)
except (ValueError, TypeError):
return False
elif condition_key.endswith('_between'):
key = condition_key[:-8]
if not isinstance(condition_value, list) or len(condition_value) != 2:
return False
try:
val = float(metadata.get(key, 0))
return float(condition_value[0]) <= val <= float(condition_value[1])
except (ValueError, TypeError):
return False
elif condition_key.endswith('_not_contains'):
key = condition_key[:-13]
value_str = str(metadata.get(key, ''))
items = [item.strip() for item in value_str.split(',') if item.strip()]
return str(condition_value) not in items
elif condition_key.endswith('_contains'):
key = condition_key[:-9]
value_str = str(metadata.get(key, ''))
# Split by comma and check if condition_value is in list
items = [item.strip() for item in value_str.split(',') if item.strip()]
return str(condition_value) in items
elif condition_key.endswith('_matches'):
key = condition_key[:-8]
value_str = str(metadata.get(key, ''))
try:
return bool(re.search(str(condition_value), value_str))
except re.error:
return False
elif condition_key.endswith('_not_exists'):
key = condition_key[:-11]
if condition_value:
return key not in metadata
else:
return key in metadata
elif condition_key.endswith('_exists'):
key = condition_key[:-7]
if condition_value:
return key in metadata
else:
return key not in metadata
else:
# Simple equality check
return metadata.get(condition_key) == condition_value
def check_conditions(metadata: Dict[str, Any], conditions: Dict[str, Any]) -> bool:
"""
Check if ALL conditions are met (AND logic).
Args:
metadata: Metadata dictionary
conditions: Dictionary of condition_key: condition_value pairs
Returns:
True if all conditions met, False otherwise
"""
if not conditions:
return True
return all(
evaluate_condition(metadata, key, value)
for key, value in conditions.items()
)
def filter_content_blocks(
content_blocks: List[Union[str, Dict[str, Any]]],
metadata: Dict[str, Any],
context: Dict[str, Any]
) -> List[str]:
"""
Filter and render content blocks based on show_if conditions.
Content blocks can be:
- Simple strings: Always shown
- Objects with 'text' and 'show_if': Conditionally shown
Args:
content_blocks: List of content blocks (strings or dicts)
metadata: Metadata dictionary for condition evaluation
context: Template rendering context
Returns:
List of rendered text strings that passed conditions
"""
result = []
for block in content_blocks:
if isinstance(block, str):
# Simple string - always show, just render templates
rendered = render_template(block, context)
result.append(rendered)
elif isinstance(block, dict):
# Conditional block - check show_if condition
text = block.get('text', '')
show_if = block.get('show_if', {})
# Check if conditions are met
if check_conditions(metadata, show_if):
rendered = render_template(text, context)
result.append(rendered)
return result
def resolve_conditional_navigation(
next_section_and_step: Union[str, List[Dict[str, Any]]],
metadata: Dict[str, Any]
) -> Optional[str]:
"""
Resolve conditional navigation (if/elif/else structure).
Args:
next_section_and_step: Either a string or list of conditional branches
metadata: Metadata dictionary for condition evaluation
Returns:
Resolved "section:step" string or None
"""
# Simple string - return as-is
if isinstance(next_section_and_step, str):
return next_section_and_step
# Conditional branches
if isinstance(next_section_and_step, list):
for branch in next_section_and_step:
if 'if' in branch:
# if branch
if check_conditions(metadata, branch['if']):
return branch.get('goto')
elif 'elif' in branch:
# elif branch
if check_conditions(metadata, branch['elif']):
return branch.get('goto')
elif 'else' in branch:
# else branch - always taken if reached
return branch.get('goto')
return None
def select_weighted_random(weighted_options: List[Dict[str, Any]]) -> Any:
"""
Select a random value from weighted options.
Args:
weighted_options: List of dicts with 'value' and 'weight' keys
Returns:
Selected value
"""
if not weighted_options:
return None
# Extract values and weights
values = [opt['value'] for opt in weighted_options]
weights = [opt.get('weight', 1) for opt in weighted_options]
# Use random.choices for weighted selection
selected = random.choices(values, weights=weights, k=1)
return selected[0]
def get_progressive_hint(
hints: List[Dict[str, Any]],
current_attempt: int,
context: Dict[str, Any]
) -> Optional[Dict[str, Any]]:
"""
Get the hint for the current attempt number, if one exists.
Args:
hints: List of hint dicts with 'attempt', 'text', 'counts_as_attempt' keys
current_attempt: Current attempt number (1, 2, 3, ...)
context: Template rendering context
Returns:
Hint dict with rendered text, or None if no hint for this attempt
"""
if not hints:
return None
for hint in hints:
if hint.get('attempt') == current_attempt:
# Render template variables in hint text
hint_text = render_template(hint.get('text', ''), context)
return {
'text': hint_text,
'counts_as_attempt': hint.get('counts_as_attempt', False)
}
return None
def create_template_context(
metadata: Dict[str, Any],
current_attempt: int,
max_attempts: int,
current_section: str,
current_step: str,
username: str = "User"
) -> Dict[str, Any]:
"""
Create a template rendering context with all built-in variables.
Args:
metadata: Activity metadata
current_attempt: Current attempt number
max_attempts: Maximum attempts allowed
current_section: Current section ID
current_step: Current step ID
username: Username of last responder
Returns:
Context dictionary for template rendering
"""
return {
'metadata': metadata,
'current_attempt': current_attempt,
'max_attempts': max_attempts,
'attempts_remaining': max(0, max_attempts - current_attempt),
'current_section': current_section,
'current_step': current_step,
'username': username
}

192
activity_yaml_validator.py
View file

@ -81,7 +81,9 @@ class ActivityYAMLValidator:
return len(self.errors) == 0, self.errors, self.warnings
except Exception as e:
import traceback
self.errors.append(f"Unexpected error: {e}")
self.errors.append(f"Traceback: {traceback.format_exc()}")
return False, self.errors, self.warnings
def _validate_structure(self, data: Dict[str, Any]):
@ -228,7 +230,7 @@ class ActivityYAMLValidator:
def _validate_content_blocks(
self, content_blocks: List[str], section_id: str, step_id: str
):
"""Validate content blocks"""
"""Validate content blocks (v2.0 supports conditional blocks)"""
if not isinstance(content_blocks, list):
self.errors.append(
f"Section {section_id}, step {step_id}: content_blocks must be a list"
@ -236,9 +238,28 @@ class ActivityYAMLValidator:
return
for i, block in enumerate(content_blocks):
if not isinstance(block, str):
if isinstance(block, str):
# Simple string block - always valid
continue
elif isinstance(block, dict):
# Conditional block (v2.0)
if 'text' not in block:
self.errors.append(
f"Section {section_id}, step {step_id}: content_blocks[{i}] dict must have 'text' field"
)
elif not isinstance(block['text'], str):
self.errors.append(
f"Section {section_id}, step {step_id}: content_blocks[{i}]['text'] must be a string"
)
if 'show_if' in block:
if not isinstance(block['show_if'], dict):
self.errors.append(
f"Section {section_id}, step {step_id}: content_blocks[{i}]['show_if'] must be a dict"
)
else:
self.errors.append(
f"Section {section_id}, step {step_id}: content_blocks[{i}] must be a string"
f"Section {section_id}, step {step_id}: content_blocks[{i}] must be a string or dict"
)
def _validate_question_step(
@ -269,6 +290,10 @@ class ActivityYAMLValidator:
step["feedback_prompts"], section_id, step_id
)
# Validate hints (v2.0 progressive hints)
if "hints" in step:
self._validate_hints(step["hints"], section_id, step_id)
# Validate buckets and transitions
if "buckets" in step:
self._validate_buckets(step["buckets"], section_id, step_id)
@ -469,16 +494,21 @@ class ActivityYAMLValidator:
)
return
# Validate next_section_and_step format
# Validate next_section_and_step format (v2.0 supports conditional navigation)
if "next_section_and_step" in transition:
next_step = transition["next_section_and_step"]
if not isinstance(next_step, str):
if isinstance(next_step, str):
# Simple string navigation
if ":" not in next_step:
self.errors.append(
f"Section {section_id}, step {step_id}, bucket {bucket}: 'next_section_and_step' must be in format 'section_id:step_id'"
)
elif isinstance(next_step, list):
# Conditional navigation (v2.0)
self._validate_conditional_navigation(next_step, bucket, section_id, step_id)
else:
self.errors.append(
f"Section {section_id}, step {step_id}, bucket {bucket}: 'next_section_and_step' must be a string"
)
elif ":" not in next_step:
self.errors.append(
f"Section {section_id}, step {step_id}, bucket {bucket}: 'next_section_and_step' must be in format 'section_id:step_id'"
f"Section {section_id}, step {step_id}, bucket {bucket}: 'next_section_and_step' must be a string or list"
)
# Validate metadata operations
@ -488,6 +518,8 @@ class ActivityYAMLValidator:
"metadata_remove",
"metadata_clear",
"metadata_feedback_filter",
"metadata_weighted_random", # v2.0
"metadata_tmp_weighted_random", # v2.0
]
for field in metadata_fields:
if field in transition:
@ -554,11 +586,110 @@ class ActivityYAMLValidator:
f"Section {section_id}, step {step_id}, bucket {bucket}: 'content_blocks' must be a list"
)
else:
for i, block in enumerate(transition["content_blocks"]):
if not isinstance(block, str):
self.errors.append(
f"Section {section_id}, step {step_id}, bucket {bucket}: content_blocks[{i}] must be a string"
)
# v2.0: content_blocks can be strings or dicts with text/show_if
self._validate_content_blocks(transition["content_blocks"], section_id, f"{step_id}:{bucket}")
def _validate_hints(self, hints: List[Dict[str, Any]], section_id: str, step_id: str):
"""Validate progressive hints system (v2.0)"""
if not isinstance(hints, list):
self.errors.append(
f"Section {section_id}, step {step_id}: 'hints' must be a list"
)
return
if not hints:
self.warnings.append(
f"Section {section_id}, step {step_id}: Empty hints list"
)
return
for i, hint in enumerate(hints):
if not isinstance(hint, dict):
self.errors.append(
f"Section {section_id}, step {step_id}: hints[{i}] must be a dictionary"
)
continue
# Validate required fields
if 'attempt' not in hint:
self.errors.append(
f"Section {section_id}, step {step_id}: hints[{i}] missing required field 'attempt'"
)
elif not isinstance(hint['attempt'], int) or hint['attempt'] < 1:
self.errors.append(
f"Section {section_id}, step {step_id}: hints[{i}]['attempt'] must be a positive integer"
)
if 'text' not in hint:
self.errors.append(
f"Section {section_id}, step {step_id}: hints[{i}] missing required field 'text'"
)
elif not isinstance(hint['text'], str):
self.errors.append(
f"Section {section_id}, step {step_id}: hints[{i}]['text'] must be a string"
)
# Validate optional fields
if 'counts_as_attempt' in hint and not isinstance(hint['counts_as_attempt'], bool):
self.errors.append(
f"Section {section_id}, step {step_id}: hints[{i}]['counts_as_attempt'] must be a boolean"
)
def _validate_conditional_navigation(
self, nav_list: List[Dict[str, Any]], bucket: str, section_id: str, step_id: str
):
"""Validate conditional navigation structure (v2.0)"""
if not isinstance(nav_list, list):
self.errors.append(
f"Section {section_id}, step {step_id}, bucket {bucket}: conditional navigation must be a list"
)
return
has_else = False
for i, branch in enumerate(nav_list):
if not isinstance(branch, dict):
self.errors.append(
f"Section {section_id}, step {step_id}, bucket {bucket}: navigation[{i}] must be a dictionary"
)
continue
# Check for if/elif/else
if 'if' in branch:
if not isinstance(branch['if'], dict):
self.errors.append(
f"Section {section_id}, step {step_id}, bucket {bucket}: navigation[{i}]['if'] must be a dict"
)
elif 'elif' in branch:
if not isinstance(branch['elif'], dict):
self.errors.append(
f"Section {section_id}, step {step_id}, bucket {bucket}: navigation[{i}]['elif'] must be a dict"
)
elif 'else' in branch:
has_else = True
# else doesn't need conditions
else:
self.errors.append(
f"Section {section_id}, step {step_id}, bucket {bucket}: navigation[{i}] must have 'if', 'elif', or 'else'"
)
# Check for goto
if 'goto' not in branch:
self.errors.append(
f"Section {section_id}, step {step_id}, bucket {bucket}: navigation[{i}] missing required field 'goto'"
)
elif not isinstance(branch['goto'], str):
self.errors.append(
f"Section {section_id}, step {step_id}, bucket {bucket}: navigation[{i}]['goto'] must be a string"
)
elif ':' not in branch['goto']:
self.errors.append(
f"Section {section_id}, step {step_id}, bucket {bucket}: navigation[{i}]['goto'] must be in format 'section_id:step_id'"
)
if not has_else:
self.warnings.append(
f"Section {section_id}, step {step_id}, bucket {bucket}: conditional navigation has no 'else' clause - may not always resolve"
)
def _validate_python_code(self, data: Dict[str, Any]):
"""Validate Python code blocks in scripts"""
@ -697,9 +828,12 @@ class ActivityYAMLValidator:
# Check if any transition continues the flow
has_continuing_transition = False
for transition in step["transitions"].values():
if "next_section_and_step" in transition:
has_continuing_transition = True
break
if isinstance(transition, dict) and "next_section_and_step" in transition:
# v2.0: next_section_and_step can be string or list (conditional)
next_step_value = transition["next_section_and_step"]
if next_step_value: # Not None or empty
has_continuing_transition = True
break
# If this is the last step of the last section and has no continuing transitions
if (
@ -808,12 +942,24 @@ class ActivityYAMLValidator:
continue
for bucket, transition in step["transitions"].items():
if "next_section_and_step" in transition:
if isinstance(transition, dict) and "next_section_and_step" in transition:
target = transition["next_section_and_step"]
if target not in all_steps:
self.errors.append(
f"Section {section_id}, step {step_id}: Invalid transition target '{target}'"
)
# v2.0: target can be string or list (conditional navigation)
if isinstance(target, str):
if target not in all_steps:
self.errors.append(
f"Section {section_id}, step {step_id}: Invalid transition target '{target}'"
)
elif isinstance(target, list):
# Conditional navigation - check all goto targets
for branch in target:
if isinstance(branch, dict) and 'goto' in branch:
goto_target = branch['goto']
if goto_target not in all_steps:
self.errors.append(
f"Section {section_id}, step {step_id}: Invalid conditional navigation target '{goto_target}'"
)
def main():

401
research/SPEC.yaml
View file

@ -632,6 +632,407 @@ sections:
# 42 → Store integer
# true / false → Store boolean
# ==============================================================================
# ADVANCED FEATURES (New in v2.0)
# ==============================================================================
# ==============================================================================
# TEMPLATE VARIABLES
# ==============================================================================
# Use {{variable_name}} syntax to insert dynamic values into content
# Available in: content_blocks, questions, ai_feedback tokens
# Built-in Variables:
# -------------------
# {{current_attempt}} → Current attempt number (1, 2, 3...)
# {{max_attempts}} → Maximum attempts allowed for this step
# {{attempts_remaining}} → How many attempts left (max - current)
# {{current_section}} → Current section_id
# {{current_step}} → Current step_id
# {{username}} → Name of the user who last responded
# Metadata Variables:
# -------------------
# {{metadata.key_name}} → Access any metadata value
# {{metadata.score}} → Example: access score
# {{metadata.player_name}} → Example: access player name
# Example Usage:
content_blocks:
- "## Your Progress"
- "Welcome back, {{metadata.player_name}}!"
- "Score: {{metadata.score}}"
- "Level: {{metadata.level}}"
- "Attempt {{current_attempt}} of {{max_attempts}}"
- "You have {{attempts_remaining}} tries remaining"
question: "{{metadata.character_name}} asks: What will you do?"
# Templates work in:
# - step content_blocks
# - transition content_blocks
# - question text
# - ai_feedback tokens_for_ai (for context, not rendered directly)
# ==============================================================================
# CONDITIONAL CONTENT BLOCKS
# ==============================================================================
# Show/hide content blocks based on metadata conditions
# Format: Each content block can be a string OR an object with conditions
content_blocks:
# Simple string - always shown
- "This is always displayed"
# Conditional block - only shown if conditions met
- text: "You're doing great! Keep going!"
show_if:
score_gte: 50 # Only show if score >= 50
- text: "Need more practice. Don't give up!"
show_if:
score_lt: 50 # Only show if score < 50
- text: "You found the secret key! 🗝️"
show_if:
inventory_contains: "key" # Only if inventory contains "key"
- text: "Welcome, warrior! ⚔️"
show_if:
class: "warrior" # Only if metadata.class equals "warrior"
- text: "Welcome, mage! 🔮"
show_if:
class: "mage"
# Conditional blocks reduce step duplication - one step, multiple paths!
# ==============================================================================
# ADVANCED METADATA CONDITIONS
# ==============================================================================
# Rich comparison operators for metadata_conditions
# Previously only supported equality:
metadata_conditions:
level: 5 # metadata.level must equal 5
# Now supports:
# --------------
# Equality & Inequality:
metadata_conditions:
status: "active" # Equal to "active"
status_ne: "inactive" # Not equal to "inactive"
# Numeric Comparisons:
metadata_conditions:
score_gte: 100 # Greater than or equal to 100
score_gt: 99 # Greater than 99
score_lt: 200 # Less than 200
score_lte: 199 # Less than or equal to 199
level_between: [5, 10] # Between 5 and 10 (inclusive)
# String Operations:
metadata_conditions:
inventory_contains: "sword" # Comma-separated list contains "sword"
inventory_not_contains: "poison" # List does NOT contain "poison"
name_matches: "^[A-Z]" # Regex match (starts with capital)
# Existence Checks:
metadata_conditions:
has_key_exists: true # Key "has_key" must exist in metadata
temp_flag_not_exists: true # Key "temp_flag" must NOT exist
# Boolean Checks:
metadata_conditions:
is_admin: true # metadata.is_admin must be true
is_locked: false # metadata.is_locked must be false
# Combining Multiple Conditions (ALL must be true):
metadata_conditions:
score_gte: 100
level_gte: 5
inventory_contains: "key"
quest_completed: true
# All four conditions must be met
# ==============================================================================
# CONDITIONAL NAVIGATION
# ==============================================================================
# Choose different paths based on metadata state
# OLD WAY (still works):
transitions:
answer_provided:
next_section_and_step: "section_2:step_1"
# NEW WAY - Conditional branches:
transitions:
answer_provided:
next_section_and_step:
- if:
score_gte: 100
goto: "expert:challenge"
- elif:
score_gte: 50
goto: "intermediate:lesson"
- elif:
score_gte: 25
goto: "beginner:practice"
- else:
goto: "tutorial:basics"
# Another example: Quest completion paths
transitions:
quest_complete:
next_section_and_step:
- if:
all_secrets_found: true
perfect_score: true
goto: "endings:perfect_ending"
- elif:
all_secrets_found: true
goto: "endings:good_ending"
- elif:
quest_failed: true
goto: "endings:bad_ending"
- else:
goto: "endings:neutral_ending"
# Conditions use same operators as metadata_conditions:
# - Equality: key: value
# - Comparisons: key_gte, key_gt, key_lt, key_lte
# - String ops: key_contains, key_not_contains, key_matches
# - Existence: key_exists, key_not_exists
# - Boolean: key: true/false
# ==============================================================================
# PROGRESSIVE HINTS SYSTEM
# ==============================================================================
# Built-in system for providing hints that escalate with attempts
#
# Example step with progressive hints:
#
# - step_id: "difficult_question"
# question: "What is the capital of Burkina Faso?"
#
# hints:
# - attempt: 1
# text: "💡 Hint: It's not the largest city in the country."
# counts_as_attempt: false
#
# - attempt: 2
# text: "💡 Hint: The name means 'City of Honest People'."
# counts_as_attempt: false
#
# - attempt: 3
# text: "💡 Strong Hint: It starts with 'Oua' and ends with 'dougou'."
# counts_as_attempt: false
#
# buckets: [correct, incorrect, need_hint]
#
# transitions:
# correct:
# content_blocks:
# - "Excellent! Ouagadougou is correct!"
# next_section_and_step: "next:step"
#
# incorrect:
# content_blocks:
# - "Not quite. Try again!"
# next_section_and_step: "current:difficult_question"
#
# need_hint:
# content_blocks:
# - "Let me help you..."
# counts_as_attempt: false
# next_section_and_step: "current:difficult_question"
#
# Hints auto-display when attempt number matches
# Hints support template variables: "Attempt {{current_attempt}} of {{max_attempts}}"
progressive_hints_example: "See activity-test-v2-features.yaml for working example"
# ==============================================================================
# WEIGHTED RANDOM SELECTION
# ==============================================================================
# Choose random values with different probabilities
# OLD WAY - Equal probability:
metadata_random:
loot: "sword" # 33% each
loot: "dagger" # 33% each
loot: "staff" # 33% each
# NEW WAY - Weighted probabilities:
metadata_weighted_random:
loot:
- value: "common_sword"
weight: 70 # 70% chance
- value: "rare_dagger"
weight: 25 # 25% chance
- value: "legendary_staff"
weight: 5 # 5% chance
# Weights don't need to sum to 100 - they're relative:
metadata_weighted_random:
reward:
- value: "gold"
weight: 10 # 10/(10+3+1) = 71.4%
- value: "gem"
weight: 3 # 3/(10+3+1) = 21.4%
- value: "artifact"
weight: 1 # 1/(10+3+1) = 7.1%
# Also works with metadata_tmp_weighted_random for temporary values
# Example: Random encounter
transitions:
explore_forest:
metadata_weighted_random:
encounter:
- value: "nothing"
weight: 50 # 50% - No encounter
- value: "merchant"
weight: 30 # 30% - Friendly merchant
- value: "goblin"
weight: 15 # 15% - Fight goblin
- value: "treasure"
weight: 5 # 5% - Find treasure!
ai_feedback:
tokens_for_ai: |
Describe what happens based on metadata.encounter:
- nothing: Peaceful walk through forest
- merchant: Meet a traveling merchant
- goblin: Surprise goblin attack!
- treasure: Discover hidden treasure chest!
# ==============================================================================
# DYNAMIC QUESTION TEXT
# ==============================================================================
# Questions can now use template variables
# Static question (old way):
question: "What is 2 + 2?"
# Dynamic question with templates (new way):
question: "What is {{metadata.num1}} + {{metadata.num2}}?"
# Example: Math quiz with random numbers
#
# - step_id: "addition"
# pre_script: |
# import random
# result = {
# "metadata": {
# "num1": random.randint(1, 10),
# "num2": random.randint(1, 10)
# }
# }
# return result
#
# question: "What is {{metadata.num1}} + {{metadata.num2}}?"
#
# tokens_for_ai: |
# Calculate the correct answer: {{metadata.num1}} + {{metadata.num2}}
# Categorize as 'correct' if their answer matches.
#
# buckets: [correct, incorrect]
#
# More personalized question examples:
# - "{{metadata.character_name}}, what is your quest?"
# - "You have {{metadata.gold}} gold. How much do you spend?"
# - "Round {{current_attempt}}: What's your move?"
dynamic_question_example: "See activity-test-v2-features.yaml for working example"
# ==============================================================================
# BUILT-IN ATTEMPT COUNTER ACCESS
# ==============================================================================
# Access attempt information in templates
# Available variables:
# - {{current_attempt}} : 1, 2, 3, ... (current attempt number)
# - {{max_attempts}} : 3 (or custom value from default_max_attempts_per_step)
# - {{attempts_remaining}} : max_attempts - current_attempt
# Examples:
content_blocks:
- "Attempt {{current_attempt}} of {{max_attempts}}"
- "You have {{attempts_remaining}} tries left"
question: "Try {{current_attempt}}: What's your answer?"
ai_feedback:
tokens_for_ai: |
This is attempt {{current_attempt}} of {{max_attempts}}.
{% if attempts_remaining == 1 %}
This is their last chance! Be clear and helpful.
{% elif attempts_remaining == 2 %}
They still have time. Provide a gentle hint.
{% else %}
Encourage them to think carefully.
{% endif %}
# Conditional content based on attempts:
content_blocks:
- text: "First try - think carefully!"
show_if:
current_attempt: 1
- text: "Second try - you're getting closer!"
show_if:
current_attempt: 2
- text: "Last chance! Here's a hint..."
show_if:
current_attempt: 3
# ==============================================================================
# SESSION PERSISTENCE (Twitch Plays Model)
# ==============================================================================
# How metadata and state persist across users and sessions
# Key Facts:
# ----------
# 1. ONE GAME STATE PER ROOM: All users in a room share the same activity state
# 2. METADATA IS SHARED: When one user updates metadata, all users see it
# 3. DATABASE PERSISTENCE: State survives browser refreshes and reconnections
# 4. ANYONE CAN CONTROL: Any user can provide input to advance the shared game
# 5. LIKE TWITCH PLAYS POKEMON: Collaborative control of single game instance
# Lifecycle:
# ----------
# Activity starts → State saved to database (room_id, section_id, step_id, metadata)
# User interacts → Metadata updates, state progresses
# Browser refreshes → State persists (loaded from database)
# Activity completes → State deleted from database
# Activity canceled → State deleted from database
# Use Cases:
# ----------
# - Classroom: Teacher projects, students call out answers collectively
# - Collaboration: Multiple people solve puzzle together
# - Public challenges: Community progresses through shared experience
# - Learning together: Everyone learns from same shared game state
# Implications for Activity Design:
# ----------------------------------
# - Design for SHARED state, not per-player state
# - Metadata represents THE GAME, not individual players
# - Multiple users may answer - first valid response advances
# - Consider: "What if 10 people are playing together?"
# ==============================================================================
# VALIDATION RULES
# ==============================================================================

203
research/activity-test-v2-features.yaml Normal file
View file

@ -0,0 +1,203 @@
default_max_attempts_per_step: 3
classifier_model: "MODEL_0"
feedback_model: "MODEL_0"
tokens_for_ai_rubric: |
Test activity for v2.0 features.
Evaluate responses generously - this is just a demo!
sections:
- section_id: intro
title: V2.0 Features Demo
steps:
# Test: Template variables in content blocks
- step_id: welcome
title: Welcome with Templates
content_blocks:
- "# Welcome to OpenCompletion V2.0! 🎉"
- ""
- "This activity demonstrates all new v2.0 features."
- "Current section: {{current_section}}"
- "Current step: {{current_step}}"
question: "What's your name?"
tokens_for_ai: |
Categorize as 'name_provided' if they give a name.
Otherwise 'off_topic'.
buckets: [name_provided, off_topic]
transitions:
name_provided:
content_blocks:
- "Great to meet you!"
metadata_add:
player_name: "the-users-response"
score: "n+1"
next_section_and_step: "templates:test_templates"
off_topic:
content_blocks:
- "Please tell me your name."
next_section_and_step: "intro:welcome"
# Section: Template Variables
- section_id: templates
title: Template Variables Test
steps:
- step_id: test_templates
title: Testing Templates
content_blocks:
- "# Template Variables Test"
- ""
- "Welcome back, {{metadata.player_name}}!"
- "Your score: {{metadata.score}}"
- "Attempt {{current_attempt}} of {{max_attempts}}"
- "Attempts remaining: {{attempts_remaining}}"
question: "Ready to test conditional content? (yes/no)"
tokens_for_ai: "Categorize as 'yes' or 'no' based on their response."
buckets: [yes, no]
transitions:
yes:
content_blocks:
- "Excellent!"
next_section_and_step: "conditionals:test_conditional_blocks"
no:
content_blocks:
- "Take your time!"
next_section_and_step: "templates:test_templates"
# Section: Conditional Content Blocks
- section_id: conditionals
title: Conditional Content Test
steps:
- step_id: test_conditional_blocks
title: Conditional Content Blocks
content_blocks:
# Always shown
- "# Conditional Content Test"
- ""
# Conditional - only if score >= 1
- text: "🌟 You have points! Great job!"
show_if:
score_gte: 1
# Conditional - only if score < 1
- text: "Start earning points!"
show_if:
score_lt: 1
# Conditional - personalized
- text: "Hello {{metadata.player_name}}, let's continue!"
show_if:
player_name_exists: true
question: "What's 5 + 3?"
tokens_for_ai: "Categorize as 'correct' if 8 or eight, otherwise 'incorrect'."
buckets: [correct, incorrect]
# Progressive hints test
hints:
- attempt: 1
text: "💡 Hint: It's less than 10"
counts_as_attempt: false
- attempt: 2
text: "💡 Strong Hint: 5 + 3 = ?"
counts_as_attempt: false
transitions:
correct:
content_blocks:
- "Perfect! ✅"
metadata_add:
score: "n+5"
next_section_and_step: "weighted_random:test_weighted"
incorrect:
content_blocks:
- "Try again!"
next_section_and_step: "conditionals:test_conditional_blocks"
# Section: Weighted Random
- section_id: weighted_random
title: Weighted Random Test
steps:
- step_id: test_weighted
title: Weighted Random Selection
content_blocks:
- "# Weighted Random Test"
- ""
- "Let's test weighted random selection!"
question: "Roll the dice! (type 'roll')"
tokens_for_ai: "Categorize as 'roll'."
buckets: [roll]
transitions:
roll:
metadata_weighted_random:
loot:
- value: "common_item"
weight: 70
- value: "rare_item"
weight: 25
- value: "legendary_item"
weight: 5
ai_feedback:
tokens_for_ai: |
The user found: {{metadata.loot}}
If common_item: "You found a Common Item"
If rare_item: "You found a Rare Item! 🌟"
If legendary_item: "LEGENDARY ITEM FOUND! 🏆"
metadata_add:
score: "n+1"
next_section_and_step: "conditional_nav:test_nav"
# Section: Conditional Navigation
- section_id: conditional_nav
title: Conditional Navigation Test
steps:
- step_id: test_nav
title: Conditional Navigation
content_blocks:
- "# Conditional Navigation Test"
- ""
- "Your current score: {{metadata.score}}"
- ""
- "Based on your score, you'll be routed to different paths!"
question: "Continue? (yes)"
tokens_for_ai: "Categorize as 'continue'."
buckets: [continue]
transitions:
continue:
# Conditional navigation based on score
next_section_and_step:
- if:
score_gte: 10
goto: "endings:high_score"
- elif:
score_gte: 5
goto: "endings:medium_score"
- else:
goto: "endings:low_score"
# Section: Different Endings
- section_id: endings
title: Endings
steps:
- step_id: high_score
title: High Score Ending
content_blocks:
- "# 🏆 AMAZING! High Score!"
- ""
- "{{metadata.player_name}}, you scored {{metadata.score}} points!"
- ""
- "You're a V2.0 features master!"
- step_id: medium_score
title: Medium Score Ending
content_blocks:
- "# 🌟 GOOD JOB! Medium Score!"
- ""
- "{{metadata.player_name}}, you scored {{metadata.score}} points!"
- ""
- "Great understanding of V2.0 features!"
- step_id: low_score
title: Low Score Ending
content_blocks:
- "# ✨ Good Start!"
- ""
- "{{metadata.player_name}}, you scored {{metadata.score}} points!"
- ""
- "You've learned the basics of V2.0 features!"

137
research/guarded_ai.py
View file

@ -3,8 +3,23 @@ import yaml
import json
import random
import os
import sys
from openai import OpenAI
# Add parent directory to path to import activity_utils
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# Import activity utilities for v2.0 features
from activity_utils import (
render_template,
check_conditions,
filter_content_blocks,
resolve_conditional_navigation,
select_weighted_random,
get_progressive_hint,
create_template_context
)
# Global model-client mapping
MODEL_CLIENT_MAP = {}
@ -366,11 +381,32 @@ def simulate_activity(yaml_file_path):
# Get the user's language preference from metadata
user_language = metadata.get("language", "English")
# Translate and print all content blocks once per step
# Initialize attempts and max_attempts for this step
attempts = 0
step_max_attempts = step.get("max_attempts_per_step", max_attempts)
# Create template context for rendering
context = create_template_context(
metadata=metadata,
current_attempt=attempts,
max_attempts=step_max_attempts,
current_section=current_section_id,
current_step=current_step_id,
username="User"
)
# Translate and print all content blocks once per step (v2.0 with templates & conditionals)
if "content_blocks" in step:
content = "\n\n".join(step["content_blocks"])
translated_content = translate_text(content, user_language, feedback_model)
print(translated_content)
# Filter and render content blocks
filtered_blocks = filter_content_blocks(
step["content_blocks"],
metadata,
context
)
if filtered_blocks:
content = "\n\n".join(filtered_blocks)
translated_content = translate_text(content, user_language, feedback_model)
print(translated_content)
# Skip classification and feedback if there's no question
if "question" not in step:
@ -379,12 +415,22 @@ def simulate_activity(yaml_file_path):
)
continue
question = step["question"]
# Render template variables in question (v2.0)
question = render_template(step["question"], context)
translated_question = translate_text(question, user_language, feedback_model)
print(f"\nQuestion: {translated_question}")
attempts = 0
while attempts < max_attempts:
while attempts < step_max_attempts:
# Update context with current attempt
context = create_template_context(
metadata=metadata,
current_attempt=attempts + 1, # 1-indexed for display
max_attempts=step_max_attempts,
current_section=current_section_id,
current_step=current_step_id,
username="User"
)
user_response = input("\nYour Response: ")
# Roll for random buckets BEFORE categorization
@ -470,24 +516,42 @@ def simulate_activity(yaml_file_path):
print(f"Processing transition for bucket: '{bucket_name}'")
print(f"{'='*60}")
# Check metadata conditions
# Check metadata conditions (v2.0 advanced conditions)
if "metadata_conditions" in transition:
conditions_met = all(
metadata.get(key) == value
for key, value in transition["metadata_conditions"].items()
conditions_met = check_conditions(
metadata,
transition["metadata_conditions"]
)
if not conditions_met:
print(f"⚠️ Skipping '{bucket_name}' - metadata conditions not met")
print(f"Current Metadata: {json.dumps(metadata, indent=2)}")
continue
# Print transition content blocks if they exist
# Print transition content blocks if they exist (v2.0 with templates & conditionals)
if "content_blocks" in transition:
transition_content = "\n\n".join(transition["content_blocks"])
translated_transition_content = translate_text(
transition_content, user_language, feedback_model
# Create template context
context = create_template_context(
metadata=metadata,
current_attempt=attempts,
max_attempts=max_attempts,
current_section=current_section_id,
current_step=current_step_id,
username="User"
)
print(translated_transition_content)
# Filter and render content blocks (supports conditional blocks and templates)
filtered_blocks = filter_content_blocks(
transition["content_blocks"],
metadata,
context
)
if filtered_blocks:
transition_content = "\n\n".join(filtered_blocks)
translated_transition_content = translate_text(
transition_content, user_language, feedback_model
)
print(translated_transition_content)
# Update metadata based on user actions
if "metadata_add" in transition:
@ -604,6 +668,19 @@ def simulate_activity(yaml_file_path):
metadata[random_key] = random_value
metadata_tmp_keys.append(random_key) # Track temporary keys
# Handle metadata_weighted_random (v2.0)
if "metadata_weighted_random" in transition:
for key, weighted_options in transition["metadata_weighted_random"].items():
selected_value = select_weighted_random(weighted_options)
metadata[key] = selected_value
# Handle metadata_tmp_weighted_random (v2.0)
if "metadata_tmp_weighted_random" in transition:
for key, weighted_options in transition["metadata_tmp_weighted_random"].items():
selected_value = select_weighted_random(weighted_options)
metadata[key] = selected_value
metadata_tmp_keys.append(key)
# Execute the processing script if it exists
if "processing_script" in step and transition.get(
"run_processing_script", False
@ -674,6 +751,24 @@ def simulate_activity(yaml_file_path):
# End of multi-bucket processing loop
# Check for progressive hints (v2.0)
if "hints" in step:
hint_context = create_template_context(
metadata=metadata,
current_attempt=attempts + 1, # Next attempt
max_attempts=step_max_attempts,
current_section=current_section_id,
current_step=current_step_id,
username="User"
)
hint = get_progressive_hint(step["hints"], attempts + 1, hint_context)
if hint:
translated_hint = translate_text(hint['text'], user_language, feedback_model)
print(f"\n💡 Hint: {translated_hint}")
# If hint doesn't count as attempt, adjust counting
if not hint['counts_as_attempt']:
any_counts_as_attempt = False
# Check if we should break or continue attempting
if category not in [
"partial_understanding",
@ -688,7 +783,7 @@ def simulate_activity(yaml_file_path):
if any_counts_as_attempt:
attempts += 1
if attempts == max_attempts:
if attempts == step_max_attempts:
print("\nMaximum attempts reached. Moving to the next step.")
# Remove temporary metadata at the end of the step
@ -697,8 +792,14 @@ def simulate_activity(yaml_file_path):
del metadata[key]
# Use the final navigation target (from LAST processed transition)
# v2.0: Resolve conditional navigation
if final_next_section_and_step:
current_section_id, current_step_id = final_next_section_and_step.split(":")
resolved_navigation = resolve_conditional_navigation(
final_next_section_and_step,
metadata
)
if resolved_navigation:
current_section_id, current_step_id = resolved_navigation.split(":")
else:
# No navigation specified, move to next step automatically
current_section_id, current_step_id = get_next_section_and_step(

578
tests/unit/test_activity_utils.py Normal file
View file

@ -0,0 +1,578 @@
"""
Unit tests for activity_utils.py v2.0 features
Tests cover:
- Template variable rendering ({{variable}})
- Condition evaluation (gte, lt, contains, regex, etc.)
- Content block filtering (conditional show_if)
- Conditional navigation (if/elif/else)
- Weighted random selection
- Progressive hints system
- Template context creation
"""
import pytest
import re
from activity_utils import (
render_template,
evaluate_condition,
check_conditions,
filter_content_blocks,
resolve_conditional_navigation,
select_weighted_random,
get_progressive_hint,
create_template_context
)
class TestRenderTemplate:
"""Test template variable rendering with {{variable}} syntax"""
def test_simple_variable(self):
"""Test simple variable substitution"""
context = {"score": 100}
result = render_template("Score: {{score}}", context)
assert result == "Score: 100"
def test_metadata_variable(self):
"""Test metadata.key syntax"""
context = {"metadata": {"player_name": "Alice", "level": 5}}
result = render_template("Player: {{metadata.player_name}}, Level: {{metadata.level}}", context)
assert result == "Player: Alice, Level: 5"
def test_built_in_variables(self):
"""Test built-in variables (current_attempt, max_attempts, etc.)"""
context = {
"current_attempt": 2,
"max_attempts": 3,
"attempts_remaining": 1,
"current_section": "intro",
"current_step": "welcome",
"username": "Bob"
}
result = render_template(
"Attempt {{current_attempt}}/{{max_attempts}} ({{attempts_remaining}} left) - {{username}}",
context
)
assert result == "Attempt 2/3 (1 left) - Bob"
def test_missing_variable(self):
"""Test that missing variables are preserved in output"""
context = {"score": 100}
result = render_template("Score: {{score}}, Level: {{level}}", context)
assert result == "Score: 100, Level: {{level}}"
def test_missing_metadata_key(self):
"""Test missing metadata key is preserved"""
context = {"metadata": {"score": 50}}
result = render_template("{{metadata.score}} - {{metadata.missing}}", context)
assert result == "50 - {{metadata.missing}}"
def test_non_string_values(self):
"""Test rendering non-string values"""
context = {"score": 0, "active": True, "metadata": {"value": None}}
result = render_template("{{score}} {{active}} {{metadata.value}}", context)
assert result == "0 True "
def test_no_variables(self):
"""Test text with no variables"""
result = render_template("Plain text", {})
assert result == "Plain text"
def test_multiple_same_variable(self):
"""Test same variable used multiple times"""
context = {"name": "Test"}
result = render_template("{{name}} says {{name}}", context)
assert result == "Test says Test"
def test_non_string_input(self):
"""Test non-string input returns unchanged"""
assert render_template(123, {}) == 123
assert render_template(None, {}) is None
class TestEvaluateCondition:
"""Test single condition evaluation with various operators"""
def test_equality(self):
"""Test simple equality check"""
assert evaluate_condition({"level": 5}, "level", 5) is True
assert evaluate_condition({"level": 5}, "level", 4) is False
def test_not_equal(self):
"""Test not equal operator (_ne)"""
assert evaluate_condition({"status": "active"}, "status_ne", "inactive") is True
assert evaluate_condition({"status": "active"}, "status_ne", "active") is False
def test_greater_than(self):
"""Test greater than operator (_gt)"""
assert evaluate_condition({"score": 100}, "score_gt", 99) is True
assert evaluate_condition({"score": 100}, "score_gt", 100) is False
assert evaluate_condition({"score": 100}, "score_gt", 101) is False
def test_greater_than_or_equal(self):
"""Test greater than or equal operator (_gte)"""
assert evaluate_condition({"score": 100}, "score_gte", 99) is True
assert evaluate_condition({"score": 100}, "score_gte", 100) is True
assert evaluate_condition({"score": 100}, "score_gte", 101) is False
def test_less_than(self):
"""Test less than operator (_lt)"""
assert evaluate_condition({"score": 50}, "score_lt", 51) is True
assert evaluate_condition({"score": 50}, "score_lt", 50) is False
assert evaluate_condition({"score": 50}, "score_lt", 49) is False
def test_less_than_or_equal(self):
"""Test less than or equal operator (_lte)"""
assert evaluate_condition({"score": 50}, "score_lte", 51) is True
assert evaluate_condition({"score": 50}, "score_lte", 50) is True
assert evaluate_condition({"score": 50}, "score_lte", 49) is False
def test_between(self):
"""Test between operator (_between)"""
assert evaluate_condition({"level": 5}, "level_between", [1, 10]) is True
assert evaluate_condition({"level": 5}, "level_between", [5, 5]) is True
assert evaluate_condition({"level": 5}, "level_between", [1, 4]) is False
assert evaluate_condition({"level": 5}, "level_between", [6, 10]) is False
def test_contains(self):
"""Test contains operator (_contains) for comma-separated lists"""
assert evaluate_condition({"inventory": "sword,shield,potion"}, "inventory_contains", "sword") is True
assert evaluate_condition({"inventory": "sword,shield,potion"}, "inventory_contains", "axe") is False
assert evaluate_condition({"inventory": "sword"}, "inventory_contains", "sword") is True
assert evaluate_condition({"inventory": ""}, "inventory_contains", "sword") is False
def test_not_contains(self):
"""Test not contains operator (_not_contains)"""
assert evaluate_condition({"inventory": "sword,shield"}, "inventory_not_contains", "axe") is True
assert evaluate_condition({"inventory": "sword,shield"}, "inventory_not_contains", "sword") is False
def test_matches(self):
"""Test regex match operator (_matches)"""
assert evaluate_condition({"name": "Alice"}, "name_matches", r"^[A-Z]") is True
assert evaluate_condition({"name": "alice"}, "name_matches", r"^[A-Z]") is False
assert evaluate_condition({"email": "test@example.com"}, "email_matches", r".*@.*\.com") is True
def test_exists(self):
"""Test existence check operator (_exists)"""
assert evaluate_condition({"has_key": True}, "has_key_exists", True) is True
assert evaluate_condition({"has_key": True}, "has_key_exists", False) is False
assert evaluate_condition({}, "missing_exists", True) is False
assert evaluate_condition({}, "missing_exists", False) is True
def test_not_exists(self):
"""Test non-existence check operator (_not_exists)"""
assert evaluate_condition({}, "missing_not_exists", True) is True
assert evaluate_condition({"has_key": True}, "has_key_not_exists", True) is False
def test_invalid_number_comparison(self):
"""Test numeric comparison with non-numeric values"""
assert evaluate_condition({"value": "text"}, "value_gt", 5) is False
assert evaluate_condition({}, "missing_gte", 5) is False
def test_invalid_between(self):
"""Test between with invalid format"""
assert evaluate_condition({"value": 5}, "value_between", [1]) is False
assert evaluate_condition({"value": 5}, "value_between", "invalid") is False
def test_invalid_regex(self):
"""Test matches with invalid regex"""
assert evaluate_condition({"value": "test"}, "value_matches", "[invalid") is False
class TestCheckConditions:
"""Test multiple condition evaluation (AND logic)"""
def test_empty_conditions(self):
"""Test empty conditions returns True"""
assert check_conditions({}, {}) is True
def test_all_conditions_met(self):
"""Test all conditions must be met"""
metadata = {"score": 100, "level": 5, "inventory": "sword,shield"}
conditions = {
"score_gte": 100,
"level": 5,
"inventory_contains": "sword"
}
assert check_conditions(metadata, conditions) is True
def test_some_conditions_not_met(self):
"""Test fails if any condition not met"""
metadata = {"score": 50, "level": 5}
conditions = {
"score_gte": 100,
"level": 5
}
assert check_conditions(metadata, conditions) is False
def test_mixed_operators(self):
"""Test mix of different operators"""
metadata = {"score": 75, "status": "active", "name": "Alice"}
conditions = {
"score_gte": 50,
"score_lt": 100,
"status_ne": "inactive",
"name_matches": r"^[A-Z]"
}
assert check_conditions(metadata, conditions) is True
class TestFilterContentBlocks:
"""Test conditional content block filtering"""
def test_simple_strings(self):
"""Test that simple strings are always shown"""
blocks = ["Always shown", "Another one"]
context = {"metadata": {}}
result = filter_content_blocks(blocks, {}, context)
assert result == ["Always shown", "Another one"]
def test_conditional_block_shown(self):
"""Test conditional block shown when condition met"""
blocks = [
{"text": "High score!", "show_if": {"score_gte": 50}}
]
metadata = {"score": 100}
context = {"metadata": metadata}
result = filter_content_blocks(blocks, metadata, context)
assert result == ["High score!"]
def test_conditional_block_hidden(self):
"""Test conditional block hidden when condition not met"""
blocks = [
{"text": "High score!", "show_if": {"score_gte": 50}}
]
metadata = {"score": 20}
context = {"metadata": metadata}
result = filter_content_blocks(blocks, metadata, context)
assert result == []
def test_mixed_blocks(self):
"""Test mix of simple strings and conditional blocks"""
blocks = [
"Always shown",
{"text": "High score!", "show_if": {"score_gte": 50}},
{"text": "Low score", "show_if": {"score_lt": 50}},
"Also always shown"
]
metadata = {"score": 75}
context = {"metadata": metadata}
result = filter_content_blocks(blocks, metadata, context)
assert result == ["Always shown", "High score!", "Also always shown"]
def test_template_rendering_in_blocks(self):
"""Test that templates are rendered in filtered blocks"""
blocks = [
"Score: {{metadata.score}}",
{"text": "Level: {{metadata.level}}", "show_if": {"level_gte": 1}}
]
metadata = {"score": 100, "level": 5}
context = {"metadata": metadata}
result = filter_content_blocks(blocks, metadata, context)
assert result == ["Score: 100", "Level: 5"]
def test_empty_blocks(self):
"""Test empty block list"""
result = filter_content_blocks([], {}, {})
assert result == []
class TestResolveConditionalNavigation:
"""Test if/elif/else conditional navigation resolution"""
def test_simple_string(self):
"""Test simple string navigation (pass-through)"""
result = resolve_conditional_navigation("section:step", {})
assert result == "section:step"
def test_if_branch_matches(self):
"""Test if branch when condition matches"""
nav = [
{"if": {"score_gte": 100}, "goto": "expert:challenge"},
{"else": {}, "goto": "beginner:tutorial"}
]
metadata = {"score": 150}
result = resolve_conditional_navigation(nav, metadata)
assert result == "expert:challenge"
def test_elif_branch_matches(self):
"""Test elif branch when if fails but elif matches"""
nav = [
{"if": {"score_gte": 100}, "goto": "expert:challenge"},
{"elif": {"score_gte": 50}, "goto": "intermediate:lesson"},
{"else": {}, "goto": "beginner:tutorial"}
]
metadata = {"score": 75}
result = resolve_conditional_navigation(nav, metadata)
assert result == "intermediate:lesson"
def test_else_branch(self):
"""Test else branch when all conditions fail"""
nav = [
{"if": {"score_gte": 100}, "goto": "expert:challenge"},
{"elif": {"score_gte": 50}, "goto": "intermediate:lesson"},
{"else": {}, "goto": "beginner:tutorial"}
]
metadata = {"score": 20}
result = resolve_conditional_navigation(nav, metadata)
assert result == "beginner:tutorial"
def test_no_match_no_else(self):
"""Test returns None when no conditions match and no else"""
nav = [
{"if": {"score_gte": 100}, "goto": "expert:challenge"},
{"elif": {"score_gte": 50}, "goto": "intermediate:lesson"}
]
metadata = {"score": 20}
result = resolve_conditional_navigation(nav, metadata)
assert result is None
def test_multiple_conditions_in_branch(self):
"""Test branch with multiple conditions (AND logic)"""
nav = [
{"if": {"score_gte": 100, "level_gte": 10}, "goto": "expert:challenge"},
{"else": {}, "goto": "beginner:tutorial"}
]
metadata = {"score": 100, "level": 10}
result = resolve_conditional_navigation(nav, metadata)
assert result == "expert:challenge"
def test_first_matching_branch_wins(self):
"""Test that first matching branch is used"""
nav = [
{"if": {"score_gte": 50}, "goto": "first:path"},
{"elif": {"score_gte": 50}, "goto": "second:path"}
]
metadata = {"score": 75}
result = resolve_conditional_navigation(nav, metadata)
assert result == "first:path"
class TestSelectWeightedRandom:
"""Test weighted random selection"""
def test_weighted_selection(self):
"""Test basic weighted selection (statistical test)"""
options = [
{"value": "common", "weight": 70},
{"value": "rare", "weight": 25},
{"value": "legendary", "weight": 5}
]
# Run multiple times and check distribution is roughly correct
results = [select_weighted_random(options) for _ in range(1000)]
common_count = results.count("common")
rare_count = results.count("rare")
legendary_count = results.count("legendary")
# Allow 10% variance from expected distribution
assert 600 < common_count < 800 # Expected ~700
assert 150 < rare_count < 350 # Expected ~250
assert 0 < legendary_count < 100 # Expected ~50
def test_single_option(self):
"""Test selection with single option"""
options = [{"value": "only_choice", "weight": 100}]
result = select_weighted_random(options)
assert result == "only_choice"
def test_equal_weights(self):
"""Test equal weights distribution"""
options = [
{"value": "a", "weight": 1},
{"value": "b", "weight": 1},
{"value": "c", "weight": 1}
]
results = [select_weighted_random(options) for _ in range(300)]
# Each should appear roughly 100 times (allow variance)
assert 50 < results.count("a") < 150
assert 50 < results.count("b") < 150
assert 50 < results.count("c") < 150
def test_empty_list(self):
"""Test empty options list"""
result = select_weighted_random([])
assert result is None
def test_missing_weight(self):
"""Test option with missing weight defaults to 1"""
options = [
{"value": "a", "weight": 10},
{"value": "b"} # No weight
]
# Should not crash
result = select_weighted_random(options)
assert result in ["a", "b"]
class TestGetProgressiveHint:
"""Test progressive hints retrieval"""
def test_exact_attempt_match(self):
"""Test hint for exact attempt number"""
hints = [
{"attempt": 1, "text": "First hint", "counts_as_attempt": False},
{"attempt": 2, "text": "Second hint", "counts_as_attempt": False},
{"attempt": 3, "text": "Third hint", "counts_as_attempt": False}
]
context = {}
result = get_progressive_hint(hints, 2, context)
assert result == {"text": "Second hint", "counts_as_attempt": False}
def test_no_hint_for_attempt(self):
"""Test returns None when no hint for attempt"""
hints = [
{"attempt": 1, "text": "First hint", "counts_as_attempt": False}
]
result = get_progressive_hint(hints, 2, {})
assert result is None
def test_empty_hints_list(self):
"""Test empty hints list returns None"""
result = get_progressive_hint([], 1, {})
assert result is None
def test_template_rendering_in_hint(self):
"""Test that templates are rendered in hint text"""
hints = [
{"attempt": 1, "text": "Attempt {{current_attempt}} of {{max_attempts}}", "counts_as_attempt": False}
]
context = {"current_attempt": 1, "max_attempts": 3}
result = get_progressive_hint(hints, 1, context)
assert result["text"] == "Attempt 1 of 3"
def test_counts_as_attempt_field(self):
"""Test counts_as_attempt field is preserved"""
hints = [
{"attempt": 1, "text": "Hint", "counts_as_attempt": True}
]
result = get_progressive_hint(hints, 1, {})
assert result["counts_as_attempt"] is True
def test_missing_counts_as_attempt(self):
"""Test missing counts_as_attempt defaults to False"""
hints = [
{"attempt": 1, "text": "Hint"}
]
result = get_progressive_hint(hints, 1, {})
assert result["counts_as_attempt"] is False
class TestCreateTemplateContext:
"""Test template context creation"""
def test_all_fields_present(self):
"""Test all fields are in context"""
metadata = {"score": 100, "level": 5}
context = create_template_context(
metadata=metadata,
current_attempt=2,
max_attempts=3,
current_section="intro",
current_step="welcome",
username="Alice"
)
assert context["metadata"] == metadata
assert context["current_attempt"] == 2
assert context["max_attempts"] == 3
assert context["attempts_remaining"] == 1
assert context["current_section"] == "intro"
assert context["current_step"] == "welcome"
assert context["username"] == "Alice"
def test_attempts_remaining_calculation(self):
"""Test attempts_remaining is calculated correctly"""
context = create_template_context(
metadata={},
current_attempt=1,
max_attempts=3,
current_section="s",
current_step="st",
username="User"
)
assert context["attempts_remaining"] == 2
def test_attempts_remaining_zero(self):
"""Test attempts_remaining doesn't go negative"""
context = create_template_context(
metadata={},
current_attempt=5,
max_attempts=3,
current_section="s",
current_step="st",
username="User"
)
assert context["attempts_remaining"] == 0
def test_default_username(self):
"""Test username defaults"""
context = create_template_context(
metadata={},
current_attempt=1,
max_attempts=3,
current_section="s",
current_step="st"
)
assert context["username"] == "User"
class TestIntegration:
"""Integration tests combining multiple features"""
def test_template_and_conditions_together(self):
"""Test templates work with conditions in content blocks"""
blocks = [
{"text": "Welcome {{metadata.player_name}}!", "show_if": {"player_name_exists": True}},
{"text": "Score: {{metadata.score}}", "show_if": {"score_gte": 0}}
]
metadata = {"player_name": "Alice", "score": 50}
context = create_template_context(
metadata=metadata,
current_attempt=1,
max_attempts=3,
current_section="intro",
current_step="welcome",
username="Alice"
)
# Add exists condition to metadata for testing
metadata["player_name_exists"] = True
result = filter_content_blocks(blocks, metadata, context)
assert "Welcome Alice!" in result
assert "Score: 50" in result
def test_conditional_nav_with_complex_conditions(self):
"""Test conditional navigation with multiple conditions"""
nav = [
{
"if": {"score_gte": 100, "level_gte": 10, "inventory_contains": "key"},
"goto": "secret:room"
},
{
"elif": {"score_gte": 50},
"goto": "intermediate:level"
},
{
"else": {},
"goto": "beginner:start"
}
]
# Test first branch
metadata1 = {"score": 100, "level": 10, "inventory": "sword,key,shield"}
assert resolve_conditional_navigation(nav, metadata1) == "secret:room"
# Test second branch
metadata2 = {"score": 75, "level": 5, "inventory": "sword"}
assert resolve_conditional_navigation(nav, metadata2) == "intermediate:level"
# Test else branch
metadata3 = {"score": 20, "level": 1, "inventory": ""}
assert resolve_conditional_navigation(nav, metadata3) == "beginner:start"
if __name__ == "__main__":
pytest.main([__file__, "-v"])