59 KiB
59 KiB
Technical Document: Choose Your Own Adventure
REVIEW FEEDBACK: I have read my waythrough the document, and the bones are good. The suggested document centralised the code, and created monster/mamoth code. I have left feedback suggesting where I think we can split it out. It also advised the buildin of BFF endpoints, but until we design the UI we can't suggest that, so I have removed all reference. In general, let's push for a bit more separation between our layers, especially the way LLMs are communicated with over HTTP - the code that makes the call, the code that generates the prompts, and the code that requires the LLM result to do domain-y things are all separate roles, and would all change for different reasons. Let's not get too philosophical or atomic, let's stay pragmatic - but having massive methods or loads of methods on a class was definitely a code smell.
This document translates the design doc into a concrete, actionable set of changes for the API. It is intended to be handed directly to an LLM for implementation.
Summary
The Choose Your Own Adventure (CYOA) feature generates interactive story content using Claude. A learner creates an adventure with creative parameters (genre, setting, vibes, protagonist). The API generates the first story entry via the Anthropic API, then translates it via DeepL and produces TTS audio via Gemini. The learner reads the entry and chooses from four options at the end. Each choice triggers the next entry generation. Adventures run for a configurable number of entries (default 6).
All LLM/translation/TTS work runs through the existing in-process worker queue (app/worker.py).
State Machines
Adventure.status
'awaiting_first_entry'
→ first entry pipeline completes successfully → 'active'
→ first entry pipeline errors → 'error'
'active'
→ decision recorded AND entry_count reaches max_entry_count → 'complete'
→ subsequent entry pipeline errors → 'error'
'complete' (terminal)
'error' (terminal — user can see it errored; no auto-retry in MVP)
AdventureEntry.status
'generating'
→ pipeline completes (story_text set, choices saved, translation saved, audio saved) → 'complete'
→ any pipeline step throws an unhandled exception → 'error'
Generation Pipeline
The pipeline runs inside a worker task (enqueued via app.worker.enqueue). The worker processes tasks serially.
First entry (entry_index = 0)
- Create
AdventureEntryrow withstatus='generating',entry_index=0,generated_from_choice_id=None - Build system prompt (see §AnthropicClient)
- Build initial user message from adventure parameters
- Call
AnthropicClient.generate_adventure_entry(...)— returns(raw_text, usage_dict) - Parse raw_text into
(story_text, choices_raw, gamemaster_notes)usingparse_llm_response() - Parse
choices_rawinto list of(label, text)pairs - Update entry: set
story_text,gamemaster_notes,llm_data,status='complete' - Create
AdventureEntryPossibleChoicerows (one per option, typically 4) - Call
DeepLClient.translate(story_text, source_language)→ createAdventureEntryTranslation - Call
GeminiClient.generate_audio(story_text, voice)→ upload to S3 → createAdventureEntryAudio - Call
AnthropicClient.generate_adventure_title_and_description(story_text, ...)→ returns(title, description) - Update adventure: set
title,description,status='active'
On any exception during steps 2–12: set entry.status='error' and adventure.status='error'.
Subsequent entries (entry_index > 0)
Same pipeline except:
generated_from_choice_idis set to the chosenAdventureEntryPossibleChoice.id- Step 3: conversation history is passed to
generate_adventure_entry(see §Conversation History) - If this entry completes AND
entry_index + 1 == adventure.max_entry_count: setadventure.status='complete'; no choices are created for the final entry - Step 11–12 (title/description) is skipped — already set
Conversation History
Each call to AnthropicClient.generate_adventure_entry (after the first) must include the full conversation history so the model maintains narrative continuity. Reconstruct it from the stored entries:
messages = [
{"role": "user", "content": <initial_setup_message>}, # constructed from adventure params
{"role": "assistant", "content": <reconstructed_entry_0_response>},
{"role": "user", "content": "<label of chosen option>"}, # e.g. "2"
{"role": "assistant", "content": <reconstructed_entry_1_response>},
{"role": "user", "content": "<label of chosen option>"},
...
]
Reconstruct an entry's assistant message by re-joining stored fields:
def _reconstruct_entry_response(entry: AdventureEntry, choices: list[AdventureEntryPossibleChoice]) -> str:
options_block = "\n".join(f"{c.label}. {c.text}" for c in sorted(choices, key=lambda c: c.index))
gm_block = entry.gamemaster_notes or "no notes"
return f"{entry.story_text}\n-----\n{options_block}\n-----\n{gm_block}"
The label of the chosen option comes from looking up the AdventureEntryPossibleChoice referenced by generated_from_choice_id of the next entry.
LLM Response Parsing
The LLM is instructed to return three sections separated by \n-----\n. A parser function lives in the service (or a helper module):
def parse_llm_response(text: str) -> tuple[str, list[tuple[str, str]], str]:
"""Returns (story_text, choices, gm_notes).
choices is a list of (label, text) e.g. [("1", "Go into the house"), ...]
Raises ValueError if format is invalid.
"""
parts = text.split("\n-----\n")
if len(parts) < 3:
# Try relaxed split as fallback
parts = text.split("-----\n")
if len(parts) < 3:
raise ValueError(f"LLM response has {len(parts)} section(s); expected 3")
story_text = parts[0].strip()
options_raw = parts[1].strip()
gm_notes = parts[2].strip()
choices: list[tuple[str, str]] = []
for line in options_raw.splitlines():
line = line.strip()
if not line:
continue
# Match "1. Text" or "1) Text"
import re
m = re.match(r'^(\d+)[.)]\s+(.+)$', line)
if m:
choices.append((m.group(1), m.group(2).strip()))
if not choices:
raise ValueError("No choices parsed from LLM response")
return story_text, choices, gm_notes
New Files
app/domain/models/adventure.py
app/domain/services/adventure_service.py
app/routers/api/adventures.py
app/outbound/postgres/entities/adventure_entities.py
app/outbound/postgres/repositories/adventure_repository.py
alembic/versions/20260503_0016_add_choose_your_own_adventure.py
tests/test_adventures.py
Modified files:
app/outbound/anthropic/anthropic_client.py (add 2 methods)
app/routers/api/main.py (register router)
Database Migration
File: alembic/versions/20260503_0016_add_choose_your_own_adventure.py
The choose_your_own_adventure_entry and choose_your_own_adventure_entry_possible_choice tables have a circular FK relationship:
entry.generated_from_choice_id → possible_choice.idpossible_choice.entry_id → entry.id
Resolve this by creating both tables without the circular FK, then adding it with ALTER TABLE.
"""add choose_your_own_adventure tables
Revision ID: 0016
Revises: 0015
Create Date: 2026-05-03
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
def upgrade() -> None:
# 1. Adventure header
op.create_table(
"choose_your_own_adventure",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column(
"user_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column("status", sa.Text(), nullable=False, server_default="awaiting_first_entry"),
sa.Column("language", sa.Text(), nullable=False),
sa.Column("source_language", sa.Text(), nullable=False),
sa.Column("competencies", postgresql.JSONB(), nullable=False, server_default="[]"),
sa.Column("max_entry_count", sa.Integer(), nullable=False, server_default="6"),
sa.Column(
"entry_story_text_target_length",
postgresql.JSONB(),
nullable=False,
server_default='{"min": 700, "max": 800}',
),
sa.Column("title", sa.Text(), nullable=False, server_default="Untitled adventure"),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("plot_summary", sa.Text(), nullable=True),
sa.Column("genres", postgresql.JSONB(), nullable=False, server_default="[]"),
sa.Column("setting", postgresql.JSONB(), nullable=False, server_default="[]"),
sa.Column("vibes", postgresql.JSONB(), nullable=False, server_default="[]"),
sa.Column("protagonist", postgresql.JSONB(), nullable=False, server_default="[]"),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
nullable=False,
server_default=sa.func.now(),
),
sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
)
op.create_index("ix_cyoa_user_id", "choose_your_own_adventure", ["user_id"])
op.create_index("ix_cyoa_status", "choose_your_own_adventure", ["status"])
# 2. Entry table — created WITHOUT the circular FK to possible_choice (added below)
op.create_table(
"choose_your_own_adventure_entry",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column(
"adventure_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("choose_your_own_adventure.id", ondelete="CASCADE"),
nullable=False,
),
# generated_from_choice_id FK added after possible_choice table is created
sa.Column("generated_from_choice_id", postgresql.UUID(as_uuid=True), nullable=True),
sa.Column("status", sa.Text(), nullable=False, server_default="generating"),
sa.Column("entry_index", sa.Integer(), nullable=False),
sa.Column("story_text", sa.Text(), nullable=True),
sa.Column("gamemaster_notes", sa.Text(), nullable=True),
sa.Column("llm_data", postgresql.JSONB(), nullable=True),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
nullable=False,
server_default=sa.func.now(),
),
sa.UniqueConstraint("adventure_id", "entry_index", name="uq_cyoa_entry_adventure_index"),
)
op.create_index("ix_cyoa_entry_adventure_id", "choose_your_own_adventure_entry", ["adventure_id"])
# 3. Possible choices for each entry
op.create_table(
"choose_your_own_adventure_entry_possible_choice",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column(
"entry_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("choose_your_own_adventure_entry.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column("index", sa.Integer(), nullable=False),
sa.Column("label", sa.Text(), nullable=False),
sa.Column("text", sa.Text(), nullable=False),
sa.UniqueConstraint("entry_id", "index", name="uq_cyoa_choice_entry_index"),
)
op.create_index(
"ix_cyoa_choice_entry_id",
"choose_your_own_adventure_entry_possible_choice",
["entry_id"],
)
# 4. Now add the circular FK from entry → possible_choice
op.create_foreign_key(
"fk_cyoa_entry_generated_from_choice",
"choose_your_own_adventure_entry",
"choose_your_own_adventure_entry_possible_choice",
["generated_from_choice_id"],
["id"],
ondelete="SET NULL",
)
# 5. Decision: which choice the user made
op.create_table(
"choose_your_own_adventure_entry_possible_choice_decision",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column(
"choice_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey(
"choose_your_own_adventure_entry_possible_choice.id", ondelete="CASCADE"
),
nullable=False,
),
sa.Column(
"user_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
nullable=False,
server_default=sa.func.now(),
),
)
op.create_index(
"ix_cyoa_decision_choice_id",
"choose_your_own_adventure_entry_possible_choice_decision",
["choice_id"],
)
op.create_index(
"ix_cyoa_decision_user_id",
"choose_your_own_adventure_entry_possible_choice_decision",
["user_id"],
)
# 6. Translation of entry story_text
op.create_table(
"choose_your_own_adventure_entry_translation",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column(
"entry_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("choose_your_own_adventure_entry.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column("component_type", sa.Text(), nullable=False, server_default="story_text"),
sa.Column("target_language", sa.Text(), nullable=False),
sa.Column("translated_text", sa.Text(), nullable=False),
sa.UniqueConstraint(
"entry_id", "component_type", "target_language",
name="uq_cyoa_translation_entry_component_lang",
),
)
op.create_index(
"ix_cyoa_translation_entry_id",
"choose_your_own_adventure_entry_translation",
["entry_id"],
)
# 7. TTS audio for entry story_text
op.create_table(
"choose_your_own_adventure_entry_audio",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column(
"entry_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("choose_your_own_adventure_entry.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column("component_type", sa.Text(), nullable=False, server_default="story_text"),
sa.Column("tts_provider", sa.Text(), nullable=False, server_default="google_gemini"),
sa.Column("tts_options", postgresql.JSONB(), nullable=True),
sa.Column("file_name", sa.Text(), nullable=False),
sa.UniqueConstraint(
"entry_id", "component_type", name="uq_cyoa_audio_entry_component"
),
)
op.create_index(
"ix_cyoa_audio_entry_id",
"choose_your_own_adventure_entry_audio",
["entry_id"],
)
def downgrade() -> None:
op.drop_table("choose_your_own_adventure_entry_audio")
op.drop_table("choose_your_own_adventure_entry_translation")
op.drop_table("choose_your_own_adventure_entry_possible_choice_decision")
op.drop_constraint(
"fk_cyoa_entry_generated_from_choice", "choose_your_own_adventure_entry", type_="foreignkey"
)
op.drop_table("choose_your_own_adventure_entry_possible_choice")
op.drop_table("choose_your_own_adventure_entry")
op.drop_table("choose_your_own_adventure")
ORM Entities
File: app/outbound/postgres/entities/adventure_entities.py
import uuid
from datetime import datetime, timezone
from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, Text, UniqueConstraint
from sqlalchemy.dialects.postgresql import JSONB, UUID
from sqlalchemy.orm import Mapped, mapped_column
from ..database import Base
class AdventureEntity(Base):
__tablename__ = "choose_your_own_adventure"
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
status: Mapped[str] = mapped_column(Text, nullable=False, default="awaiting_first_entry")
language: Mapped[str] = mapped_column(Text, nullable=False)
source_language: Mapped[str] = mapped_column(Text, nullable=False)
competencies: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
max_entry_count: Mapped[int] = mapped_column(Integer, nullable=False, default=6)
entry_story_text_target_length: Mapped[dict] = mapped_column(
JSONB, nullable=False, default=lambda: {"min": 700, "max": 800}
)
title: Mapped[str] = mapped_column(Text, nullable=False, default="Untitled adventure")
description: Mapped[str | None] = mapped_column(Text, nullable=True)
plot_summary: Mapped[str | None] = mapped_column(Text, nullable=True)
genres: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
setting: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
vibes: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
protagonist: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)
)
deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
class AdventureEntryEntity(Base):
__tablename__ = "choose_your_own_adventure_entry"
__table_args__ = (
UniqueConstraint("adventure_id", "entry_index", name="uq_cyoa_entry_adventure_index"),
)
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
adventure_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("choose_your_own_adventure.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
generated_from_choice_id: Mapped[uuid.UUID | None] = mapped_column(
UUID(as_uuid=True),
ForeignKey(
"choose_your_own_adventure_entry_possible_choice.id", ondelete="SET NULL"
),
nullable=True,
)
status: Mapped[str] = mapped_column(Text, nullable=False, default="generating")
entry_index: Mapped[int] = mapped_column(Integer, nullable=False)
story_text: Mapped[str | None] = mapped_column(Text, nullable=True)
gamemaster_notes: Mapped[str | None] = mapped_column(Text, nullable=True)
llm_data: Mapped[dict | None] = mapped_column(JSONB, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)
)
class AdventureEntryPossibleChoiceEntity(Base):
__tablename__ = "choose_your_own_adventure_entry_possible_choice"
__table_args__ = (
UniqueConstraint("entry_id", "index", name="uq_cyoa_choice_entry_index"),
)
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
entry_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("choose_your_own_adventure_entry.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
index: Mapped[int] = mapped_column(Integer, nullable=False)
label: Mapped[str] = mapped_column(Text, nullable=False)
text: Mapped[str] = mapped_column(Text, nullable=False)
class AdventureEntryPossibleChoiceDecisionEntity(Base):
__tablename__ = "choose_your_own_adventure_entry_possible_choice_decision"
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
choice_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey(
"choose_your_own_adventure_entry_possible_choice.id", ondelete="CASCADE"
),
nullable=False,
index=True,
)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)
)
class AdventureEntryTranslationEntity(Base):
__tablename__ = "choose_your_own_adventure_entry_translation"
__table_args__ = (
UniqueConstraint(
"entry_id", "component_type", "target_language",
name="uq_cyoa_translation_entry_component_lang",
),
)
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
entry_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("choose_your_own_adventure_entry.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
component_type: Mapped[str] = mapped_column(Text, nullable=False, default="story_text")
target_language: Mapped[str] = mapped_column(Text, nullable=False)
translated_text: Mapped[str] = mapped_column(Text, nullable=False)
class AdventureEntryAudioEntity(Base):
__tablename__ = "choose_your_own_adventure_entry_audio"
__table_args__ = (
UniqueConstraint("entry_id", "component_type", name="uq_cyoa_audio_entry_component"),
)
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
entry_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("choose_your_own_adventure_entry.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
component_type: Mapped[str] = mapped_column(Text, nullable=False, default="story_text")
tts_provider: Mapped[str] = mapped_column(Text, nullable=False, default="google_gemini")
tts_options: Mapped[dict | None] = mapped_column(JSONB, nullable=True)
file_name: Mapped[str] = mapped_column(Text, nullable=False)
Domain Models
File: app/domain/models/adventure.py
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class Adventure:
id: str
user_id: str
status: str # 'awaiting_first_entry' | 'active' | 'complete' | 'error'
language: str
source_language: str
competencies: list[str]
max_entry_count: int
entry_story_text_target_length: dict # {"min": int, "max": int}
title: str
description: str | None
plot_summary: str | None
genres: list[str]
setting: list[str]
vibes: list[str]
protagonist: list[str]
created_at: datetime
deleted_at: datetime | None
@dataclass
class AdventureEntry:
id: str
adventure_id: str
generated_from_choice_id: str | None
status: str # 'generating' | 'complete' | 'error'
entry_index: int
story_text: str | None
gamemaster_notes: str | None
llm_data: dict | None # {"provider": "anthropic", "model": "...", "input_tokens": int, "output_tokens": int}
created_at: datetime
@dataclass
class AdventureEntryPossibleChoice:
id: str
entry_id: str
index: int
label: str
text: str
@dataclass
class AdventureEntryPossibleChoiceDecision:
id: str
choice_id: str
user_id: str
created_at: datetime
@dataclass
class AdventureEntryTranslation:
id: str
entry_id: str
component_type: str # 'story_text'
target_language: str
translated_text: str
@dataclass
class AdventureEntryAudio:
id: str
entry_id: str
component_type: str # 'story_text'
tts_provider: str
tts_options: dict | None
file_name: str
Repository Layer
REVIEW FEEDBACK: Having a single repository that handle multiple tables gives the repository too many things to do. Break the below suggestions down into a set of smaller repositories, one repository per table.
File: app/outbound/postgres/repositories/adventure_repository.py
Protocol
from typing import Protocol
import uuid
from ....domain.models.adventure import (
Adventure,
AdventureEntry,
AdventureEntryAudio,
AdventureEntryPossibleChoice,
AdventureEntryPossibleChoiceDecision,
AdventureEntryTranslation,
)
class AdventureRepository(Protocol):
async def create_adventure(
self,
user_id: uuid.UUID,
language: str,
source_language: str,
competencies: list[str],
genres: list[str],
setting: list[str],
vibes: list[str],
protagonist: list[str],
max_entry_count: int,
entry_story_text_target_length: dict,
) -> Adventure: ...
async def get_adventure_by_id(self, adventure_id: uuid.UUID) -> Adventure | None: ...
async def list_adventures_for_user(self, user_id: uuid.UUID) -> list[Adventure]: ...
async def update_adventure_status(
self, adventure_id: uuid.UUID, status: str
) -> Adventure: ...
async def update_adventure_title_and_description(
self, adventure_id: uuid.UUID, title: str, description: str
) -> Adventure: ...
async def soft_delete_adventure(self, adventure_id: uuid.UUID) -> Adventure: ...
async def create_entry(
self,
adventure_id: uuid.UUID,
entry_index: int,
generated_from_choice_id: uuid.UUID | None,
) -> AdventureEntry: ...
async def update_entry_content(
self,
entry_id: uuid.UUID,
story_text: str,
gamemaster_notes: str,
llm_data: dict,
status: str,
) -> AdventureEntry: ...
async def update_entry_status(
self, entry_id: uuid.UUID, status: str
) -> AdventureEntry: ...
async def get_entry_by_id(self, entry_id: uuid.UUID) -> AdventureEntry | None: ...
async def list_entries_for_adventure(
self, adventure_id: uuid.UUID
) -> list[AdventureEntry]: ...
async def create_choices(
self,
entry_id: uuid.UUID,
choices: list[tuple[int, str, str]], # (index, label, text)
) -> list[AdventureEntryPossibleChoice]: ...
async def get_choices_for_entry(
self, entry_id: uuid.UUID
) -> list[AdventureEntryPossibleChoice]: ...
async def get_choice_by_id(
self, choice_id: uuid.UUID
) -> AdventureEntryPossibleChoice | None: ...
async def create_decision(
self, choice_id: uuid.UUID, user_id: uuid.UUID
) -> AdventureEntryPossibleChoiceDecision: ...
async def get_decision_for_entry(
self, adventure_id: uuid.UUID, entry_id: uuid.UUID
) -> AdventureEntryPossibleChoiceDecision | None: ...
async def create_translation(
self,
entry_id: uuid.UUID,
component_type: str,
target_language: str,
translated_text: str,
) -> AdventureEntryTranslation: ...
async def get_translation_for_entry(
self, entry_id: uuid.UUID, component_type: str, target_language: str
) -> AdventureEntryTranslation | None: ...
async def create_audio(
self,
entry_id: uuid.UUID,
component_type: str,
tts_provider: str,
tts_options: dict,
file_name: str,
) -> AdventureEntryAudio: ...
async def get_audio_for_entry(
self, entry_id: uuid.UUID, component_type: str
) -> AdventureEntryAudio | None: ...
async def count_complete_entries(self, adventure_id: uuid.UUID) -> int: ...
Implementation: PostgresAdventureRepository
Follow the exact same pattern as PostgresVocabRepository:
- Private
_to_adventure(entity),_to_entry(entity), etc. conversion functions at module level - Class constructor takes
db: AsyncSession - Create operations:
db.add(entity)→db.commit()→db.refresh(entity)→ convert - Read operations:
db.execute(select(...).where(...))→scalar_one_or_none()orscalars().all() - Update operations: load entity → mutate →
db.commit()→db.refresh()→ convert
Key query patterns:
# list_adventures_for_user — exclude soft-deleted, order newest first
select(AdventureEntity)
.where(AdventureEntity.user_id == user_id, AdventureEntity.deleted_at.is_(None))
.order_by(AdventureEntity.created_at.desc())
# list_entries_for_adventure — order by entry_index ascending
select(AdventureEntryEntity)
.where(AdventureEntryEntity.adventure_id == adventure_id)
.order_by(AdventureEntryEntity.entry_index.asc())
# count_complete_entries — for checking if adventure should be marked complete
select(func.count()).select_from(AdventureEntryEntity).where(
AdventureEntryEntity.adventure_id == adventure_id,
AdventureEntryEntity.status == "complete",
)
# get_decision_for_entry — find if user already made a choice on this entry
# Join decision → choice → entry to check entry_id
select(AdventureEntryPossibleChoiceDecisionEntity)
.join(
AdventureEntryPossibleChoiceEntity,
AdventureEntryPossibleChoiceDecisionEntity.choice_id == AdventureEntryPossibleChoiceEntity.id
)
.where(
AdventureEntryPossibleChoiceEntity.entry_id == entry_id,
AdventureEntryPossibleChoiceDecisionEntity.user_id == user_id,
)
AnthropicClient Additions
REVIEW FEEBACK: Can we extract out the generation of the "system prompt" here to be more generic, and not tied specifically to the AnthropicClient. In the ports-and-adapters architecture we should view the AntrhopicClient as a way to interact with an LLM with a given prompt, rather than something which has business/domain decisions. I imagine some system components with a name like adventure_generation_helpers, which can generate both the system prompts, and also collate the previous conversations into one place - and then the LLM Clients (like AnthropicClient) could be passed mroe simply the text system prompt, and a list of dicts for previous messages to generate the next response.
File: app/outbound/anthropic/anthropic_client.py
Add two new methods. Keep the same asyncio.to_thread(_call) pattern.
generate_adventure_entry
async def generate_adventure_entry(
self,
language: str, # e.g. "French"
competencies: list[str], # e.g. ["B1"]
genres: list[str],
setting: list[str],
vibes: list[str],
protagonist: list[str],
max_entry_count: int,
entry_story_text_target_length: dict, # {"min": 700, "max": 800}
conversation_history: list[dict], # [{"role": "user"|"assistant", "content": str}]
model: str = "claude-sonnet-4-6",
) -> tuple[str, dict]:
"""Returns (response_text, usage_dict).
usage_dict: {"provider": "anthropic", "model": str, "input_tokens": int, "output_tokens": int}
"""
System prompt template (adapt from the design doc example):
You are an experienced tabletop game master running a single-player one-shot campaign in a "choose your own adventure" format.
You are helping the player learn {language}. Your writing respects their intelligence, avoids too many clichés, delivers satisfying plot beats, and reads naturally.
The session is {max_entry_count} turns. Each turn: you write a story passage, then offer
4 numbered choices. The player replies with their choice; you continue accordingly.
By turn {max_entry_count} there needs to be a clear end. As the player's choices reveal
their character, weave those details back into the story. Don't railroad them until at
least turn {max_entry_count * 0.5}.
Rules:
- Write entirely in {language} at {competency} level on the CEFR scale. No markdown — plaintext only.
- Your response MUST be in exactly three parts, each separated by a line containing only "-----".
- Part 1: the story entry, {min_length}–{max_length} words, speaking directly to the player.
- Part 2: exactly 4 numbered player options, one per line, labelled "1.", "2.", "3.", "4.".
- Part 3: GM notes to your future self (hidden from the player). If no notes, write "no notes".
- Your first message must establish: who the player is, the setting, and the broad direction.
- No sexual content or graphic violence. Romance, threat, and adventure are fine (12-certificate).
Initial user message (first entry only; constructed in the service):
Please begin the adventure with the following details:
- Genre: {genres_joined}
- Setting: {setting_joined}
- Vibes: {vibes_joined}
- Protagonist: {protagonist_joined}
For subsequent entries, conversation_history already contains the setup message and all prior turns. The method appends it as the messages array.
generate_adventure_title_and_description
async def generate_adventure_title_and_description(
self,
first_entry_text: str,
language: str,
genres: list[str],
model: str = "claude-sonnet-4-6",
) -> tuple[str, str]:
"""Returns (title, description).
Asks the LLM for a short adventure title and a one-sentence description,
based on the first entry. Returns these as a (title, description) tuple.
Instruct the LLM to respond with exactly two lines:
Line 1: the title (plain text, no label, max 60 characters)
Line 2: the description (plain text, max 200 characters)
"""
Service Layer
REVIEW REVIEW This service method has got too big, it's doing too many things. I like that it creates the Adventures and entires, and records decisions (that feels like what it should be doing). Find some natural points to break out, e.g. I suspec that we could isolate the long-runningl logic in the _run_entry_pipeline method (though that's so domain-specific, and is very service-layer in its orchestration/choreographic role, so maybe it does belong here). But at least the LLM response parsing should be moved as separate - as the connector between the service layer and the outbound port/adapter part.
File: app/domain/services/adventure_service.py
import uuid
from sqlalchemy.ext.asyncio import AsyncSession
from ...outbound.anthropic.anthropic_client import AnthropicClient
from ...outbound.deepl.deepl_client import DeepLClient
from ...outbound.gemini.gemini_client import GeminiClient
from ...outbound.postgres.repositories.adventure_repository import AdventureRepository
from ...storage import upload_audio
from ..models.adventure import Adventure, AdventureEntry, AdventureEntryPossibleChoiceDecision
from ... import worker
class AdventureService:
def __init__(
self,
adventure_repo: AdventureRepository,
anthropic_client: AnthropicClient,
deepl_client: DeepLClient,
gemini_client: GeminiClient,
) -> None:
self.adventure_repo = adventure_repo
self.anthropic_client = anthropic_client
self.deepl_client = deepl_client
self.gemini_client = gemini_client
async def create_adventure_for_user(
self,
db: AsyncSession,
user_id: uuid.UUID,
language: str,
source_language: str,
competencies: list[str],
genres: list[str],
setting: list[str],
vibes: list[str],
protagonist: list[str],
max_entry_count: int = 6,
) -> Adventure:
"""Creates the adventure record and enqueues first-entry generation."""
adventure = await self.adventure_repo.create_adventure(
user_id=user_id,
language=language,
source_language=source_language,
competencies=competencies,
genres=genres,
setting=setting,
vibes=vibes,
protagonist=protagonist,
max_entry_count=max_entry_count,
entry_story_text_target_length={"min": 700, "max": 800},
)
entry = await self.adventure_repo.create_entry(
adventure_id=uuid.UUID(adventure.id),
entry_index=0,
generated_from_choice_id=None,
)
await worker.enqueue(
lambda: self._run_entry_pipeline(db, uuid.UUID(adventure.id), uuid.UUID(entry.id))
)
return adventure
async def record_decision_and_generate_next_entry(
self,
db: AsyncSession,
adventure_id: uuid.UUID,
choice_id: uuid.UUID,
user_id: uuid.UUID,
) -> AdventureEntryPossibleChoiceDecision:
"""Validates and records the player's decision, then enqueues the next entry.
Raises ValueError if:
- The adventure does not belong to this user
- The adventure is not in 'active' status
- The choice does not belong to this adventure
- The user has already made a decision on this entry
"""
adventure = await self.adventure_repo.get_adventure_by_id(adventure_id)
if adventure is None or adventure.user_id != str(user_id):
raise ValueError("adventure_not_found")
if adventure.status != "active":
raise ValueError(f"adventure_not_active:{adventure.status}")
choice = await self.adventure_repo.get_choice_by_id(choice_id)
if choice is None:
raise ValueError("choice_not_found")
# Verify this choice belongs to an entry in this adventure
entry = await self.adventure_repo.get_entry_by_id(uuid.UUID(choice.entry_id))
if entry is None or entry.adventure_id != str(adventure_id):
raise ValueError("choice_not_in_adventure")
# Prevent double-deciding on the same entry
existing_decision = await self.adventure_repo.get_decision_for_entry(
adventure_id=adventure_id,
entry_id=uuid.UUID(choice.entry_id),
)
if existing_decision is not None:
raise ValueError("decision_already_made")
decision = await self.adventure_repo.create_decision(
choice_id=choice_id, user_id=user_id
)
next_index = entry.entry_index + 1
next_entry = await self.adventure_repo.create_entry(
adventure_id=adventure_id,
entry_index=next_index,
generated_from_choice_id=choice_id,
)
await worker.enqueue(
lambda: self._run_entry_pipeline(db, adventure_id, uuid.UUID(next_entry.id))
)
return decision
async def _run_entry_pipeline(
self,
db: AsyncSession,
adventure_id: uuid.UUID,
entry_id: uuid.UUID,
) -> None:
"""Full entry generation pipeline. Runs in the worker queue.
On any error, marks the entry and adventure as 'error'.
"""
try:
adventure = await self.adventure_repo.get_adventure_by_id(adventure_id)
assert adventure is not None
all_entries = await self.adventure_repo.list_entries_for_adventure(adventure_id)
current_entry = next(e for e in all_entries if e.id == str(entry_id))
is_first_entry = current_entry.entry_index == 0
is_final_entry = current_entry.entry_index + 1 == adventure.max_entry_count
# Build conversation history for all entries before this one
conversation_history = await self._build_conversation_history(
adventure=adventure,
entries=[e for e in all_entries if e.entry_index < current_entry.entry_index],
)
# LLM generation
from ...languages import SUPPORTED_LANGUAGES
language_name = SUPPORTED_LANGUAGES.get(adventure.language, adventure.language)
raw_text, usage_dict = await self.anthropic_client.generate_adventure_entry(
language=language_name,
competencies=adventure.competencies,
genres=adventure.genres,
setting=adventure.setting,
vibes=adventure.vibes,
protagonist=adventure.protagonist,
max_entry_count=adventure.max_entry_count,
entry_story_text_target_length=adventure.entry_story_text_target_length,
conversation_history=conversation_history,
)
# Parse LLM response
story_text, choices_parsed, gm_notes = parse_llm_response(raw_text)
# Persist entry content
await self.adventure_repo.update_entry_content(
entry_id=entry_id,
story_text=story_text,
gamemaster_notes=gm_notes,
llm_data=usage_dict,
status="complete",
)
# Persist choices — omit for the final entry
if not is_final_entry:
choices_for_db = [
(i, label, text) for i, (label, text) in enumerate(choices_parsed)
]
await self.adventure_repo.create_choices(entry_id=entry_id, choices=choices_for_db)
# Translation
translated = await self.deepl_client.translate(story_text, adventure.source_language)
await self.adventure_repo.create_translation(
entry_id=entry_id,
component_type="story_text",
target_language=adventure.source_language,
translated_text=translated,
)
# TTS audio
voice = self.gemini_client.get_voice_by_language(adventure.language)
wav_bytes = await self.gemini_client.generate_audio(story_text, voice)
audio_key = f"adventure-audio/{entry_id}.wav"
upload_audio(audio_key, wav_bytes)
await self.adventure_repo.create_audio(
entry_id=entry_id,
component_type="story_text",
tts_provider="google_gemini",
tts_options={"voice": voice},
file_name=audio_key,
)
# Update adventure: title (first entry only) and status
if is_first_entry:
title, description = await self.anthropic_client.generate_adventure_title_and_description(
first_entry_text=story_text,
language=language_name,
genres=adventure.genres,
)
await self.adventure_repo.update_adventure_title_and_description(
adventure_id=adventure_id, title=title, description=description
)
new_adventure_status = "complete" if is_final_entry else "active"
await self.adventure_repo.update_adventure_status(
adventure_id=adventure_id, status=new_adventure_status
)
except Exception:
import logging
logging.getLogger(__name__).exception("Entry pipeline failed for entry %s", entry_id)
await self.adventure_repo.update_entry_status(entry_id=entry_id, status="error")
await self.adventure_repo.update_adventure_status(
adventure_id=adventure_id, status="error"
)
async def _build_conversation_history(
self,
adventure: "Adventure",
entries: list["AdventureEntry"],
) -> list[dict]:
"""Reconstruct the full conversation history for the LLM from stored entries.
Returns a list of {role, content} dicts ready to pass to Anthropic.
The first message is always the initial user setup message.
"""
from ...languages import SUPPORTED_LANGUAGES
language_name = SUPPORTED_LANGUAGES.get(adventure.language, adventure.language)
competency = adventure.competencies[0] if adventure.competencies else "B1"
setup_message = (
f"Please begin the adventure with the following details:\n"
f"- Genre: {', '.join(adventure.genres)}\n"
f"- Setting: {', '.join(adventure.setting)}\n"
f"- Vibes: {', '.join(adventure.vibes)}\n"
f"- Protagonist: {', '.join(adventure.protagonist)}"
)
history: list[dict] = [{"role": "user", "content": setup_message}]
for entry in sorted(entries, key=lambda e: e.entry_index):
choices = await self.adventure_repo.get_choices_for_entry(uuid.UUID(entry.id))
assistant_text = _reconstruct_entry_response(entry, choices)
history.append({"role": "assistant", "content": assistant_text})
# Find the user's choice that led to the next entry
# (The next entry's generated_from_choice_id tells us which choice was made)
next_entries = [
e for e in entries if e.entry_index == entry.entry_index + 1
]
if next_entries and next_entries[0].generated_from_choice_id:
chosen_choice_id = uuid.UUID(next_entries[0].generated_from_choice_id)
chosen_choice = next(
(c for c in choices if c.id == str(chosen_choice_id)), None
)
if chosen_choice:
history.append({"role": "user", "content": chosen_choice.label})
return history
def parse_llm_response(text: str) -> tuple[str, list[tuple[str, str]], str]:
"""Parse LLM response into (story_text, choices, gm_notes).
choices is a list of (label, text) e.g. [("1", "Go into the house"), ...]
Raises ValueError if the response format cannot be parsed.
"""
import re
parts = text.split("\n-----\n")
if len(parts) < 3:
parts = text.split("-----\n")
if len(parts) < 3:
raise ValueError(f"LLM response has {len(parts)} section(s); expected 3")
story_text = parts[0].strip()
options_raw = parts[1].strip()
gm_notes = "\n-----\n".join(parts[2:]).strip() # In case gm_notes themselves contain dashes
choices: list[tuple[str, str]] = []
for line in options_raw.splitlines():
line = line.strip()
if not line:
continue
m = re.match(r'^(\d+)[.)]\s+(.+)$', line)
if m:
choices.append((m.group(1), m.group(2).strip()))
if not choices:
raise ValueError("No choices parsed from LLM response options section")
return story_text, choices, gm_notes
def _reconstruct_entry_response(
entry: "AdventureEntry",
choices: list["AdventureEntryPossibleChoice"],
) -> str:
"""Rebuild the original LLM response format from stored entry data."""
options_block = "\n".join(
f"{c.label}. {c.text}" for c in sorted(choices, key=lambda c: c.index)
)
gm_block = entry.gamemaster_notes or "no notes"
return f"{entry.story_text}\n-----\n{options_block}\n-----\n{gm_block}"
API Routes
File: app/routers/api/adventures.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
import uuid
from ...auth import verify_token
from ...outbound.postgres.database import get_db
# ... imports for service, repo, clients
router = APIRouter(prefix="/adventures", tags=["adventures"])
Dependency factory
def _service(db: AsyncSession) -> AdventureService:
from ...config import settings
return AdventureService(
adventure_repo=PostgresAdventureRepository(db),
anthropic_client=AnthropicClient.new(settings.anthropic_api_key),
deepl_client=DeepLClient(settings.deepl_api_key),
gemini_client=GeminiClient(settings.gemini_api_key),
)
POST /api/adventures
Creates a new adventure and enqueues first-entry generation.
Request body:
class CreateAdventureRequest(BaseModel):
language: str # ISO code, e.g. "fr"
source_language: str # ISO code, e.g. "en"
competencies: list[str] # e.g. ["B1"]
genres: list[str]
setting: list[str]
vibes: list[str]
protagonist: list[str]
max_entry_count: int = 6
Response: 201 AdventureResponse
class AdventureResponse(BaseModel):
id: str
user_id: str
status: str
language: str
source_language: str
competencies: list[str]
max_entry_count: int
title: str
description: str | None
genres: list[str]
setting: list[str]
vibes: list[str]
protagonist: list[str]
created_at: str # ISO datetime
Errors: 400 if language is not in SUPPORTED_LANGUAGES or DeepL cannot translate source_language.
Handler:
@router.post("", response_model=AdventureResponse, status_code=201)
async def create_adventure(
body: CreateAdventureRequest,
db: AsyncSession = Depends(get_db),
token_data: dict = Depends(verify_token),
) -> AdventureResponse:
user_id = uuid.UUID(token_data["sub"])
# Validate language support
adventure = await _service(db).create_adventure_for_user(
db=db, user_id=user_id, **body.model_dump()
)
return _to_adventure_response(adventure)
GET /api/adventures
Returns all non-deleted adventures for the authenticated user.
Response: 200 list[AdventureResponse]
GET /api/adventures/{adventure_id}
Returns a single adventure (must belong to the authenticated user).
Response: 200 AdventureResponse
Errors: 404 if not found or belongs to another user.
DELETE /api/adventures/{adventure_id}
Soft-deletes the adventure by setting deleted_at.
Response: 204 No Content
Errors: 404 if not found or belongs to another user.
POST /api/adventures/{adventure_id}/decisions
Records the player's choice and enqueues the next entry.
Request body:
class CreateDecisionRequest(BaseModel):
choice_id: str # UUID of AdventureEntryPossibleChoice
Response: 201 DecisionResponse
class DecisionResponse(BaseModel):
id: str
choice_id: str
user_id: str
created_at: str
Errors:
| Condition | Status |
|---|---|
| Adventure not found or belongs to another user | 404 |
Adventure status is not active |
409 with detail "adventure_not_active" |
| Choice does not belong to this adventure | 404 |
| Decision already made for this entry | 409 with detail "decision_already_made" |
Handler: Parse ValueError from service into appropriate HTTPException.
GET /api/adventures/{adventure_id}/entries
Returns all entries for the adventure (ordered by entry_index ascending).
Response: 200 list[EntryResponse]
class EntryResponse(BaseModel):
id: str
adventure_id: str
generated_from_choice_id: str | None
status: str
entry_index: int
story_text: str | None
created_at: str
Errors: 404 if adventure not found or belongs to another user.
GET /api/adventures/{adventure_id}/entries/{entry_id}
Returns a single entry. Includes choices, translation, and audio URL.
Response: 200 EntryDetailResponse
class ChoiceResponse(BaseModel):
id: str
index: int
label: str
text: str
class EntryDetailResponse(BaseModel):
id: str
adventure_id: str
generated_from_choice_id: str | None
status: str
entry_index: int
story_text: str | None
created_at: str
choices: list[ChoiceResponse]
translation: str | None # translated_text for 'story_text' component
audio_url: str | None # pre-signed S3 URL if audio exists; None otherwise
Audio URL generation: use the existing download_audio / S3 pattern, or generate a pre-signed URL with a 1-hour expiry.
Errors: 404 if adventure or entry not found / unauthorised.
Route Registration
app/routers/api/main.py — add:
from .adventures import router as adventures_router
api_router.include_router(adventures_router)
Configuration
No new environment variables are required. The adventure service uses the existing anthropic_api_key, deepl_api_key, and gemini_api_key from Settings.
Tests
File: tests/test_adventures.py
All tests follow the existing patterns in tests/test_packs.py and tests/test_auth.py:
httpx.Clientagainst the test Docker container- Each test uses fresh data with unique emails
- Register + login to get a Bearer token
Authorization: Bearer <token>header on authenticated requests
The worker queue runs in-process in the test server. Since generation involves real API calls (expensive/slow for tests), stub the generation clients using a test configuration. The recommended approach is:
- Add a
STUB_GENERATION=trueenvironment variable indocker-compose.test.yml - When
STUB_GENERATION=true,AnthropicClient.generate_adventure_entryreturns a hardcoded valid response string instead of calling the real API - Similarly stub
DeepLClient.translateandGeminiClient.generate_audio - This keeps tests fast, isolated, and free of external API dependencies
Stub response for generate_adventure_entry:
Vous vous retrouvez dans une ruelle sombre de Paris. Une silhouette s'approche.
-----
1. Suivez la silhouette
2. Restez dans l'ombre
3. Demandez de l'aide
4. Courez vers la lumière
-----
no notes
Stub response for generate_adventure_title_and_description:
La Nuit Parisienne
Une aventure mystérieuse dans les rues sombres de Paris.
Test helper
import time
def _wait_for_adventure_status(client, adventure_id: str, expected_status: str, timeout: int = 15):
"""Poll until adventure.status matches, or raise TimeoutError."""
deadline = time.time() + timeout
while time.time() < deadline:
resp = client.get(f"/api/adventures/{adventure_id}")
if resp.json()["status"] == expected_status:
return resp.json()
time.sleep(0.5)
raise TimeoutError(f"Adventure {adventure_id} did not reach status '{expected_status}'")
Test 1: Adventure creation and first-entry pipeline
def test_create_adventure_generates_first_entry(user_client):
resp = user_client.post("/api/adventures", json={
"language": "fr",
"source_language": "en",
"competencies": ["B1"],
"genres": ["crime fiction"],
"setting": ["Paris", "city"],
"vibes": ["dark"],
"protagonist": ["male", "late-teens"],
})
assert resp.status_code == 201
adventure_id = resp.json()["id"]
assert resp.json()["status"] == "awaiting_first_entry"
adventure = _wait_for_adventure_status(user_client, adventure_id, "active")
assert adventure["title"] != "Untitled adventure"
assert adventure["description"] is not None
entries_resp = user_client.get(f"/api/adventures/{adventure_id}/entries")
assert entries_resp.status_code == 200
entries = entries_resp.json()
assert len(entries) == 1
assert entries[0]["status"] == "complete"
assert entries[0]["entry_index"] == 0
assert entries[0]["story_text"] is not None
entry_resp = user_client.get(f"/api/adventures/{adventure_id}/entries/{entries[0]['id']}")
assert entry_resp.status_code == 200
detail = entry_resp.json()
assert len(detail["choices"]) == 4
assert detail["translation"] is not None
assert detail["audio_url"] is not None
Test 2: Recording a decision and generating the next entry
def test_record_decision_generates_next_entry(user_client):
resp = user_client.post("/api/adventures", json={...}) # same as above
adventure_id = resp.json()["id"]
_wait_for_adventure_status(user_client, adventure_id, "active")
entries = user_client.get(f"/api/adventures/{adventure_id}/entries").json()
first_entry_id = entries[0]["id"]
first_entry_detail = user_client.get(
f"/api/adventures/{adventure_id}/entries/{first_entry_id}"
).json()
choice_id = first_entry_detail["choices"][0]["id"]
decision_resp = user_client.post(
f"/api/adventures/{adventure_id}/decisions",
json={"choice_id": choice_id},
)
assert decision_resp.status_code == 201
assert decision_resp.json()["choice_id"] == choice_id
_wait_for_adventure_status(user_client, adventure_id, "active")
entries = user_client.get(f"/api/adventures/{adventure_id}/entries").json()
assert len(entries) == 2
second_entry = next(e for e in entries if e["entry_index"] == 1)
assert second_entry["status"] == "complete"
assert second_entry["generated_from_choice_id"] == choice_id
Test 3: Adventure completes after max_entry_count entries
def test_adventure_completes_at_max_entries(user_client):
# Use max_entry_count=2 to keep the test fast (requires only 1 decision)
resp = user_client.post("/api/adventures", json={
...,
"max_entry_count": 2,
})
adventure_id = resp.json()["id"]
_wait_for_adventure_status(user_client, adventure_id, "active")
entries = user_client.get(f"/api/adventures/{adventure_id}/entries").json()
first_entry_detail = user_client.get(
f"/api/adventures/{adventure_id}/entries/{entries[0]['id']}"
).json()
choice_id = first_entry_detail["choices"][0]["id"]
user_client.post(f"/api/adventures/{adventure_id}/decisions", json={"choice_id": choice_id})
adventure = _wait_for_adventure_status(user_client, adventure_id, "complete")
assert adventure["status"] == "complete"
# Final entry should have no choices
entries = user_client.get(f"/api/adventures/{adventure_id}/entries").json()
assert len(entries) == 2
final_entry = user_client.get(
f"/api/adventures/{adventure_id}/entries/{entries[1]['id']}"
).json()
assert final_entry["choices"] == []
# Decision on a complete adventure returns 409
resp = user_client.post(
f"/api/adventures/{adventure_id}/decisions",
json={"choice_id": choice_id},
)
assert resp.status_code == 409
assert "not_active" in resp.json()["detail"]
Test 4: Double-decision prevention
def test_cannot_make_second_decision_on_same_entry(user_client):
resp = user_client.post("/api/adventures", json={...})
adventure_id = resp.json()["id"]
_wait_for_adventure_status(user_client, adventure_id, "active")
entries = user_client.get(f"/api/adventures/{adventure_id}/entries").json()
first_entry_detail = user_client.get(
f"/api/adventures/{adventure_id}/entries/{entries[0]['id']}"
).json()
choice_id = first_entry_detail["choices"][0]["id"]
other_choice_id = first_entry_detail["choices"][1]["id"]
user_client.post(
f"/api/adventures/{adventure_id}/decisions", json={"choice_id": choice_id}
)
# Wait briefly for state to settle (decision is sync; generation is async)
resp = user_client.post(
f"/api/adventures/{adventure_id}/decisions", json={"choice_id": other_choice_id}
)
assert resp.status_code == 409
assert "decision_already_made" in resp.json()["detail"]
Test 5: User isolation
def test_user_cannot_access_another_users_adventure(user_client, second_user_client):
resp = user_client.post("/api/adventures", json={...})
adventure_id = resp.json()["id"]
# Second user cannot GET it
resp = second_user_client.get(f"/api/adventures/{adventure_id}")
assert resp.status_code == 404
# Second user cannot POST a decision on it
resp = second_user_client.post(
f"/api/adventures/{adventure_id}/decisions", json={"choice_id": str(uuid.uuid4())}
)
assert resp.status_code == 404
Unit test: LLM response parser
These run without Docker and test parsing edge cases directly:
# tests/test_adventure_parser.py (or inside test_adventures.py)
from app.domain.services.adventure_service import parse_llm_response
def test_parse_valid_three_section_response():
text = "Story text here.\n-----\n1. Option one\n2. Option two\n3. Option three\n4. Option four\n-----\nno notes"
story, choices, notes = parse_llm_response(text)
assert story == "Story text here."
assert len(choices) == 4
assert choices[0] == ("1", "Option one")
assert notes == "no notes"
def test_parse_with_period_delimiters():
text = "Story.\n-----\n1) First\n2) Second\n3) Third\n4) Fourth\n-----\nGM note here."
story, choices, notes = parse_llm_response(text)
assert len(choices) == 4
assert choices[2] == ("3", "Third")
def test_parse_missing_sections_raises():
import pytest
with pytest.raises(ValueError, match="section"):
parse_llm_response("Only one section here, no separators.")