diff --git a/apps/api/src/lunaris_api/app.py b/apps/api/src/lunaris_api/app.py index 76dc2950..35c64f58 100644 --- a/apps/api/src/lunaris_api/app.py +++ b/apps/api/src/lunaris_api/app.py @@ -27,6 +27,7 @@ get_video_storage, ) from .live import router as live_router +from .live.session import router as live_session_router from .routers import ( activity, admin_users, @@ -133,6 +134,7 @@ def _register_routers(app: FastAPI) -> None: app.include_router(courses.router) # Live's region, mounted as a whole; Studio's app never names lunaris_live itself. app.include_router(live_router) + app.include_router(live_session_router) app.include_router(briefs.router) app.include_router(runs.router) app.include_router(bridge.router) diff --git a/apps/api/src/lunaris_api/config.py b/apps/api/src/lunaris_api/config.py index 3c64705e..9004cadd 100644 --- a/apps/api/src/lunaris_api/config.py +++ b/apps/api/src/lunaris_api/config.py @@ -68,6 +68,17 @@ class Settings: live_compile_max_concurrent: int = 1 live_extend_daily_cap: int = 50 live_graph_budget_usd: float = 10.0 + #: How long a Live session runs before the director closes it (plan §6: 25-40 minutes). A + #: bound on the learner's wall clock, not on turns — a session of long, slow turns would + #: otherwise run for hours. + live_session_budget_s: float = 1800.0 + #: Lunaris Live session admission. A session is bounded by its clock already, so only the + #: *opening* is rationed per owner per day — a runaway starts by opening sessions in a loop, and + #: capping turns would end sittings that were going well for a reason nobody could explain. + #: ``live_session_budget_usd`` is the ceiling on one session's whole spend, read from the + #: ledger's rollup: a runaway guard, not a ration — 0 turns it off. + live_session_daily_cap: int = 20 + live_session_budget_usd: float = 2.0 device_bridge_liveness_s: float = _DEFAULT_BRIDGE_LIMITS.liveness_s device_bridge_completion_timeout_s: float = _DEFAULT_BRIDGE_LIMITS.completion_timeout_s # The explainer-video operator kill-switch (plan §3.0 item 5). Default OFF — fail-closed, so a @@ -164,6 +175,9 @@ def get_settings() -> Settings: live_compile_max_concurrent=_env_int("LUNARIS_LIVE_COMPILE_MAX_CONCURRENT", default=1), live_extend_daily_cap=_env_int("LUNARIS_LIVE_EXTEND_DAILY_CAP", default=50), live_graph_budget_usd=_env_float("LUNARIS_LIVE_GRAPH_BUDGET_USD", default=10.0), + live_session_budget_s=_env_float("LUNARIS_LIVE_SESSION_BUDGET_S", default=1800.0), + live_session_daily_cap=_env_int("LUNARIS_LIVE_SESSION_DAILY_CAP", default=20), + live_session_budget_usd=_env_float("LUNARIS_LIVE_SESSION_BUDGET_USD", default=2.0), device_bridge_liveness_s=_env_float( "LUNARIS_DEVICE_BRIDGE_LIVENESS_S", default=_DEFAULT_BRIDGE_LIMITS.liveness_s ), diff --git a/apps/api/src/lunaris_api/live/dependencies.py b/apps/api/src/lunaris_api/live/dependencies.py index 769de12e..c5687469 100644 --- a/apps/api/src/lunaris_api/live/dependencies.py +++ b/apps/api/src/lunaris_api/live/dependencies.py @@ -25,9 +25,30 @@ #: Decomposition decides what the course is made of and every later phase reads it, so it runs on #: the strong tier, not the bulk one. Same env knob Studio's tiers use, so a tenant's model choice #: covers both products. (Splitting decomposition from spec authoring across two tiers is a -#: latency/cost lever T4 owns; one tier keeps this honest until there is a measurement to tune to.) +#: latency/cost lever; one tier keeps this honest until there is a measurement to tune to.) _DEFAULT_MODEL = "claude-opus-4-8" + +def resolve_strong_model() -> str: + """The model Live's quality surfaces run on — the compiler's decomposition and the tutor. + + One function rather than a constant per composition root, so a tenant's model choice cannot end + up meaning two different things inside one product: a session taught by a weaker model than the + map it walks would be a difference nobody configured. + """ + return resolve_config("LUNARIS_MODEL_STRONG") or _DEFAULT_MODEL + + +#: Grading one answer against one explicit do-statement is a classification, not a quality surface, +#: so it runs on the bulk tier — the same knob Studio's workers read. +_DEFAULT_WORKER_MODEL = "claude-haiku-4-5-20251001" + + +def resolve_worker_model() -> str: + """The model Live's classification surfaces run on (A1).""" + return resolve_config("LUNARIS_MODEL_WORKER") or _DEFAULT_WORKER_MODEL + + # One durable store per process, same lazy-client rationale as Studio's stores: the service-role # client is built on first write, so the singleton needs no creds and no network until then. _supabase_graph_store = SupabaseGraphStore() @@ -37,8 +58,13 @@ _memory_graph_store = MemoryGraphStore() -def _resolve_graph_store(settings: Settings) -> IGraphStore: - """Durable where Supabase is configured, in-process otherwise (offline dev and the suite).""" +def resolve_graph_store(settings: Settings) -> IGraphStore: + """Durable where Supabase is configured, in-process otherwise (offline dev and the suite). + + Public because the session plane composes against the *same* store: a session reading from a + different store than the compiler wrote to would find no maps, and it would read as a data + problem rather than the wiring one it is. + """ return _supabase_graph_store if settings.has_supabase else _memory_graph_store @@ -51,10 +77,7 @@ def _resolve_compiler(settings: Settings) -> IGraphCompiler: """ if settings.pipeline == "stub": return StubGraphCompiler() - return ClaudeGraphCompiler( - resolve_config("LUNARIS_MODEL_STRONG") or _DEFAULT_MODEL, - deadline_s=settings.live_compile_deadline_s, - ) + return ClaudeGraphCompiler(resolve_strong_model(), deadline_s=settings.live_compile_deadline_s) @lru_cache @@ -100,7 +123,7 @@ def get_live_graph_service( """ return LiveGraphService( _resolve_compiler(settings), - _resolve_graph_store(settings), + resolve_graph_store(settings), cost_event_store=cost_event_store, subject_cost_store=subject_cost_store, credential_resolver=get_live_credential_resolver(settings), diff --git a/apps/api/src/lunaris_api/live/graph_throttle.py b/apps/api/src/lunaris_api/live/graph_throttle.py index c546e627..dcabeab6 100644 --- a/apps/api/src/lunaris_api/live/graph_throttle.py +++ b/apps/api/src/lunaris_api/live/graph_throttle.py @@ -24,22 +24,7 @@ from datetime import datetime from ..daily_counter import DailyCounter - - -class LiveWorkRefusedError(Exception): - """Base for work Live declines to start. - - Subclasses set two class attributes the routers read so EVERY refusal maps the same way (one - ``except`` clause, no per-reason branching): ``status_code`` and ``detail``, the learner-facing - sentence. The base seeds the exception message from ``detail`` so logs carry the reason rather - than a blank ``SomeError:``. - """ - - status_code: int = 429 - detail: str = "Live can't take that on right now." - - def __init__(self, detail: str | None = None) -> None: - super().__init__(detail or self.detail) +from .work_refused import LiveWorkRefusedError class LiveCompileBusyError(LiveWorkRefusedError): diff --git a/apps/api/src/lunaris_api/live/router.py b/apps/api/src/lunaris_api/live/router.py index f5fe10cf..d5111a1a 100644 --- a/apps/api/src/lunaris_api/live/router.py +++ b/apps/api/src/lunaris_api/live/router.py @@ -14,8 +14,8 @@ from ..dependencies import OptionalUserIdDep from .dependencies import LiveGraphServiceDep -from .graph_throttle import LiveWorkRefusedError from .schemas import CompileFailure, LiveGraphExtendRequest, LiveGraphRequest +from .work_refused import LiveWorkRefusedError logger = structlog.get_logger() diff --git a/apps/api/src/lunaris_api/live/session/__init__.py b/apps/api/src/lunaris_api/live/session/__init__.py new file mode 100644 index 00000000..5bc0c2e6 --- /dev/null +++ b/apps/api/src/lunaris_api/live/session/__init__.py @@ -0,0 +1,3 @@ +from .router import router + +__all__ = ["router"] diff --git a/apps/api/src/lunaris_api/live/session/dependencies.py b/apps/api/src/lunaris_api/live/session/dependencies.py new file mode 100644 index 00000000..e6cc870a --- /dev/null +++ b/apps/api/src/lunaris_api/live/session/dependencies.py @@ -0,0 +1,125 @@ +from functools import lru_cache +from typing import Annotated + +from fastapi import Depends +from lunaris_live.session import ( + ClaudeGrader, + ClaudeTutor, + IGrader, + IKnowledgeStore, + ISessionStore, + ITutor, + MemoryKnowledgeStore, + MemorySessionStore, + StubGrader, + StubTutor, + SupabaseKnowledgeStore, + SupabaseSessionStore, +) + +from ...config import Settings, get_settings +from ...dependencies import CostEventStoreDep, SubjectCostStoreDep +from ..dependencies import ( + get_live_credential_resolver, + resolve_graph_store, + resolve_strong_model, + resolve_worker_model, +) +from .service import LiveSessionService +from .throttle import LiveSessionThrottle + +# One durable store per process — same lazy-client rationale as the graph store: the service-role +# client is built on first write, so the singleton needs no creds and no network until then. +_supabase_session_store = SupabaseSessionStore() +_supabase_knowledge_store = SupabaseKnowledgeStore() + +# The in-memory fallbacks MUST be singletons: opening a session and the next turn of it are +# separate requests, so a per-request store would lose the session — and the learner's beliefs — +# between them. +_memory_session_store = MemorySessionStore() +_memory_knowledge_store = MemoryKnowledgeStore() + + +def _resolve_session_store(settings: Settings) -> ISessionStore: + """Durable where Supabase is configured, in-process otherwise (offline dev and the suite).""" + return _supabase_session_store if settings.has_supabase else _memory_session_store + + +def _resolve_knowledge_store(settings: Settings) -> IKnowledgeStore: + """Durable where Supabase is configured, in-process otherwise (offline dev and the suite).""" + return _supabase_knowledge_store if settings.has_supabase else _memory_knowledge_store + + +def get_live_tutor(settings: Annotated[Settings, Depends(get_settings)]) -> ITutor: + """The model-backed tutor, or the deterministic one under ``LUNARIS_PIPELINE=stub``. + + A dependency in its own right rather than something the service composes privately, because it + is the collaborator most worth substituting: the failure that matters most here is a tutor that + cannot speak, and a test can only stage that by putting a silent one in its place. + + Teaching runs on the strong tier (A1) — the same tier the map was compiled on. It is the + quality surface of the whole product, and a session taught by a cheaper model than the map it + walks would be a difference no one chose. + """ + return StubTutor() if settings.pipeline == "stub" else ClaudeTutor(resolve_strong_model()) + + +def get_live_grader(settings: Annotated[Settings, Depends(get_settings)]) -> IGrader: + """The model-backed grader, or the deterministic one under ``LUNARIS_PIPELINE=stub``. + + A dependency of its own for the same reason the tutor is: a grader that cannot answer is the + failure worth staging in a test, and an answer wrongly scored is the mistake that compounds — + every verdict it gets wrong is written into a belief the director acts on for the rest of the + session. + + Runs on the worker tier (A1): teaching is the quality surface, judging one answer against one + explicit do-statement is a classification. + """ + return StubGrader() if settings.pipeline == "stub" else ClaudeGrader(resolve_worker_model()) + + +@lru_cache +def _get_live_session_throttle(settings: Settings) -> LiveSessionThrottle: + """The process-wide session throttle for these settings. + + Cached on the frozen ``Settings`` value because the per-day counts have to be shared across + requests: the service is built per request, and a per-request throttle would never see the + openings it is supposed to be counting. Keyed on the whole ``Settings`` (not just the Live + fields) so it cannot drift as the config surface grows; tests reset it via + ``_get_live_session_throttle.cache_clear()``. + """ + return LiveSessionThrottle(open_daily_cap=settings.live_session_daily_cap) + + +def get_live_session_service( + settings: Annotated[Settings, Depends(get_settings)], + tutor: Annotated[ITutor, Depends(get_live_tutor)], + grader: Annotated[IGrader, Depends(get_live_grader)], + cost_event_store: CostEventStoreDep, + subject_cost_store: SubjectCostStoreDep, +) -> LiveSessionService: + """Live's session plane as a request dependency. + + Takes the *same* graph store the compile plane writes to rather than composing a second one — + a session that read from a different store than the compiler wrote to would find no maps at all, + and it would look like a data problem rather than the wiring one it is. + """ + return LiveSessionService( + resolve_graph_store(settings), + _resolve_session_store(settings), + knowledge=_resolve_knowledge_store(settings), + tutor=tutor, + grader=grader, + session_budget_s=settings.live_session_budget_s, + # The ledger and the tenant's keys come from Studio's composition root: Live is a second + # product, not a second platform, so a tenant's spend and a tenant's key are the same ones + # either way. What differs is only the subject a cost is filed under (``LIVE_SESSION``). + cost_event_store=cost_event_store, + subject_cost_store=subject_cost_store, + credential_resolver=get_live_credential_resolver(settings), + throttle=_get_live_session_throttle(settings), + session_budget_usd=settings.live_session_budget_usd, + ) + + +LiveSessionServiceDep = Annotated[LiveSessionService, Depends(get_live_session_service)] diff --git a/apps/api/src/lunaris_api/live/session/router.py b/apps/api/src/lunaris_api/live/session/router.py new file mode 100644 index 00000000..331c8946 --- /dev/null +++ b/apps/api/src/lunaris_api/live/session/router.py @@ -0,0 +1,191 @@ +from uuid import uuid4 + +import structlog +from fastapi import APIRouter, HTTPException, Response, status +from lunaris_live.session import ( + GraderUnavailableError, + Session, + SessionClosedError, + SessionFormatError, + StaleAnswerError, + TutorUnavailableError, +) +from lunaris_runtime.persistence import PersistenceError + +from ...dependencies import OptionalUserIdDep +from ..work_refused import LiveWorkRefusedError +from .dependencies import LiveSessionServiceDep +from .schemas import AnswerRequest, SessionStartRequest + +logger = structlog.get_logger() + +router = APIRouter(prefix="/api/live/sessions", tags=["live"]) + +_UNAVAILABLE = "Live is having trouble reaching its storage. Try again shortly." + +#: The tutor being down is a different outage from storage being down, and a learner can tell: one +#: means their session did not open, the other means it may not have been saved. Both end, so both +#: are worth retrying — which is why this is a 503 rather than a session opened with nothing in it. +_TUTOR_UNAVAILABLE = "Live's tutor could not reach its model. Try again shortly." + +#: The grader is a different outage from the tutor, and the learner's next step differs: their +#: answer was not scored, so nothing they said has been lost or held against them. +_GRADER_UNAVAILABLE = "Live could not score that answer just now. Try sending it again." + +#: A row this build cannot parse never becomes readable by waiting, so it must not read as an +#: outage: "try again" on a permanently unreadable session is an invitation to reload forever. +_UNREADABLE = "This session was saved in a format Live can no longer read." + + +@router.post("", response_model=Session, status_code=status.HTTP_201_CREATED) +async def start_session( + payload: SessionStartRequest, + service: LiveSessionServiceDep, + response: Response, + owner_id: OptionalUserIdDep, +) -> Session: + """Open a session on a compiled map and hand back its first turn. + + Answers 201 with the session already teaching rather than an empty shell the surface then has to + poll: a session that opens with nothing to show is a loading spinner with a database row behind + it. ``X-Session-Id`` rides the response so a learner reporting "it went wrong" can name the + session across every layer's logs. + """ + # Minted here, and put on the response before the work: a header set only after success is + # absent from exactly the failures somebody needs to report. Every raise below carries it + # explicitly, because raising an HTTPException discards the response object built here. + session_id = uuid4().hex + response.headers["X-Session-Id"] = session_id + correlated = {"X-Session-Id": session_id} + try: + return await service.start(payload.graph_id, session_id=session_id, owner_id=owner_id) + except FileNotFoundError as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Map not found", headers=correlated + ) from exc + except Exception as exc: + # Logged with the traceback because nothing below the router does — the stores stay silent + # and ``guard`` only translates. Without this an outage is a bare 500 with no way to tell + # what broke, on the path where a learner has just lost a session before it began. + logger.warning("live.session.start_failed", session_id=session_id, exc_info=True) + if (translated := _translate(exc, correlated)) is not None: + raise translated from exc + raise + + +@router.post("/{session_id}/turns", response_model=Session) +async def answer_turn( + session_id: str, + payload: AnswerRequest, + service: LiveSessionServiceDep, + response: Response, + owner_id: OptionalUserIdDep, +) -> Session: + """Answer the criterion the last turn staged, and get the session back with its next turn. + + The whole session comes back rather than just the new turn: the answered turn changes too — it + gains the learner's words and the verdict on them — and a surface patching two shapes together + is a surface that can disagree with the row behind it. + """ + correlated = {"X-Session-Id": session_id} + try: + session = await service.answer( + session_id, payload.answer, answering_seq=payload.answering_seq, owner_id=owner_id + ) + except FileNotFoundError as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Session not found", headers=correlated + ) from exc + except Exception as exc: + logger.warning("live.session.answer_failed", session_id=session_id, exc_info=True) + if (translated := _translate(exc, correlated)) is not None: + raise translated from exc + raise + response.headers["X-Session-Id"] = session_id + return session + + +@router.get("/{session_id}", response_model=Session) +async def read_session( + session_id: str, + service: LiveSessionServiceDep, + response: Response, + owner_id: OptionalUserIdDep, +) -> Session: + """Re-read a session, so a reloaded tab lands back where the learner was (U2). + + Another learner's session is 404, not 403 — a session's existence is itself owner-scoped + information, the same posture Phase 1 took for graphs. + """ + correlated = {"X-Session-Id": session_id} + try: + session = await service.load(session_id, owner_id=owner_id) + except FileNotFoundError as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Session not found", headers=correlated + ) from exc + except Exception as exc: + logger.warning("live.session.read_failed", session_id=session_id, exc_info=True) + if (translated := _translate(exc, correlated)) is not None: + raise translated from exc + raise + response.headers["X-Session-Id"] = session_id + return session + + +def _translate(exc: Exception, correlated: dict[str, str]) -> HTTPException | None: + """The HTTP answer to a failure a learner could plausibly act on, or ``None`` to let it fly. + + One place, because every entry point fails in the same ways and drifting apart would mean a + learner learning what "try again" means from whichever endpoint they hit first. Ordered + deliberately: ``SessionFormatError`` IS a ``PersistenceError``, and reading as a retryable + outage is exactly the mistake it exists to prevent. + """ + if isinstance(exc, SessionFormatError): + return HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=_UNREADABLE, + headers=correlated, + ) + if isinstance(exc, LiveWorkRefusedError): + # Every refusal maps the same way, from one place: the throttle and the budget each carry + # their own status and the sentence a learner reads, so a new one cannot arrive with no + # words or the wrong code. + return HTTPException(status_code=exc.status_code, detail=exc.detail, headers=correlated) + if isinstance(exc, StaleAnswerError): + # The same 409 family as a closed session, and for the same reason: the request is well + # formed and the learner did nothing wrong — the question they answered is simply not the + # one in front of them any more. Re-reading the session is the recovery, not retrying. + return HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="That question has already been answered. Reload to catch up.", + headers=correlated, + ) + if isinstance(exc, SessionClosedError): + # The request is well formed and the session is real; it is the session's *state* that + # refuses. A learner with a stale tab should be told it ended, not that they did something + # wrong and not that it might work on a retry. + return HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="This session has already ended.", + headers=correlated, + ) + if isinstance(exc, GraderUnavailableError): + return HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=_GRADER_UNAVAILABLE, + headers=correlated, + ) + if isinstance(exc, TutorUnavailableError): + return HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=_TUTOR_UNAVAILABLE, + headers=correlated, + ) + if isinstance(exc, PersistenceError): + return HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=_UNAVAILABLE, + headers=correlated, + ) + return None diff --git a/apps/api/src/lunaris_api/live/session/schemas.py b/apps/api/src/lunaris_api/live/session/schemas.py new file mode 100644 index 00000000..1cddaf68 --- /dev/null +++ b/apps/api/src/lunaris_api/live/session/schemas.py @@ -0,0 +1,41 @@ +from lunaris_live.session import MAX_ANSWER_CHARS +from pydantic import BaseModel, ConfigDict, Field +from pydantic.alias_generators import to_camel + + +class SessionStartRequest(BaseModel): + """Open a session on a map the learner already has. + + The graph id and nothing else: everything the session needs to teach is already on the map, and + a request carrying teaching preferences would be configuring the tutor at the door rather than + letting the director learn them from what the learner does (P2c's placement interview is where + priors come from). + """ + + model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True) + + graph_id: str = Field(min_length=1, max_length=100) + + +class AnswerRequest(BaseModel): + """What the learner said in reply to the criterion the last turn staged. + + Bounded at both ends at the trust boundary. Empty is refused rather than graded as a miss: an + empty POST is a client bug, and scoring it would lower a belief — and so change what the + director teaches next — on the strength of a stray keystroke. The ceiling is generous enough for + somebody explaining a concept properly and small enough that a pasted chapter never becomes a + model prompt and a stored row. + """ + + # Stripped BEFORE the length check, or " " passes it and reaches the grader as "" — which the + # stub scores NOT_MET, lowering a belief and changing what the director teaches next on the + # strength of one stray keystroke. The bound is the domain's, imported rather than restated. + model_config = ConfigDict( + alias_generator=to_camel, populate_by_name=True, str_strip_whitespace=True + ) + + answer: str = Field(min_length=1, max_length=MAX_ANSWER_CHARS) + #: The turn the learner was looking at. Required, not inferred: a double-submit would otherwise + #: be graded against the question that replaced it, and the words would go into the record under + #: a criterion they were never written for. + answering_seq: int = Field(ge=1) diff --git a/apps/api/src/lunaris_api/live/session/service.py b/apps/api/src/lunaris_api/live/session/service.py new file mode 100644 index 00000000..721e61c5 --- /dev/null +++ b/apps/api/src/lunaris_api/live/session/service.py @@ -0,0 +1,344 @@ +import asyncio +from collections.abc import Mapping +from contextlib import AbstractContextManager, nullcontext +from datetime import UTC, datetime +from uuid import uuid4 + +import structlog +from lunaris_live.graph import ConceptGraph, IGraphStore +from lunaris_live.session import ( + IGrader, + IKnowledgeStore, + ISessionStore, + ITutor, + LearnerModel, + Session, + SessionClock, + TurnOutcome, + open_session, + take_turn, +) +from lunaris_runtime.credentials import CredentialResolver, run_credentials +from lunaris_runtime.logging import bind_request_id, bind_run_id +from lunaris_runtime.metering import ( + CostScope, + drain_cost_scope, + enter_cost_scope, + make_cost_scope, +) +from lunaris_runtime.persistence import ICostEventStore, ISubjectCostStore, PersistenceError +from lunaris_runtime.schema import CostSubjectType + +from .throttle import LiveSessionBudgetExhaustedError, LiveSessionThrottle + +logger = structlog.get_logger() + +#: What the ledger is allowed to take out of a turn. Both are bounded for the same reason the +#: compile plane bounds its own: a store that *fails* is survivable — metering is observability and +#: the turn goes on — but a store that HANGS is worse than a failure, because it ties up the request +#: with no recovery path and nothing above it imposes a timeout. So a slow ledger costs telemetry, +#: never the answer a learner is waiting on. +_DRAIN_TIMEOUT_S = 2.0 +_BUDGET_CHECK_TIMEOUT_S = 1.0 + + +class LiveSessionService: + """Opens and re-reads a learner's sessions. + + Orchestration only, like ``LiveGraphService``: mint the ids, bind correlation, read the map and + what this learner already knows of it, take the turn, persist. What the turn *should* be belongs + to the director and the tutor, which is why both arrive from outside rather than being built + here. + + The graph is read fresh on every session, never cached: C1 grows a map at runtime, so a copy + taken once would be stale the first time a learner asked something off it. + """ + + def __init__( + self, + graphs: IGraphStore, + sessions: ISessionStore, + *, + knowledge: IKnowledgeStore, + tutor: ITutor, + grader: IGrader, + session_budget_s: float, + cost_event_store: ICostEventStore | None = None, + subject_cost_store: ISubjectCostStore | None = None, + credential_resolver: CredentialResolver | None = None, + throttle: LiveSessionThrottle | None = None, + session_budget_usd: float = 0.0, + ) -> None: + self._graphs = graphs + self._sessions = sessions + self._knowledge = knowledge + self._tutor = tutor + self._grader = grader + self._session_budget_s = session_budget_s + # Both optional: metering is observability, and it is simply off when either store is + # unwired (offline dev, the suite). A turn must never fail for want of a ledger. + self._cost_event_store = cost_event_store + self._subject_cost_store = subject_cost_store + # None when BYOK is off — the tutor and grader then read the process environment. + self._credential_resolver = credential_resolver + # None leaves openings unrationed, which is what the suites predating this compose. + self._throttle = throttle + # Ceiling on one session's whole spend, read from the ledger's rollup. 0 is uncapped; it is + # a runaway guard, not a ration — the clock is what bounds an ordinary sitting. + self._session_budget_usd = session_budget_usd + + async def start( + self, graph_id: str, *, session_id: str, owner_id: str | None = None + ) -> Session: + """Open a session on ``graph_id`` and take its first turn. + + The id is minted by the router and passed *in* so it can ride a failure response as well as + a success one — a learner reporting "it went wrong" needs to name the session precisely when + it went wrong, which is the case a header set after the work would never cover. + + Raises ``FileNotFoundError`` when the learner has no such map — including when it is + somebody else's, which is not-found rather than forbidden. Raises + ``TutorUnavailableError`` when the first turn could not be taught; nothing is persisted in + that case, because a session whose first turn never happened is not a session. + """ + # Two ids, because they answer different questions (R6): ``session_id`` is the learner's + # whole session and ``run_id`` is the work of taking THIS turn. Both bound before any I/O — + # the run is the reading and the teaching, not just the model call — so a hung read still + # leaves a trace that the turn was attempted. + run_id = uuid4().hex + bind_run_id(run_id, graph_id=graph_id, session_id=session_id) + logger.info("live.session.starting", graph_id=graph_id, session_id=session_id) + + # Before any work: a refused opening should cost a lookup, not a tutor call. + if self._throttle is not None: + self._throttle.admit_open(owner_id) + + # The stores are synchronous (supabase-py is), so keep the loop free while they work. + graph = await asyncio.to_thread(self._graphs.load, graph_id, owner_id=owner_id) + # What this learner already knows of this map (T2). Without it every session would open on + # the map's first concept and re-teach a returning learner what they came back having + # learned — the director cannot adapt to a model nobody read. + known = await asyncio.to_thread(self._knowledge.load, graph_id, owner_id=owner_id) + + credentials = await self._resolve_credentials(owner_id) + cost = self._cost_scope(run_id=run_id, session_id=session_id, owner_id=owner_id) + try: + # The credential scope has to wrap the turn, not merely be resolved before it: the model + # client is built on first use *inside* the tutor, and it reads the tenant's key off + # this contextvar. Without it a BYOK tenant is taught on the platform's key — money + # spent on their behalf that they never authorized and cannot see. + with self._credential_scope(credentials), enter_cost_scope(cost): + session = await open_session( + graph, + known, + SessionClock(turn=1, elapsed_s=0.0, budget_s=self._session_budget_s), + session_id=session_id, + run_id=run_id, + tutor=self._tutor, + ) + finally: + # Drained even when the turn failed: a tutor call that timed out after the tokens went + # out really spent them, and a ledger recording only successes would under-report + # exactly the runs somebody needs to go and look at. + await self._drain(cost, run_id=run_id, session_id=session_id) + # After the turn, deliberately: a session row written before the tutor spoke would be a + # resumable transcript with nothing in it if the tutor then failed. + await asyncio.to_thread(self._sessions.save, session, owner_id=owner_id) + + # No explicit ids: this line rides the contextvars binding above, so the correlation test + # proves propagation rather than proving they were threaded through by hand. + logger.info("live.session.started", turn_count=len(session.turns)) + return session + + async def answer( + self, session_id: str, answer: str, *, answering_seq: int, owner_id: str | None = None + ) -> Session: + """Score what the learner said, move what the system believes, and take the next turn. + + Two writes, not one transaction, and the order is chosen for how each half fails. The + transcript goes first and the belief second, so a crash between them under-counts evidence + rather than over-counting it: the learner sees a graded turn whose belief did not move, and + the concept simply comes round again. The other order looks safer and is not — the response + is a "try again" (503), a retry re-grades the same answer against a transcript that never + recorded it, and ``apply_evidence`` runs twice on one answer. That is the one thing the + ``_PULL`` / ``_MASTERED`` relationship exists to prevent: two pulls clear the mastery bar, + so a single lucky guess plus a storage blip would unlock a dependent concept. + + The session's age is measured from the row rather than from anything held in this process: + a session outlives the request that opened it and every process that has served it since, + and a clock that reset on a reload would let a learner extend a bounded session forever by + refreshing. + + Raises ``FileNotFoundError`` (no such session for this learner), ``SessionClosedError`` + (the director already ended it), and ``GraderUnavailableError`` / ``TutorUnavailableError`` + when the turn could not be taken at all — nothing has moved in that case, so a retry means + exactly what the learner expects it to. + """ + run_id = uuid4().hex + bind_run_id(run_id, session_id=session_id) + + # Admission first, and in this order: a session already over its ceiling should cost a + # rollup read rather than two store reads and a pair of billed model calls. + await self._refuse_if_budget_spent(session_id, owner_id) + + session = await asyncio.to_thread(self._sessions.load, session_id, owner_id=owner_id) + graph = await asyncio.to_thread(self._graphs.load, session.graph_id, owner_id=owner_id) + known = await asyncio.to_thread(self._knowledge.load, session.graph_id, owner_id=owner_id) + + credentials = await self._resolve_credentials(owner_id) + cost = self._cost_scope(run_id=run_id, session_id=session_id, owner_id=owner_id) + try: + # The slot is what makes the ceiling mean anything. Two answers sent at once both load + # the same session, both pass every check made against that snapshot, and both pay a + # grader and a tutor before either tries to write — and the ceiling cannot see spend + # that has not been drained yet. The compare-and-set on the write settles which answer + # *counts*; only this settles which one is *paid for*. + with ( + self._turn_slot(session_id), + self._credential_scope(credentials), + enter_cost_scope(cost), + ): + outcome = await self._take(session, graph, known, answer, answering_seq, run_id) + finally: + await self._drain(cost, run_id=run_id, session_id=session_id) + # Conditional on the session still being the length this request read. Two answers in + # flight at once both pass ``take_turn``'s check — they loaded the same head — and only the + # store can settle which one lands. The loser is a stale answer, which is what the learner + # is told (409), rather than a graded turn that quietly disappeared. + await asyncio.to_thread( + self._sessions.save, outcome.session, owner_id=owner_id, expect_turns=len(session.turns) + ) + await asyncio.to_thread(self._knowledge.save, outcome.model, owner_id=owner_id) + + logger.info( + "live.session.answered", + turn_count=len(outcome.session.turns), + status=outcome.session.status.value, + ) + return outcome.session + + def _turn_slot(self, session_id: str) -> AbstractContextManager[None]: + """This session's single in-flight turn, or a no-op when nothing is rationing turns.""" + return self._throttle.taking_turn(session_id) if self._throttle else nullcontext() + + async def _take( + self, + session: Session, + graph: ConceptGraph, + known: LearnerModel, + answer: str, + answering_seq: int, + run_id: str, + ) -> TurnOutcome: + """One turn of the loop, with the session's own clock read off its row. + + Clamped: ``SessionClock.elapsed_s`` is ``ge=0``, and a host whose clock steps backwards + between opening a session and answering in it (an NTP correction, a container resync) would + otherwise fail the turn on a validation error the router cannot translate. The same guard + ``recall_of`` applies to its own elapsed count. + """ + return await take_turn( + session, + graph, + known, + answer=answer, + answering_seq=answering_seq, + grader=self._grader, + tutor=self._tutor, + run_id=run_id, + elapsed_s=max(0.0, (datetime.now(UTC) - session.started_at).total_seconds()), + budget_s=self._session_budget_s, + ) + + def _cost_scope( + self, *, run_id: str, session_id: str, owner_id: str | None + ) -> CostScope | None: + """This turn's cost scope, or ``None`` when metering is off. + + Keyed ``LIVE_SESSION`` and stated explicitly (D2): a session's id, a graph's and a course's + are minted from independent sequences, so filing spend under the wrong namespace would + eventually merge two subjects' totals in rows nobody may correct. The subject is the + *session*, not the map it walks — a map outlives every sitting on it, and "what did this + session cost" is a question about one sitting. + """ + return make_cost_scope( + self._cost_event_store, + self._subject_cost_store, + run_id=run_id, + subject_type=CostSubjectType.LIVE_SESSION, + subject_id=session_id, + owner_id=owner_id, + ) + + async def _drain(self, cost: CostScope | None, *, run_id: str, session_id: str) -> None: + """Persist what this turn spent. Never fatal: ``drain_cost_scope`` swallows its own + failures, because a learner losing a turn to a slow ledger would be the telemetry costing + more than it measures.""" + if cost is None: + return + try: + async with asyncio.timeout(_DRAIN_TIMEOUT_S): + await drain_cost_scope(cost, self._cost_event_store, self._subject_cost_store) + except TimeoutError: + # ``drain_cost_scope`` swallows its own failures but not its own duration. Losing the + # row costs a line of telemetry; holding the turn open costs the learner their session. + logger.warning( + "live.session.cost_drain_timed_out", run_id=run_id, session_id=session_id + ) + return + logger.debug("live.session.cost_drained", run_id=run_id, session_id=session_id) + + async def _refuse_if_budget_spent(self, session_id: str, owner_id: str | None) -> None: + """Stop a session that has reached its ceiling, before it spends past it. + + Read from the ledger's rollup rather than counted again in memory: the number already + exists, and a second count would be a second truth. Fails **open** — a rollup that cannot + be read refuses nobody, because a telemetry outage must not end somebody's lesson. + """ + if self._subject_cost_store is None or self._session_budget_usd <= 0: + return + try: + async with asyncio.timeout(_BUDGET_CHECK_TIMEOUT_S): + spent = await self._subject_cost_store.get( + subject_type=CostSubjectType.LIVE_SESSION, + subject_id=session_id, + owner_id=owner_id, + ) + except (PersistenceError, TimeoutError): + logger.warning("live.session.budget_unreadable", session_id=session_id, exc_info=True) + return + if spent is not None and spent.total_amount >= self._session_budget_usd: + logger.info( + "live.session.budget_exhausted", + session_id=session_id, + spent=spent.total_amount, + cap=self._session_budget_usd, + ) + raise LiveSessionBudgetExhaustedError(spent.total_amount, self._session_budget_usd) + + async def _resolve_credentials(self, owner_id: str | None) -> Mapping[str, str] | None: + """The owner's BYOK keys for this turn, or ``None`` to run on the process environment.""" + if owner_id is None or self._credential_resolver is None: + return None + return await self._credential_resolver(owner_id) + + @staticmethod + def _credential_scope( + credentials: Mapping[str, str] | None, + ) -> AbstractContextManager[None]: + """The turn's credential context: the tenant's keys when present, else a no-op (env + fallback). Mirrors the compile plane's, so one tenant's key never outlives its request.""" + return run_credentials(credentials) if credentials else nullcontext() + + async def load(self, session_id: str, *, owner_id: str | None = None) -> Session: + """Re-read a session so a reloaded tab lands back in it (U2). + + Correlated like the open: the resume path is the one U2 exists to make work, so a resume + that fails must be as findable in the logs as an open that fails. As a *request* rather than + a run, though — a resume takes no turn, and a ``run_id`` that was really a session id would + undo the distinction the turns depend on. + """ + bind_request_id(session_id, session_id=session_id) + session = await asyncio.to_thread(self._sessions.load, session_id, owner_id=owner_id) + logger.info("live.session.resumed", turn_count=len(session.turns)) + return session diff --git a/apps/api/src/lunaris_api/live/session/throttle.py b/apps/api/src/lunaris_api/live/session/throttle.py new file mode 100644 index 00000000..1eb689c5 --- /dev/null +++ b/apps/api/src/lunaris_api/live/session/throttle.py @@ -0,0 +1,116 @@ +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from datetime import datetime + +from ...daily_counter import DailyCounter +from ...local_owner_key import LOCAL_OWNER_KEY +from ..work_refused import LiveWorkRefusedError + + +class LiveSessionDailyCapReachedError(LiveWorkRefusedError): + """The owner has opened their allowance of sessions today.""" + + def __init__(self, cap: int) -> None: + self.cap = cap + self.detail = ( + f"You've opened today's {cap} sessions. They reset tomorrow — the maps you already " + "have stay where they are." + ) + super().__init__(self.detail) + + +class LiveSessionBusyError(LiveWorkRefusedError): + """A turn is already being taken in this session, so a second would pay for the same one twice. + + Not a theoretical race. A double-click, a client retry, or a script sends two answers naming the + same turn; both load the same session, both pass every check made against that snapshot, and + both call a grader and a tutor before either tries to write. The write is settled afterwards by + the store's compare-and-set — but by then the money is spent, and the per-session ceiling cannot + see spend that has not been drained yet. The only place to stop it is before the billed calls. + """ + + status_code = 409 + detail = ( + "Your last answer is still being marked. Give it a moment rather than sending it again." + ) + + +class LiveSessionBudgetExhaustedError(LiveWorkRefusedError): + """This session has spent its ceiling, so it takes no further turns. + + A runaway guard rather than a ration: a session is bounded by its clock already, and this only + binds when something is wrong — a loop of retries, or a learner (or a script) answering far past + the point where the sitting was a sitting. + """ + + def __init__(self, spent_usd: float, budget_usd: float) -> None: + self.spent_usd = spent_usd + self.budget_usd = budget_usd + self.detail = ( + "This session has reached its cost ceiling, so it has stopped here. What you " + "demonstrated is saved — starting a fresh session picks up from it." + ) + super().__init__(self.detail) + + +class LiveSessionThrottle: + """Admission control for the session plane. + + Only the *opening* is counted. A turn is not rationed by number, because a session is already + bounded by its clock and by its own cost ceiling — capping turns as well would end sittings that + were going well for a reason nobody could explain to the learner. Opening is where a runaway + starts: a script that opens sessions in a loop pays a tutor call each time. + + In-process, like the compile plane's: the counters live in memory, reset on restart and do not + coordinate across replicas — sufficient for a single-replica deploy, with a DB-backed ledger as + the documented upgrade path. + """ + + def __init__(self, *, open_daily_cap: int, clock: Callable[[], datetime] | None = None) -> None: + self._open_daily_cap = open_daily_cap + self._opens = DailyCounter(clock=clock) if clock else DailyCounter() + # Which sessions have a turn in flight. A set rather than a count: a session takes one turn + # at a time by definition — the learner is answering the question in front of them. + self._in_flight: set[str] = set() + + def admit_open(self, owner_id: str | None) -> None: + """Count one session opening against today's allowance, or refuse it. + + Raises ``LiveSessionDailyCapReachedError`` when the owner has used the allowance. A cap of + 0 or less leaves openings unrationed, which is what the suites that predate this compose. + """ + if self._open_daily_cap <= 0: + return + key = _key(owner_id) + # Checked before counting, so a refused opening does not spend the allowance it was refused + # by — otherwise a learner who hit the cap could never get back under it. + if self._opens.used(key) >= self._open_daily_cap: + raise LiveSessionDailyCapReachedError(self._open_daily_cap) + self._opens.count(key) + + @contextmanager + def taking_turn(self, session_id: str) -> Iterator[None]: + """Hold this session's single turn slot for the length of a turn. + + Claimed *before* the grader and the tutor are called, so a duplicate submission is refused + while it is still free — the compare-and-set on the write settles which answer counts, but + it settles it after both have been paid for. Released in a ``finally`` so a failed turn does + not lock a learner out of their own session. + + Safe without a lock: claiming and releasing are synchronous, and the event loop cannot + interleave two of them. + """ + if session_id in self._in_flight: + raise LiveSessionBusyError + self._in_flight.add(session_id) + try: + yield + finally: + self._in_flight.discard(session_id) + + +def _key(owner_id: str | None) -> str: + """Who the allowance belongs to. An unowned request is the single-user path, which is one + learner however many tabs they have — so they share one bucket rather than getting none. The + same constant every other throttle in this API uses for that caller.""" + return owner_id or LOCAL_OWNER_KEY diff --git a/apps/api/src/lunaris_api/live/work_refused.py b/apps/api/src/lunaris_api/live/work_refused.py new file mode 100644 index 00000000..4c47bb9b --- /dev/null +++ b/apps/api/src/lunaris_api/live/work_refused.py @@ -0,0 +1,22 @@ +"""What Live declines to start, and how every refusal reaches a learner. + +Its own module because both planes raise it: the compile plane rations maps and extensions, the +session plane rations sittings and their spend. Leaving the base in the compile plane's throttle +would have the session plane import a graph module for a class about neither. +""" + + +class LiveWorkRefusedError(Exception): + """Base for work Live declines to start. + + Subclasses set two class attributes the routers read so EVERY refusal maps the same way (one + ``except`` clause, no per-reason branching): ``status_code`` and ``detail``, the learner-facing + sentence. The base seeds the exception message from ``detail`` so logs carry the reason rather + than a blank ``SomeError:``. + """ + + status_code: int = 429 + detail: str = "Live can't take that on right now." + + def __init__(self, detail: str | None = None) -> None: + super().__init__(detail or self.detail) diff --git a/apps/api/tests/conftest.py b/apps/api/tests/conftest.py index 0cd8fb6b..f2fabc06 100644 --- a/apps/api/tests/conftest.py +++ b/apps/api/tests/conftest.py @@ -31,6 +31,18 @@ def _reset_live_throttle() -> Iterator[None]: dependencies._get_live_graph_throttle.cache_clear() +@pytest.fixture(autouse=True) +def _reset_live_session_throttle() -> Iterator[None]: + """And the same again for the session plane's (T8): its per-day opening counts and its in-flight + turn slots are process-wide for a given Settings, so without this one test's sessions could + exhaust another's allowance — or leave a slot held, refusing a turn a later test never sent.""" + from lunaris_api.live.session import dependencies + + dependencies._get_live_session_throttle.cache_clear() + yield + dependencies._get_live_session_throttle.cache_clear() + + class ReleasablePipeline: """A pipeline that emits one progress beat, parks on a release ``Event``, then builds a real course via the stub orchestrator. Lets a test drop the SSE consumer *before* the build finishes diff --git a/apps/api/tests/live/test_live_session_cost.py b/apps/api/tests/live/test_live_session_cost.py new file mode 100644 index 00000000..aa1877b6 --- /dev/null +++ b/apps/api/tests/live/test_live_session_cost.py @@ -0,0 +1,328 @@ +"""What a session spends, and what stops it spending without end (Phase 2a, T8). + +A session is the third thing in this system that can spend money, after a course and a Live map. It +is not either of them: a map outlives every sitting on it and is purged separately, and "what did +this session cost" is a question about one sitting. So it is its own subject in the ledger — which +D2 generalized the rollup key for, precisely so a new spender would be an enum value rather than a +migration on immutable financial rows. + +The admission rules follow the same shape as the compile plane's, with one deliberate difference: +only the *opening* is rationed by count. A session is already bounded by its clock and by its own +cost ceiling, and capping turns as well would end sittings that were going well for a reason nobody +could explain to the learner. +""" + +import asyncio +from datetime import UTC, datetime +from pathlib import Path + +import httpx +from lunaris_api.app import create_app +from lunaris_api.config import Settings, get_settings +from lunaris_api.dependencies import get_cost_event_store, get_subject_cost_store +from lunaris_api.live.dependencies import resolve_graph_store +from lunaris_api.live.session.dependencies import _resolve_session_store +from lunaris_live.session import MemoryKnowledgeStore, StubGrader +from lunaris_runtime.metering import record_cost +from lunaris_runtime.persistence import InMemoryCostEventStore, InMemorySubjectCostStore +from lunaris_runtime.schema import CostSubjectType, SubjectCost + + +def _settings(tmp_path: Path, **overrides: object) -> Settings: + return Settings( + pipeline="stub", + course_dir=tmp_path, + cors_origins=(), + env_file=tmp_path / ".env", + **overrides, # type: ignore[arg-type] + ) + + +async def _graph(client: httpx.AsyncClient) -> dict: + return (await client.post("/api/live/graphs", json={"topic": "How tides work"})).json() + + +async def test_a_sessions_spend_is_filed_under_the_session(tmp_path: Path) -> None: + """A cost recorded inside a turn lands on the session, keyed by its own id. + + Filing it under the map would merge every sitting anyone ever has on that map into one total — + and under the *course* namespace it could collide with a real course id, in append-only rows + nobody may correct. + """ + # Arrange — a tutor that costs something, standing in for a real model call. + from lunaris_api.live.session.dependencies import get_live_tutor + from lunaris_runtime.schema import CostProvider, CostUnit + + class CostlyTutor: + async def teach(self, move, node, *, topic, criterion=None, already_said=(), run_id): + record_cost( + component="live_tutor", + provider=CostProvider.ANTHROPIC, + model="claude-opus-4-8", + usage={CostUnit.INPUT_TOKENS: 1000.0, CostUnit.OUTPUT_TOKENS: 500.0}, + ) + return f"Teaching {node.name}." + + events, rollup = InMemoryCostEventStore(), InMemorySubjectCostStore() + app = create_app() + app.dependency_overrides[get_settings] = lambda: _settings(tmp_path) + app.dependency_overrides[get_cost_event_store] = lambda: events + app.dependency_overrides[get_subject_cost_store] = lambda: rollup + app.dependency_overrides[get_live_tutor] = lambda: CostlyTutor() + transport = httpx.ASGITransport(app=app) + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + graph = await _graph(client) + + # Act + session = ( + await client.post("/api/live/sessions", json={"graphId": graph["graphId"]}) + ).json() + + # Assert + spend = await rollup.get( + subject_type=CostSubjectType.LIVE_SESSION, subject_id=session["sessionId"], owner_id=None + ) + assert spend is not None, "the turn's spend was never filed" + assert spend.total_amount > 0 + assert spend.subject_type is CostSubjectType.LIVE_SESSION + assert spend.subject_id == session["sessionId"] + + +async def test_a_session_that_has_spent_its_ceiling_takes_no_more_turns(tmp_path: Path) -> None: + """The runaway guard. A session is bounded by its clock, so this only binds when something is + wrong — and it is read from the ledger's rollup rather than counted a second time in memory, + because the number already exists and a second count would be a second truth.""" + # Arrange — a ledger already showing this session well past its ceiling. + events, rollup = InMemoryCostEventStore(), InMemorySubjectCostStore() + app = create_app() + app.dependency_overrides[get_settings] = lambda: _settings( + tmp_path, live_session_budget_usd=0.01 + ) + app.dependency_overrides[get_cost_event_store] = lambda: events + app.dependency_overrides[get_subject_cost_store] = lambda: rollup + transport = httpx.ASGITransport(app=app) + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + graph = await _graph(client) + session = ( + await client.post("/api/live/sessions", json={"graphId": graph["graphId"]}) + ).json() + await rollup.upsert( + cost=SubjectCost( + subject_type=CostSubjectType.LIVE_SESSION, + subject_id=session["sessionId"], + total_amount=5.0, + currency="USD", + breakdown={}, + price_book_version="test", + updated_at=datetime.now(UTC), + ), + owner_id=None, + ) + + # Act + response = await client.post( + f"/api/live/sessions/{session['sessionId']}/turns", + json={"answer": "Another one.", "answeringSeq": 1}, + ) + + # Assert — refused with words, not a bare 500, and the learner is told their work is kept. + assert response.status_code == 429, response.text + assert "cost ceiling" in response.json()["detail"] + + +async def test_a_learner_cannot_open_sessions_without_end(tmp_path: Path) -> None: + """Opening is where a runaway starts: a script that opens sessions in a loop pays a tutor call + every time. The cap is per owner per day, and a refused opening does not spend the allowance it + was refused by.""" + # Arrange + app = create_app() + app.dependency_overrides[get_settings] = lambda: _settings(tmp_path, live_session_daily_cap=2) + transport = httpx.ASGITransport(app=app) + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + graph = await _graph(client) + opened = [ + ( + await client.post("/api/live/sessions", json={"graphId": graph["graphId"]}) + ).status_code + for _ in range(3) + ] + + # Assert + assert opened == [201, 201, 429] + + +async def test_an_unreadable_ledger_does_not_end_a_lesson(tmp_path: Path) -> None: + """Fails open, deliberately. This is a cap on money and metering is observability — if a + degraded ledger could refuse work, a telemetry outage would become a product outage, and the + learner would be told they are out of budget when in truth nobody knows what they spent.""" + # Arrange + from lunaris_runtime.persistence import PersistenceError + + class BrokenRollup(InMemorySubjectCostStore): + async def get(self, **kwargs: object): # type: ignore[override] + raise PersistenceError("the ledger is down") + + app = create_app() + app.dependency_overrides[get_settings] = lambda: _settings( + tmp_path, live_session_budget_usd=0.01 + ) + app.dependency_overrides[get_cost_event_store] = lambda: InMemoryCostEventStore() + app.dependency_overrides[get_subject_cost_store] = lambda: BrokenRollup() + transport = httpx.ASGITransport(app=app) + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + graph = await _graph(client) + session = ( + await client.post("/api/live/sessions", json={"graphId": graph["graphId"]}) + ).json() + + # Act + response = await client.post( + f"/api/live/sessions/{session['sessionId']}/turns", + json={"answer": "Something real.", "answeringSeq": 1}, + ) + + # Assert + assert response.status_code == 200, response.text + + +async def test_a_tenants_session_is_taught_on_their_own_key(tmp_path: Path) -> None: + """BYOK reaches the *turn*, not just the compile. + + The model client is built on first use inside the tutor, and it reads the key off a contextvar. + Resolving the tenant's credentials without scoping the call would teach their session on the + platform's key — money spent on their behalf that they never authorized and cannot see, on a + surface that spends on every single turn. + """ + # Arrange — a tutor that reports which key it would actually have used. + from lunaris_api.dependencies import optional_user_id + from lunaris_api.live.session.dependencies import get_live_session_service, get_live_tutor + from lunaris_api.live.session.service import LiveSessionService + from lunaris_runtime.credentials import resolve_secret + + seen: list[str | None] = [] + + class ReportingTutor: + async def teach(self, move, node, *, topic, criterion=None, already_said=(), run_id): + seen.append(resolve_secret("ANTHROPIC_API_KEY")) + return f"Teaching {node.name}." + + async def tenant_keys(owner_id: str) -> dict[str, str]: + return {"ANTHROPIC_API_KEY": f"sk-{owner_id}"} + + settings = _settings(tmp_path) + app = create_app() + app.dependency_overrides[get_settings] = lambda: settings + app.dependency_overrides[optional_user_id] = lambda: "learner-1" + app.dependency_overrides[get_live_tutor] = lambda: ReportingTutor() + app.dependency_overrides[get_live_session_service] = lambda: LiveSessionService( + resolve_graph_store(settings), + _resolve_session_store(settings), + knowledge=MemoryKnowledgeStore(), + tutor=ReportingTutor(), + grader=StubGrader(), + session_budget_s=settings.live_session_budget_s, + credential_resolver=tenant_keys, + ) + transport = httpx.ASGITransport(app=app) + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + graph = await _graph(client) + + # Act — an owned request, which is what BYOK is for. + opened = await client.post("/api/live/sessions", json={"graphId": graph["graphId"]}) + + # Assert + assert opened.status_code == 201, opened.text + assert seen == ["sk-learner-1"] + + +async def test_two_answers_at_once_pay_for_one_turn(tmp_path: Path) -> None: + """The ceiling only means something if the spending is gated before it happens. + + Two answers sent at the same moment both load the same session, both name the turn in front of + the learner, and both pass every check made against that snapshot — the store's compare-and-set + settles which one *counts*, but only after both have paid a grader and a tutor. A ceiling read + from the ledger cannot see spend that has not been drained yet, so the refusal has to happen + before the billed calls. + """ + # Arrange — a tutor slow enough that the second request arrives while the first is inside it. + from lunaris_api.live.session.dependencies import get_live_tutor + + calls = 0 + + class SlowTutor: + async def teach(self, move, node, *, topic, criterion=None, already_said=(), run_id): + nonlocal calls + calls += 1 + await asyncio.sleep(0.05) + return f"Teaching {node.name}." + + app = create_app() + app.dependency_overrides[get_settings] = lambda: _settings(tmp_path) + app.dependency_overrides[get_live_tutor] = lambda: SlowTutor() + transport = httpx.ASGITransport(app=app) + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + graph = await _graph(client) + session = ( + await client.post("/api/live/sessions", json={"graphId": graph["graphId"]}) + ).json() + opening_calls = calls + + # Act — the same answer, twice, at once. + answer = {"answer": "It points downhill.", "answeringSeq": 1} + first, second = await asyncio.gather( + client.post(f"/api/live/sessions/{session['sessionId']}/turns", json=answer), + client.post(f"/api/live/sessions/{session['sessionId']}/turns", json=answer), + ) + + # Assert — one turn taken, one refused, and exactly one tutor call paid for. + assert sorted([first.status_code, second.status_code]) == [200, 409] + assert calls - opening_calls == 1 + + +async def test_a_ledger_that_hangs_does_not_hang_the_turn(tmp_path: Path) -> None: + """Worse than a ledger that fails is one that never answers. + + ``drain_cost_scope`` swallows its own failures but not its own duration, and nothing above this + imposes a request timeout — so an unbounded read or write would tie the turn up with no + recovery path at all. A slow ledger must cost telemetry, never the answer somebody is waiting + on. + """ + + # Arrange — a rollup that never returns, on both the paths a turn touches. + class HangingRollup(InMemorySubjectCostStore): + async def get(self, **kwargs: object): # type: ignore[override] + await asyncio.sleep(30) + raise AssertionError("should have been given up on") + + async def upsert(self, **kwargs: object) -> None: # type: ignore[override] + await asyncio.sleep(30) + raise AssertionError("should have been given up on") + + app = create_app() + app.dependency_overrides[get_settings] = lambda: _settings( + tmp_path, live_session_budget_usd=0.01 + ) + app.dependency_overrides[get_cost_event_store] = lambda: InMemoryCostEventStore() + app.dependency_overrides[get_subject_cost_store] = lambda: HangingRollup() + transport = httpx.ASGITransport(app=app) + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + graph = await _graph(client) + + # Act / Assert — bounded well inside the 30s the store would take, and the turn lands. + async with asyncio.timeout(10): + session = ( + await client.post("/api/live/sessions", json={"graphId": graph["graphId"]}) + ).json() + answered = await client.post( + f"/api/live/sessions/{session['sessionId']}/turns", + json={"answer": "Something real.", "answeringSeq": 1}, + ) + + assert answered.status_code == 200, answered.text diff --git a/apps/api/tests/live/test_live_session_service.py b/apps/api/tests/live/test_live_session_service.py new file mode 100644 index 00000000..726b275f --- /dev/null +++ b/apps/api/tests/live/test_live_session_service.py @@ -0,0 +1,187 @@ +"""The session plane below HTTP: what the service does with a learner's beliefs (Phase 2a, T4). + +Nothing writes a belief through the API yet — the grader that does is T5 — so the claim that a +returning learner is met where they left off cannot be made through the endpoint. It is made here +instead, against the real stores and the real package, because the alternative is shipping the +knowledge read untested and discovering at T5 that the service was passing an empty model all along. + +Owner scoping rides along rather than getting its own test here: the beliefs are stored under the +learner, so a service that read them unscoped would find an empty model and open at the root, which +is exactly what the first test would catch. The leak in the other direction — one learner reaching +another's map — is closed a layer earlier by the graph store, which refuses the read outright. +""" + +from datetime import UTC, datetime, timedelta + +import pytest +from lunaris_api.live.session.service import LiveSessionService +from lunaris_live.graph import ConceptGraph, MemoryGraphStore, StubGraphCompiler +from lunaris_live.session import ( + EvidenceKind, + LearnerModel, + MemoryKnowledgeStore, + MemorySessionStore, + Session, + StubGrader, + StubTutor, + apply_evidence, +) +from lunaris_runtime.persistence import PersistenceError + +_TOPIC = "How neural networks learn" + + +async def _map() -> ConceptGraph: + return await StubGraphCompiler().compile(_TOPIC, graph_id="g1", run_id="r0") + + +def _mastering(graph: ConceptGraph, node_id: str) -> LearnerModel: + """Beliefs as the grader will write them (T5): repeated met evidence on one concept.""" + model = LearnerModel(graph_id=graph.graph_id) + for turn in range(1, 4): + model = apply_evidence(model, node_id, EvidenceKind.MET, at_turn=turn) + return model + + +@pytest.fixture +async def wired() -> tuple[LiveSessionService, ConceptGraph, MemoryKnowledgeStore]: + graph = await _map() + graphs, knowledge = MemoryGraphStore(), MemoryKnowledgeStore() + graphs.save(graph, owner_id="learner-1") + service = LiveSessionService( + graphs, + MemorySessionStore(), + knowledge=knowledge, + tutor=StubTutor(), + grader=StubGrader(), + session_budget_s=1800.0, + ) + return service, graph, knowledge + + +async def test_a_returning_learner_is_met_where_their_beliefs_left_them( + wired: tuple[LiveSessionService, ConceptGraph, MemoryKnowledgeStore], +) -> None: + # Arrange — they demonstrated the opening concept in an earlier session. + service, graph, knowledge = wired + knowledge.save(_mastering(graph, graph.topo_order[0]), owner_id="learner-1") + + # Act + session = await service.start("g1", session_id="s1", owner_id="learner-1") + + # Assert — the next concept, not the one they already have. + assert session.turns[0].move.node_id == graph.topo_order[1] + + +async def test_a_first_session_opens_at_the_start_of_the_map( + wired: tuple[LiveSessionService, ConceptGraph, MemoryKnowledgeStore], +) -> None: + """The other half of the same claim: with no beliefs stored, the director gets an empty model + rather than a missing one, and the session opens where the map does.""" + # Arrange + service, graph, _ = wired + + # Act + session = await service.start("g1", session_id="s1", owner_id="learner-1") + + # Assert + assert session.turns[0].move.node_id == graph.topo_order[0] + + +async def test_beliefs_stored_for_one_learner_are_not_read_for_another( + wired: tuple[LiveSessionService, ConceptGraph, MemoryKnowledgeStore], +) -> None: + """The knowledge read has to carry the owner through. Unscoped it would return an empty model + here — harmless — but on a store where a stranger's row *could* match it would walk somebody + past concepts they have never met, and it would look like the product working.""" + # Arrange — the beliefs belong to another learner entirely. + service, graph, knowledge = wired + knowledge.save(_mastering(graph, graph.topo_order[0]), owner_id="someone-else") + + # Act + session = await service.start("g1", session_id="s2", owner_id="learner-1") + + # Assert + assert session.turns[0].move.node_id == graph.topo_order[0] + + +async def test_the_transcript_is_written_before_the_belief() -> None: + """Two writes, not one transaction, so the order is chosen for how each half fails. + + Transcript first means a crash between them under-counts evidence: the learner sees a graded + turn whose belief did not move, and the concept comes round again. The other order looks safer + and is not — the response is a retryable 503, a retry re-grades the same answer against a + transcript that never recorded it, and one lucky guess plus a storage blip clears the mastery + bar that ``_PULL`` was sized to keep two answers away. + """ + # Arrange — a session store that reads back fine and refuses every write. + graph = await _map() + graphs, knowledge = MemoryGraphStore(), MemoryKnowledgeStore() + graphs.save(graph, owner_id="learner-1") + opened = await LiveSessionService( + graphs, + MemorySessionStore(), + knowledge=knowledge, + tutor=StubTutor(), + grader=StubGrader(), + session_budget_s=1800.0, + ).start("g1", session_id="s1", owner_id="learner-1") + + class RefusesToWrite: + def save( + self, + session: Session, + *, + owner_id: str | None = None, + expect_turns: int | None = None, + ) -> None: + raise PersistenceError("storage is having trouble") + + def load(self, session_id: str, *, owner_id: str | None = None) -> Session: + return opened + + service = LiveSessionService( + graphs, + RefusesToWrite(), + knowledge=knowledge, + tutor=StubTutor(), + grader=StubGrader(), + session_budget_s=1800.0, + ) + + # Act + with pytest.raises(PersistenceError): + await service.answer("s1", "I have no idea.", answering_seq=1, owner_id="learner-1") + + # Assert — the belief never landed, because the transcript never did. + assert knowledge.load("g1", owner_id="learner-1").nodes == {} + + +async def test_a_clock_that_stepped_backwards_does_not_break_a_turn() -> None: + """Hosts correct their clocks. If the machine answering a turn has stepped behind the one that + opened the session, the elapsed time is negative — and ``SessionClock.elapsed_s`` is ``ge=0``, + so the turn would fail on a validation error no handler translates and the learner would get a + bare 500 for something entirely on our side.""" + # Arrange — a session stamped in the future, which is what a backward step looks like. + graph = await _map() + graphs, sessions, knowledge = MemoryGraphStore(), MemorySessionStore(), MemoryKnowledgeStore() + graphs.save(graph, owner_id="learner-1") + service = LiveSessionService( + graphs, + sessions, + knowledge=knowledge, + tutor=StubTutor(), + grader=StubGrader(), + session_budget_s=1800.0, + ) + opened = await service.start("g1", session_id="s1", owner_id="learner-1") + sessions.save( + opened.model_copy(update={"started_at": datetime.now(UTC) + timedelta(minutes=5)}), + owner_id="learner-1", + ) + + # Act + answered = await service.answer("s1", "An answer.", answering_seq=1, owner_id="learner-1") + + # Assert — the turn happened, treated as no time having passed. + assert answered.turns[0].answer == "An answer." diff --git a/apps/api/tests/live/test_live_sessions_api.py b/apps/api/tests/live/test_live_sessions_api.py new file mode 100644 index 00000000..a0d52adc --- /dev/null +++ b/apps/api/tests/live/test_live_sessions_api.py @@ -0,0 +1,542 @@ +"""Lunaris Live, Phase 2a — a session, end to end through the API. + +A learner opens a session on a compiled graph, the session is persisted, and it comes back with its +first turn. T1 pinned the wiring — web-facing contract → service → ``lunaris_live.session`` → store, +with one ``session_id`` correlating the lot, and a session as a row rather than connection state +(U2) so a reload returns the learner to where they were. + +T4 makes the turn real: the move is the director's (T3) over what this learner already knows (T2), +and the words are the tutor's, written over the concept's authored teaching notes. What the API +level proves is that those notes *travel* — compiled into the map, through the store, into the +session, out to the learner — which no test of any one layer can see. + +Exercised through the real ASGI app over httpx; the stubs are the compiler behind the graph and the +tutor behind the turn, both real implementations of their protocols. +""" + +import json +from collections.abc import AsyncIterator +from pathlib import Path + +import httpx +import pytest +from lunaris_api.app import create_app +from lunaris_api.config import Settings, get_settings +from lunaris_api.live.dependencies import resolve_graph_store +from lunaris_api.live.session.dependencies import ( + get_live_grader, + get_live_session_service, + get_live_tutor, +) +from lunaris_api.live.session.service import LiveSessionService +from lunaris_live.graph import ConceptNode, MasteryCriterion +from lunaris_live.session import ( + DirectorMove, + GraderUnavailableError, + MemoryKnowledgeStore, + Session, + SessionFormatError, + StubGrader, + StubTutor, + TurnGrade, + TutorUnavailableError, +) + + +@pytest.fixture +async def client(tmp_path: Path) -> AsyncIterator[httpx.AsyncClient]: + app = create_app() + app.dependency_overrides[get_settings] = lambda: Settings( + pipeline="stub", course_dir=tmp_path, cors_origins=(), env_file=tmp_path / ".env" + ) + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as http_client: + yield http_client + + +@pytest.fixture +async def client_with_a_silent_tutor(tmp_path: Path) -> AsyncIterator[httpx.AsyncClient]: + """The same app with a tutor that cannot speak — the provider being down, mid-open.""" + + class SilentTutor: + async def teach( + self, + move: DirectorMove, + node: ConceptNode, + *, + topic: str, + criterion: MasteryCriterion | None = None, + run_id: str, + ) -> str: + raise TutorUnavailableError("provider is down") + + app = create_app() + app.dependency_overrides[get_settings] = lambda: Settings( + pipeline="stub", course_dir=tmp_path, cors_origins=(), env_file=tmp_path / ".env" + ) + app.dependency_overrides[get_live_tutor] = lambda: SilentTutor() + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as http_client: + yield http_client + + +@pytest.fixture +async def client_with_an_unreadable_session(tmp_path: Path) -> AsyncIterator[httpx.AsyncClient]: + """The same app over a store holding a session this build can no longer parse.""" + + class UnreadableStore: + def save(self, session: Session, *, owner_id: str | None = None) -> None: ... + + def load(self, session_id: str, *, owner_id: str | None = None) -> Session: + raise SessionFormatError(f"session {session_id} is not in a readable format") + + settings = Settings( + pipeline="stub", course_dir=tmp_path, cors_origins=(), env_file=tmp_path / ".env" + ) + app = create_app() + app.dependency_overrides[get_settings] = lambda: settings + app.dependency_overrides[get_live_session_service] = lambda: LiveSessionService( + resolve_graph_store(settings), + UnreadableStore(), + knowledge=MemoryKnowledgeStore(), + tutor=StubTutor(), + grader=StubGrader(), + session_budget_s=settings.live_session_budget_s, + ) + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as http_client: + yield http_client + + +@pytest.fixture +async def client_with_no_time_left(tmp_path: Path) -> AsyncIterator[httpx.AsyncClient]: + """The same app with a session budget already spent by the time anyone answers.""" + app = create_app() + app.dependency_overrides[get_settings] = lambda: Settings( + pipeline="stub", + course_dir=tmp_path, + cors_origins=(), + env_file=tmp_path / ".env", + live_session_budget_s=0.001, + ) + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as http_client: + yield http_client + + +@pytest.fixture +async def client_with_a_broken_grader(tmp_path: Path) -> AsyncIterator[httpx.AsyncClient]: + """The same app with a grader that cannot score — the provider being down, mid-session.""" + + class BrokenGrader: + async def grade( + self, + answer: str, + *, + criterion: MasteryCriterion, + node: ConceptNode, + run_id: str, + ) -> TurnGrade: + raise GraderUnavailableError("provider is down") + + app = create_app() + app.dependency_overrides[get_settings] = lambda: Settings( + pipeline="stub", course_dir=tmp_path, cors_origins=(), env_file=tmp_path / ".env" + ) + app.dependency_overrides[get_live_grader] = lambda: BrokenGrader() + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as http_client: + yield http_client + + +async def _graph(client: httpx.AsyncClient) -> dict: + """A compiled map for the session to run on — Phase 1's surface, used as given.""" + return ( + await client.post("/api/live/graphs", json={"topic": "How neural networks learn"}) + ).json() + + +async def test_opening_a_session_returns_its_first_turn(client: httpx.AsyncClient) -> None: + # Arrange + graph = await _graph(client) + + # Act + response = await client.post("/api/live/sessions", json={"graphId": graph["graphId"]}) + + # Assert — the contract the surface renders: a live session, on a known map, already teaching. + assert response.status_code == 201, response.text + session = response.json() + assert session["graphId"] == graph["graphId"] + assert session["status"] == "active" + assert session["turns"], "a session that opens with nothing to show is a loading spinner" + + first = session["turns"][0] + assert first["seq"] == 1 + # Every turn is a director move plus what the tutor said about it (A2). The move is what makes + # the transcript auditable — without it a turn is prose nobody can explain the choice of. + assert first["move"]["kind"] == "introduce" + assert first["move"]["nodeId"] in {node["id"] for node in graph["nodes"]} + assert first["move"]["reason"], "a move with no reason cannot be audited (plan §7)" + assert first["tutor"].strip() + + # The run id has to ride the response: a session is many minutes of work across many turns, and + # a learner reporting "it went wrong" needs to name it. + assert response.headers["X-Session-Id"] == session["sessionId"] + + +async def test_the_opening_turn_teaches_what_the_map_authored(client: httpx.AsyncClient) -> None: + """The claim no single layer can make: a concept's teaching notes reach the learner. + + Phase 1 pays a model call per concept to author misconceptions — "the wrong models learners + commonly hold, stated as the learner would believe them" — precisely so the tutor can go after + the specific wrong idea in front of it. If they stop travelling anywhere between the compiler + and the transcript, every session silently degrades to a definition read aloud, and the map + still looks perfect. + """ + # Arrange + graph = await _graph(client) + + # Act + session = (await client.post("/api/live/sessions", json={"graphId": graph["graphId"]})).json() + + # Assert — the authored text itself, taken from the compiled map rather than restated here, so + # this cannot pass by agreeing with a copy of the stub. + taught_on = session["turns"][0]["move"]["nodeId"] + node = next(n for n in graph["nodes"] if n["id"] == taught_on) + misconception = node["teachingSpec"]["misconceptions"][0] + assert misconception in session["turns"][0]["tutor"] + + +async def test_a_turn_is_correlated_by_the_run_that_produced_it(client: httpx.AsyncClient) -> None: + """R6: the session carries a ``session_id``, each turn carries its own ``run_id``. A turn is + now one or more model calls, and the two ids answer different questions — "show me this + learner's session" and "show me what the tutor was asked on this line". + + Deliberately the *shape* half of the claim: on its own this would pass on a constant. The test + below it is the binding half, and neither is the whole claim without the other. + """ + # Arrange + graph = await _graph(client) + + # Act + session = (await client.post("/api/live/sessions", json={"graphId": graph["graphId"]})).json() + + # Assert + run_id = session["turns"][0]["runId"] + assert run_id + assert run_id != session["sessionId"], "a turn's run is not the session it belongs to" + + +async def test_the_turns_run_id_is_the_one_its_work_logged_under( + client: httpx.AsyncClient, capsys: pytest.CaptureFixture[str] +) -> None: + """Otherwise the id on the turn is decoration. The point of storing it is that somebody holding + a line of transcript can find what actually happened underneath it — so the id the learner gets + back has to be the id the work was logged with, at every layer it crossed.""" + # Arrange + graph = await _graph(client) + + # Act + session = (await client.post("/api/live/sessions", json={"graphId": graph["graphId"]})).json() + + # Assert — read off the stdout JSON rather than structlog's capture_logs, which cannot + # intercept a module logger another test has already cached (house pattern, see + # test_live_graphs_api.py). + run_id = session["turns"][0]["runId"] + correlated = [line for line in _json_log_lines(capsys) if line.get("run_id") == run_id] + events = {line.get("event") for line in correlated} + # ``live.session.started`` passes no ids of its own — it can only appear here if the contextvars + # binding propagated, which is what stops this passing with the binding deleted and the id + # merely threaded through arguments by hand. + assert "live.session.started" in events, f"turn {run_id} left no trace; saw {events}" + assert all(line.get("session_id") == session["sessionId"] for line in correlated), ( + "a turn's run must stay attached to the session it belongs to" + ) + + +def _json_log_lines(capsys: pytest.CaptureFixture[str]) -> list[dict[str, object]]: + """The structured stdout log lines emitted so far (the project logs JSON to stdout).""" + return [ + json.loads(line) + for line in capsys.readouterr().out.splitlines() + if line.startswith("{") and line.endswith("}") + ] + + +async def test_a_tutor_that_cannot_speak_leaves_no_session_behind( + client_with_a_silent_tutor: httpx.AsyncClient, +) -> None: + """A session whose first turn never happened is not a session. Persisting the shell would give + a learner a resumable transcript with nothing in it — and ``turns`` is what the surface renders, + so it would read as a session that opened and then went quiet.""" + # Arrange + graph = await _graph(client_with_a_silent_tutor) + + # Act + response = await client_with_a_silent_tutor.post( + "/api/live/sessions", json={"graphId": graph["graphId"]} + ) + + # Assert — retryable (the provider may come back), correlated, and nothing persisted. + assert response.status_code == 503, response.text + session_id = response.headers["X-Session-Id"] + assert session_id + resumed = await client_with_a_silent_tutor.get(f"/api/live/sessions/{session_id}") + assert resumed.status_code == 404 + + +async def _answer( + client: httpx.AsyncClient, session_id: str, answer: str, *, seq: int = 1 +) -> httpx.Response: + """Answer the turn the learner is looking at. The seq is part of the contract (T6): a duplicate + submit must not be graded against the question that replaced the one it was written for.""" + return await client.post( + f"/api/live/sessions/{session_id}/turns", json={"answer": answer, "answeringSeq": seq} + ) + + +async def test_answering_takes_the_session_to_its_next_turn(client: httpx.AsyncClient) -> None: + """The loop, through the endpoint the surface will drive: the learner answers what the last + turn asked, and what comes back is a session with one more beat in it.""" + # Arrange + graph = await _graph(client) + opened = (await client.post("/api/live/sessions", json={"graphId": graph["graphId"]})).json() + asked = opened["turns"][0] + assert asked["criterion"], "a turn with nothing staged can never be answered" + + # Act + response = await _answer(client, opened["sessionId"], "I have no idea about any of this.") + + # Assert — the answered turn keeps its question, gains the answer and the verdict on it. + assert response.status_code == 200, response.text + session = response.json() + answered = session["turns"][0] + assert answered["answer"] == "I have no idea about any of this." + assert answered["grade"]["kind"] == "not_met" + assert answered["grade"]["reason"], "a verdict with no reason is not feedback" + assert len(session["turns"]) == 2 + + +async def test_what_a_learner_demonstrated_outlives_the_session(client: httpx.AsyncClient) -> None: + """The claim T2 was built for and nothing could prove until now: evidence written by one + session is read by the next. Without it a learner would be met at the root of the map every + time, however much they had already shown.""" + # Arrange — answer the opening concept well enough to master it. + graph = await _graph(client) + opened = (await client.post("/api/live/sessions", json={"graphId": graph["graphId"]})).json() + session_id, first_concept = opened["sessionId"], opened["turns"][0]["move"]["nodeId"] + statement = opened["turns"][0]["criterion"]["statement"] + answered = opened + for _ in range(3): + answered = ( + await _answer(client, session_id, statement, seq=answered["turns"][-1]["seq"]) + ).json() + if answered["status"] != "active": + break + + # Act — a brand new session on the same map. + reopened = (await client.post("/api/live/sessions", json={"graphId": graph["graphId"]})).json() + + # Assert — it does not start over on the concept they just demonstrated. + assert reopened["sessionId"] != session_id + assert reopened["turns"][0]["move"]["nodeId"] != first_concept + + +async def test_a_session_out_of_time_says_goodbye_rather_than_going_quiet( + client_with_no_time_left: httpx.AsyncClient, +) -> None: + """The clock is wall time measured from the row (plan §6, AD9), so it is real across requests + and survives a reload — before T6 it was hardcoded to zero and the budget was a setting nothing + could reach. And the ending is a turn: ``status`` is a field, but the transcript is what the + learner reads, and a session that stopped talking is indistinguishable from one that crashed.""" + # Arrange + graph = await _graph(client_with_no_time_left) + opened = ( + await client_with_no_time_left.post( + "/api/live/sessions", json={"graphId": graph["graphId"]} + ) + ).json() + + # Act + session = (await _answer(client_with_no_time_left, opened["sessionId"], "Anything.")).json() + + # Assert — closed, with a goodbye that says why, and the answered turn still behind it. + assert session["status"] == "closed" + closing = session["turns"][-1] + assert closing["move"]["kind"] == "close" + assert closing["tutor"].strip() + assert "minutes are up" in closing["tutor"] + assert session["turns"][-2]["answer"] == "Anything." + + +async def test_a_session_that_has_closed_does_not_take_another_answer( + client: httpx.AsyncClient, +) -> None: + """A stale tab answering into a session the director ended would run it past the bound the + close exists to enforce. 409: the request is fine, the session's state is not.""" + # Arrange — a map the stub compiles to three concepts, answered until the director closes. + graph = await _graph(client) + opened = (await client.post("/api/live/sessions", json={"graphId": graph["graphId"]})).json() + session_id = opened["sessionId"] + session = opened + for _ in range(12): + head = session["turns"][-1] + session = (await _answer(client, session_id, head["tutor"], seq=head["seq"])).json() + if session["status"] == "closed": + break + assert session["status"] == "closed", "the director never ran out of material" + + # Act + response = await _answer(client, session_id, "One more thing?") + + # Assert + assert response.status_code == 409, response.text + + +async def test_an_answer_to_a_question_that_has_moved_on_is_refused( + client: httpx.AsyncClient, +) -> None: + """A learner pressing send twice, or a second tab. Without the seq the duplicate is graded + against the question that replaced the one they were answering — the words land in the record + under a criterion they were never written for, and the belief that moves is the wrong one.""" + # Arrange — one answer given, so the session is on turn 2. + graph = await _graph(client) + opened = (await client.post("/api/live/sessions", json={"graphId": graph["graphId"]})).json() + await _answer(client, opened["sessionId"], "First attempt.", seq=1) + + # Act — the same submit arriving again. + response = await _answer(client, opened["sessionId"], "First attempt.", seq=1) + + # Assert — a conflict with the session's state, not with the learner: reloading is the recovery. + assert response.status_code == 409, response.text + resumed = (await client.get(f"/api/live/sessions/{opened['sessionId']}")).json() + assert len(resumed["turns"]) == 2, "the refused answer must not have taken a turn" + + +async def test_an_answer_that_names_no_turn_is_refused(client: httpx.AsyncClient) -> None: + """The seq is part of the contract, not an optimisation a client may skip: omitted, the server + would have to guess which question was being answered, which is the guess this prevents.""" + # Arrange + graph = await _graph(client) + opened = (await client.post("/api/live/sessions", json={"graphId": graph["graphId"]})).json() + + # Act / Assert + response = await client.post( + f"/api/live/sessions/{opened['sessionId']}/turns", json={"answer": "Something."} + ) + assert response.status_code == 422 + + +async def test_an_answer_with_nothing_in_it_is_refused_at_the_door( + client: httpx.AsyncClient, +) -> None: + """Not graded as a miss. An empty POST is a client bug, and recording it as evidence would + lower a belief on the strength of somebody's stray keystroke.""" + # Arrange + graph = await _graph(client) + opened = (await client.post("/api/live/sessions", json={"graphId": graph["graphId"]})).json() + + # Act / Assert — including whitespace, which is the version a real client actually sends: a + # space passes a raw length check and reaches the grader as an empty answer, which scores as a + # miss and lowers a belief the director then acts on. + assert (await _answer(client, opened["sessionId"], "")).status_code == 422 + assert (await _answer(client, opened["sessionId"], " ")).status_code == 422 + assert (await _answer(client, opened["sessionId"], "x" * 4001)).status_code == 422 + + +async def test_an_answer_that_could_not_be_scored_changes_nothing( + client_with_a_broken_grader: httpx.AsyncClient, +) -> None: + """The failure U1's design exists to survive. An outage must never read as a wrong answer — the + belief the director gates progress on would move against a learner because the provider had a + bad minute — so the turn is refused whole, retryably, with the transcript untouched.""" + # Arrange + graph = await _graph(client_with_a_broken_grader) + opened = ( + await client_with_a_broken_grader.post( + "/api/live/sessions", json={"graphId": graph["graphId"]} + ) + ).json() + + # Act + response = await _answer(client_with_a_broken_grader, opened["sessionId"], "A real attempt.") + + # Assert — retryable, and the session is where it was: one turn, unanswered, ungraded. + assert response.status_code == 503, response.text + resumed = ( + await client_with_a_broken_grader.get(f"/api/live/sessions/{opened['sessionId']}") + ).json() + assert len(resumed["turns"]) == 1 + assert resumed["turns"][0]["answer"] is None + assert resumed["turns"][0]["grade"] is None + + +async def test_answering_a_session_that_is_not_there_is_not_found( + client: httpx.AsyncClient, +) -> None: + # Act / Assert + assert (await _answer(client, "no-such-session", "Anything.")).status_code == 404 + + +async def test_a_session_this_build_cannot_read_is_not_offered_as_a_retry( + client_with_an_unreadable_session: httpx.AsyncClient, +) -> None: + """Told apart from storage being down, because the advice differs: an outage ends and a reload + fixes it, a row written by a schema this build no longer understands never becomes readable. As + a 503 it would invite a learner to reload forever.""" + # Act + response = await client_with_an_unreadable_session.get("/api/live/sessions/s1") + + # Assert + assert response.status_code == 500, response.text + assert response.headers["X-Session-Id"] == "s1" + + +async def test_the_first_concept_is_one_with_nothing_before_it(client: httpx.AsyncClient) -> None: + """The director may not open in the middle of the map. The whole point of Phase 1's ordering is + that a learner is never shown a concept whose prerequisites they have not met.""" + # Arrange + graph = await _graph(client) + + # Act + session = (await client.post("/api/live/sessions", json={"graphId": graph["graphId"]})).json() + + # Assert + opened_on = session["turns"][0]["move"]["nodeId"] + node = next(n for n in graph["nodes"] if n["id"] == opened_on) + assert node["requires"] == [], f"{opened_on} was taught before its prerequisites" + + +async def test_a_session_is_a_row_not_a_connection(client: httpx.AsyncClient) -> None: + """U2: a 25-40 minute session that dies on a refresh cannot be tested end to end, and Phase 1 + already learned this shape once when a dropped stream nearly cancelled its compile (AD12).""" + # Arrange + graph = await _graph(client) + opened = (await client.post("/api/live/sessions", json={"graphId": graph["graphId"]})).json() + + # Act — a second request, the way a reloaded tab re-opens the session it was in. + response = await client.get(f"/api/live/sessions/{opened['sessionId']}") + + # Assert — the same session, at the same turn, not a fresh one. + assert response.status_code == 200, response.text + resumed = response.json() + assert resumed["sessionId"] == opened["sessionId"] + assert resumed["turns"] == opened["turns"] + + +async def test_a_session_on_a_map_that_is_not_there_is_not_found( + client: httpx.AsyncClient, +) -> None: + # Act + response = await client.post("/api/live/sessions", json={"graphId": "no-such-map"}) + + # Assert — and the id rides the failure. A header set only once the work succeeded would be + # absent from exactly the responses somebody needs to report, which is the whole point of it. + assert response.status_code == 404 + assert response.headers["X-Session-Id"] + + +async def test_another_owners_session_is_not_found(client: httpx.AsyncClient) -> None: + """A session's existence is owner-scoped information, so a stranger gets 404 rather than 403 — + the same posture Phase 1 took for graphs.""" + assert (await client.get("/api/live/sessions/no-such-session")).status_code == 404 diff --git a/apps/web/src/components/live/AnswerForm.module.css b/apps/web/src/components/live/AnswerForm.module.css new file mode 100644 index 00000000..47ff3711 --- /dev/null +++ b/apps/web/src/components/live/AnswerForm.module.css @@ -0,0 +1,76 @@ +/* Welded to the transcript above it rather than floating below: the answer is part of the record, + not a separate widget. */ +.form { + display: flex; + flex-direction: column; + gap: var(--space-2); + padding: var(--space-4) var(--space-5); + background: var(--surface); + border: var(--border-width) solid var(--border); + border-top: none; + border-radius: 0 0 var(--radius-md) var(--radius-md); +} + +.label { + font-family: var(--font-mono); + font-size: 10px; + letter-spacing: 0.14em; + text-transform: uppercase; + color: var(--text-muted); +} + +.box { + width: 100%; + padding: var(--space-3); + font-family: inherit; + font-size: var(--text-sm); + line-height: var(--lh-sm); + color: var(--text); + background: var(--bg); + border: var(--border-width) solid var(--border-strong); + border-radius: var(--radius-sm); + resize: vertical; +} + +.box:focus-visible { + outline: none; + /* A ring rather than an outline, so it follows the radius. */ + box-shadow: 0 0 0 2px var(--focus-ring); + border-color: var(--focus-ring); +} + +.box:disabled { + color: var(--text-muted); + background: var(--bg-muted); + cursor: not-allowed; +} + +.footer { + display: flex; + align-items: center; + justify-content: space-between; + gap: var(--space-3); +} + +.hint { + display: flex; + align-items: center; + gap: var(--space-1); + margin: 0; + font-size: var(--text-xs); + color: var(--text-muted); +} + +.hint[role="alert"] { + color: var(--danger); +} + +.kbd { + padding: 1px var(--space-1); + font-family: var(--font-mono); + font-size: 10px; + color: var(--text-secondary); + background: var(--bg-muted); + border: var(--border-width) solid var(--border); + border-radius: var(--radius-sm); +} diff --git a/apps/web/src/components/live/AnswerForm.tsx b/apps/web/src/components/live/AnswerForm.tsx new file mode 100644 index 00000000..c56c53d2 --- /dev/null +++ b/apps/web/src/components/live/AnswerForm.tsx @@ -0,0 +1,92 @@ +import { useId, useRef, useState, type KeyboardEvent } from "react"; + +import { MAX_ANSWER_CHARS } from "../../lib/liveSession"; +import { Button } from "../primitives/Button"; +import styles from "./AnswerForm.module.css"; + +interface AnswerFormProps { + /** What the learner is being asked to demonstrate, or null when the turn stages nothing. */ + criterion: string | null; + /** True while an answer is being marked — the box locks so one answer cannot be sent twice. */ + busy: boolean; + onAnswer: (text: string) => void; +} + +/** Where the learner replies. A textarea, because an answer is prose and the whole loop rests on + * them being able to say it in their own words. + * + * Enter stays a newline and ⌘/Ctrl+Enter sends: a stray Return mid-thought that submitted the + * answer would be the surface answering for them. Submit is never pre-disabled — a form that + * greys out its own button hides the reason it is not ready — so an empty send explains itself + * instead. */ +export function AnswerForm({ criterion, busy, onAnswer }: AnswerFormProps) { + const [text, setText] = useState(""); + const [error, setError] = useState(null); + const boxId = useId(); + const errorId = useId(); + const box = useRef(null); + + const submit = () => { + if (busy) return; + if (!text.trim()) { + setError("Write something first — even a guess is worth marking."); + box.current?.focus(); + return; + } + setError(null); + onAnswer(text.trim()); + setText(""); + }; + + const onKeyDown = (event: KeyboardEvent) => { + if (event.key === "Enter" && (event.metaKey || event.ctrlKey)) { + event.preventDefault(); + submit(); + } + }; + + return ( +
{ + event.preventDefault(); + submit(); + }} + > + +