File size: 3,116 Bytes
3ee7cb0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | """End-to-end test for Phase 3: Session Intelligence."""
import uuid
from app.services.retrieval import load_games_dataset, normalize_game_record, retrieve_examples
from app.services.generator import generate_game
from app.services.validator import validate_game, repair_game
from app.services.tracing import log_event, load_events
from app.services.journal import create_journal_entry, save_journal_entry, summarize_journal, load_journal_entries
from app.services.scoring import compute_scores
from app.services.story import build_story_packet, generate_story
# 1. Load dataset
raw = load_games_dataset("app/data/games_dataset.json")
records = [normalize_game_record(r) for r in raw]
print(f"✓ Loaded {len(records)} records")
# 2. Config
config = {
"game_type": "scavenger_hunt",
"city": "Paris",
"area": "Le Marais",
"location_type": "mixed",
"duration_minutes": 60,
"num_players": 4,
"difficulty": "medium",
"age_group": "adults",
"energy_level": "medium",
"photo_enabled": True,
}
# 3. Retrieve + Generate
retrieved = retrieve_examples(config, records, k=3)
game = generate_game(config, retrieved)
print(f"✓ Generated: {game['title']} with {len(game['tasks'])} tasks")
# 4. Validate
is_valid, failures = validate_game(game, config)
status = "PASS" if is_valid else f"FAIL ({len(failures)} issues)"
print(f"✓ Validation: {status}")
# 5. Session
session_id = f"test-{uuid.uuid4().hex[:8]}"
# 6. Log events
for t in game["tasks"]:
log_event(session_id, "task_revealed", {"task_id": t["task_id"], "title": t["title"]})
log_event(session_id, "task_completed", {"task_id": "t1", "summary": "Completed first task"}, team_id="team-a")
log_event(session_id, "task_completed", {"task_id": "t2", "summary": "Completed second task"}, team_id="team-a")
log_event(session_id, "hint_used", {"task_id": "t2", "summary": "Used hint"}, team_id="team-a")
log_event(session_id, "task_skipped", {"task_id": "t3", "summary": "Skipped"}, team_id="team-a")
events = load_events(session_id)
print(f"✓ Logged {len(events)} events")
# 7. Journal
entry = create_journal_entry(
transcript="We found the mural near the canal, it was incredible! The whole team cheered.",
session_id=session_id,
team_id="team-a",
task_id="t1",
location_note="Canal area",
)
summary = summarize_journal(entry["transcript"], task_id="t1")
entry.update(summary)
save_journal_entry(entry)
journals = load_journal_entries(session_id)
print(f"✓ Journal mood={entry['mood']}, story_value={entry['story_value']}")
# 8. Score
scores = compute_scores(events, game)
print(f"✓ Winner: {scores['winner']}")
for s in scores["team_scores"]:
print(f" - {s['team_id']}: {s['points']} pts ({s['completed_tasks']}/{s['total_tasks']} tasks)")
# 9. Story
packet = build_story_packet(game, events, scores, journals)
story = generate_story(packet, session_id=session_id)
print(f"✓ Story recap: {len(story['short_recap'])} chars short, {len(story['long_summary'])} chars long")
print()
print("=== SHORT RECAP ===")
print(story["short_recap"])
print()
print("=== PIPELINE COMPLETE ✓ ===")
|