From 94b03f53088d3d838cd71df4df91979ffb8a14a8 Mon Sep 17 00:00:00 2001 From: Rafael Bittencourt Date: Wed, 4 Mar 2026 18:30:37 -0300 Subject: [PATCH] =?UTF-8?q?feat:=20Sprint=2029A-D=20=E2=80=94=2027=20multi?= =?UTF-8?q?-model=20audit=20fixes=20(security,=20reliability,=20contracts,?= =?UTF-8?q?=20polish)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P0 Critical (29A): migration composite FK ordering, embedding dim=1536, JWT revocation fail-closed with LRU cache, tenant-scoped trace/evidence stores, setup token log redaction, aria-live SSE chat fix. High Reliability (29B): PBKDF2 async offload, JWT startup validation, CircuitBreaker 3-state machine (half-open probe), Redis PubSub real subscribe, SSE client reconnection with exponential backoff, React.memo ChatMessageBubble. Contract Alignment (29C): setup wizard endpoint/payload fix, idempotency guard, privacy panel auth+abort, safe defaults (demo=0, dev=false), GEMINI_API_KEY alias. Medium Polish (29D): SSE terminal ordering+backpressure, async tutor I/O, auth-first in tutor routes, login form cleanup, api.ts type safety, composite DB indexes, accessibility store cleanup. Deferred: D3 (auth.py DI refactor) — follow-up sprint. Co-Authored-By: Claude Opus 4.6 --- .env.example | 9 +- agents/ailine_agents/resilience.py | 88 +++++++++----- docker-compose.yml | 99 ++++++++++++---- frontend/src/app/[locale]/login/page.tsx | 3 +- .../src/components/privacy/privacy-panel.tsx | 19 ++- .../src/components/setup/setup-wizard.tsx | 37 +++++- .../components/tutor/chat-message-bubble.tsx | 6 +- frontend/src/components/tutor/tutor-chat.tsx | 6 +- frontend/src/lib/api.ts | 4 +- frontend/src/lib/sse-fetch.ts | 109 ++++++++++++------ frontend/src/stores/accessibility-store.ts | 12 +- .../2026_02_12_0001_initial_schema.py | 4 + ...006_fix_composite_fk_unique_constraints.py | 2 +- .../2026_03_04_0007_composite_indexes.py | 30 +++++ .../adapters/events/redis_bus.py | 27 ++++- .../api/middleware/tenant_context.py | 48 ++++++-- runtime/ailine_runtime/api/routers/auth.py | 20 +++- .../api/routers/observability.py | 4 +- .../api/routers/plans_stream.py | 30 ++++- .../api/routers/setup_service.py | 14 +-- runtime/ailine_runtime/api/routers/tutors.py | 27 ++--- runtime/ailine_runtime/shared/config.py | 9 +- .../shared/observability_store.py | 38 ++++-- runtime/ailine_runtime/shared/trace_store.py | 41 +++++-- runtime/tests/test_redis_event_bus.py | 6 +- 25 files changed, 510 insertions(+), 182 deletions(-) create mode 100644 runtime/ailine_runtime/adapters/db/alembic/versions/2026_03_04_0007_composite_indexes.py diff --git a/.env.example b/.env.example index 390b1ad..1e6517c 100644 --- a/.env.example +++ b/.env.example @@ -23,7 +23,8 @@ AILINE_REDIS_URL="redis://:ailine_redis_dev@localhost:6311/0" # === LLM Provider Keys === ANTHROPIC_API_KEY="" OPENAI_API_KEY="" -GEMINI_API_KEY="" +GOOGLE_API_KEY="" +# GEMINI_API_KEY is also accepted as an alias for GOOGLE_API_KEY OPENROUTER_API_KEY="" # === Models === @@ -39,7 +40,7 @@ AILINE_LLM_PROVIDER="anthropic" # === Embedding Provider (gemini|openai|local|openrouter) === AILINE_EMBEDDING_PROVIDER="gemini" AILINE_EMBEDDING_MODEL="gemini-embedding-001" -AILINE_EMBEDDING_DIMENSIONS="3072" +AILINE_EMBEDDING_DIMENSIONS="1536" # === Vector Store (pgvector|qdrant|chroma) === AILINE_VECTORSTORE_PROVIDER="pgvector" @@ -69,10 +70,10 @@ HANDTALK_API_KEY="" AILINE_JWT_SECRET="" # === Dev Mode (enables X-Teacher-ID header bypass for local dev) === -AILINE_DEV_MODE="true" +AILINE_DEV_MODE="false" # === Demo Mode (enables /demo/* endpoints and demo login) === -AILINE_DEMO_MODE="1" +AILINE_DEMO_MODE="0" # === Rate Limiting === AILINE_RATE_LIMIT_RPM="60" diff --git a/agents/ailine_agents/resilience.py b/agents/ailine_agents/resilience.py index 9fa8241..cec12f7 100644 --- a/agents/ailine_agents/resilience.py +++ b/agents/ailine_agents/resilience.py @@ -43,67 +43,99 @@ def __init__( self._cooldown_seconds = cooldown_seconds self._failure_count = 0 self._circuit_open_until: float | None = None + self._state: str = "closed" # closed, open, half_open + self._probe_in_flight: bool = False self._lock = threading.Lock() @property def failure_count(self) -> int: - """Current consecutive failure count.""" return self._failure_count + @property + def state(self) -> str: + """Current circuit breaker state: closed, open, or half_open.""" + with self._lock: + self._maybe_transition_to_half_open() + return self._state + @property def is_open(self) -> bool: - """True if the circuit is open and blocking calls.""" with self._lock: - return self._is_open_locked() + self._maybe_transition_to_half_open() + return self._state == "open" - def _is_open_locked(self) -> bool: - """Check if circuit is open (must hold _lock).""" - if self._circuit_open_until is None: - return False - # Cooldown expired -- transition to half-open - return time.monotonic() < self._circuit_open_until + def _maybe_transition_to_half_open(self) -> None: + """Transition from open to half_open if cooldown expired (must hold _lock).""" + if self._state == "open" and self._circuit_open_until is not None: + if time.monotonic() >= self._circuit_open_until: + self._state = "half_open" + self._probe_in_flight = False + log.info("circuit_breaker.half_open", failure_count=self._failure_count) def check(self) -> bool: - """Check if a call is allowed. - - Returns True if the call can proceed, False if the circuit is open. - """ with self._lock: - if self._is_open_locked(): - log.warning( - "circuit_breaker.blocked", - failure_count=self._failure_count, - open_until=self._circuit_open_until, - remaining_seconds=round( - (self._circuit_open_until or 0) - time.monotonic(), 1 - ), - ) + self._maybe_transition_to_half_open() + + if self._state == "closed": + return True + + if self._state == "half_open": + if not self._probe_in_flight: + self._probe_in_flight = True + log.info("circuit_breaker.probe_allowed") + return True + log.warning("circuit_breaker.blocked_half_open", msg="Probe already in flight") return False - return True + + # state == "open" + log.warning( + "circuit_breaker.blocked", + failure_count=self._failure_count, + open_until=self._circuit_open_until, + remaining_seconds=round( + (self._circuit_open_until or 0) - time.monotonic(), 1 + ), + ) + return False def record_success(self) -> None: - """Record a successful call -- resets the failure counter.""" with self._lock: + prev_state = self._state prev_count = self._failure_count self._failure_count = 0 self._circuit_open_until = None - if prev_count > 0: + self._state = "closed" + self._probe_in_flight = False + if prev_count > 0 or prev_state != "closed": log.info( "circuit_breaker.reset", previous_failures=prev_count, + previous_state=prev_state, ) def record_failure(self) -> None: - """Record a failed call. Opens the circuit if threshold is reached.""" with self._lock: self._failure_count += 1 + + if self._state == "half_open": + # Probe failed -- reopen + self._state = "open" + self._circuit_open_until = time.monotonic() + self._cooldown_seconds + self._probe_in_flight = False + log.error( + "circuit_breaker.probe_failed", + failure_count=self._failure_count, + cooldown_seconds=self._cooldown_seconds, + ) + return + if self._failure_count >= self._failure_threshold: + self._state = "open" self._circuit_open_until = time.monotonic() + self._cooldown_seconds log.error( "circuit_breaker.opened", failure_count=self._failure_count, cooldown_seconds=self._cooldown_seconds, - open_until=self._circuit_open_until, ) def reset(self) -> None: @@ -111,6 +143,8 @@ def reset(self) -> None: with self._lock: self._failure_count = 0 self._circuit_open_until = None + self._state = "closed" + self._probe_in_flight = False class CircuitOpenError(Exception): diff --git a/docker-compose.yml b/docker-compose.yml index c64f916..a53c35e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,9 +1,40 @@ -# AiLine -- Docker Compose (full dev stack) -# Services: API + Frontend + PostgreSQL 16/pgvector + Redis 7 -# Usage: docker compose up -d --build +# AiLine -- Docker Compose (full stack) +# Services: DB (Postgres 16 + pgvector) | Redis 7 | Migrations | API | Frontend +# Usage: +# docker compose up -d --build # Full stack (dev, with override) +# docker compose -f docker-compose.yml up -d --build # Production-like +# docker compose ps # Health status +# docker compose logs -f api # Follow API logs name: ailine +# --------------------------------------------------------------------------- +# Shared anchors (DRY) +# --------------------------------------------------------------------------- +x-logging-default: &logging-default + driver: json-file + options: + max-size: "10m" + max-file: "3" + +x-db-env: &db-env + AILINE_DB_URL: postgresql+asyncpg://${POSTGRES_USER:-ailine}:${POSTGRES_PASSWORD:-ailine_dev}@db:5432/${POSTGRES_DB:-ailine} + +x-redis-env: &redis-env + AILINE_REDIS_URL: redis://:${REDIS_PASSWORD:-ailine_redis_dev}@redis:6379/0 + +x-api-build: &api-build + context: . + dockerfile: runtime/Dockerfile + args: + INSTALL_DEV: "true" + +# --------------------------------------------------------------------------- +# Services +# --------------------------------------------------------------------------- services: + + # --- Data Layer ----------------------------------------------------------- + db: image: pgvector/pgvector:0.8.2-pg16 environment: @@ -23,11 +54,7 @@ services: restart: unless-stopped networks: - backend - logging: - driver: json-file - options: - max-size: "10m" - max-file: "3" + logging: *logging-default deploy: resources: limits: @@ -37,7 +64,11 @@ services: image: redis:7.4-alpine environment: REDIS_PASSWORD: ${REDIS_PASSWORD:-ailine_redis_dev} - command: redis-server --maxmemory 128mb --maxmemory-policy allkeys-lru --requirepass ${REDIS_PASSWORD:-ailine_redis_dev} + command: >- + redis-server + --maxmemory 128mb + --maxmemory-policy allkeys-lru + --requirepass ${REDIS_PASSWORD:-ailine_redis_dev} ports: - "127.0.0.1:${REDIS_HOST_PORT:-6311}:6379" healthcheck: @@ -58,21 +89,42 @@ services: limits: memory: 256M + # --- Migrations (one-shot, runs before API) -------------------------------- + + migrations: + build: *api-build + command: >- + python -m alembic -c alembic.ini upgrade head + working_dir: /app + environment: + <<: *db-env + env_file: + - path: .env + required: false + depends_on: + db: + condition: service_healthy + restart: "no" + networks: + - backend + logging: *logging-default + deploy: + resources: + limits: + memory: 256M + + # --- Backend API ----------------------------------------------------------- + api: - build: - context: . - dockerfile: runtime/Dockerfile - args: - INSTALL_DEV: "true" + build: *api-build ports: - "${API_HOST_PORT:-8011}:8000" environment: - AILINE_DB_URL: postgresql+asyncpg://${POSTGRES_USER:-ailine}:${POSTGRES_PASSWORD:-ailine_dev}@db:5432/${POSTGRES_DB:-ailine} - AILINE_REDIS_URL: redis://:${REDIS_PASSWORD:-ailine_redis_dev}@redis:6379/0 + <<: [*db-env, *redis-env] AILINE_LLM_PROVIDER: ${AILINE_LLM_PROVIDER:-anthropic} AILINE_CORS_ORIGINS: "http://localhost:3011,http://127.0.0.1:3011,http://localhost:3000,http://127.0.0.1:3000,http://frontend:3000" AILINE_DEV_MODE: "${AILINE_DEV_MODE:-false}" - AILINE_DEMO_MODE: "${AILINE_DEMO_MODE:-1}" + AILINE_DEMO_MODE: "${AILINE_DEMO_MODE:-0}" AILINE_JWT_SECRET: "${AILINE_JWT_SECRET:-}" AILINE_LOCAL_STORE: /app/.local_store env_file: @@ -83,6 +135,8 @@ services: condition: service_healthy redis: condition: service_healthy + migrations: + condition: service_completed_successfully healthcheck: test: ["CMD", "python", "-c", "import httpx; httpx.get('http://localhost:8000/health').raise_for_status()"] interval: 10s @@ -104,6 +158,8 @@ services: limits: memory: 1G + # --- Frontend (Next.js 16) ------------------------------------------------ + frontend: build: context: ./frontend @@ -126,16 +182,15 @@ services: restart: unless-stopped networks: - frontend - logging: - driver: json-file - options: - max-size: "10m" - max-file: "3" + logging: *logging-default deploy: resources: limits: memory: 2G +# --------------------------------------------------------------------------- +# Volumes & Networks +# --------------------------------------------------------------------------- volumes: pgdata: diff --git a/frontend/src/app/[locale]/login/page.tsx b/frontend/src/app/[locale]/login/page.tsx index 8052037..4b8b1c7 100644 --- a/frontend/src/app/[locale]/login/page.tsx +++ b/frontend/src/app/[locale]/login/page.tsx @@ -6,7 +6,7 @@ import { motion, AnimatePresence, useReducedMotion } from 'motion/react' import { useTranslations } from 'next-intl' import { cn } from '@/lib/cn' import { useAuthStore, type UserRole } from '@/stores/auth-store' -import { API_BASE, demoLogin, setDemoProfile, getAuthHeaders } from '@/lib/api' +import { API_BASE, demoLogin, setDemoProfile } from '@/lib/api' import { useAccessibilityStore } from '@/stores/accessibility-store' import { cssTheme } from '@/hooks/use-theme' import { DEMO_PROFILES_BY_ROLE, type DemoProfile } from '@/components/auth/login-data' @@ -80,7 +80,6 @@ export default function LoginPage() { method: 'POST', headers: { 'Content-Type': 'application/json', - ...getAuthHeaders(), }, body: JSON.stringify({ email, password, role: selectedRole }), }) diff --git a/frontend/src/components/privacy/privacy-panel.tsx b/frontend/src/components/privacy/privacy-panel.tsx index 488d8de..0c342c9 100644 --- a/frontend/src/components/privacy/privacy-panel.tsx +++ b/frontend/src/components/privacy/privacy-panel.tsx @@ -3,7 +3,7 @@ import { useState, useEffect, useCallback } from 'react' import { useTranslations } from 'next-intl' import { cn } from '@/lib/cn' -import { API_BASE } from '@/lib/api' +import { API_BASE, getAuthHeaders } from '@/lib/api' interface DataSummary { plans: number @@ -23,9 +23,13 @@ export function PrivacyPanel() { const [actionState, setActionState] = useState<'idle' | 'exporting' | 'deleting' | 'confirm_delete'>('idle') useEffect(() => { + const controller = new AbortController() async function fetchSummary() { try { - const res = await fetch(`${API_BASE}/api/v1/privacy/data-summary`) + const res = await fetch(`${API_BASE}/privacy/data-summary`, { + headers: getAuthHeaders(), + signal: controller.signal, + }) if (res.ok) { const data: DataSummary = await res.json() setSummary(data) @@ -37,12 +41,16 @@ export function PrivacyPanel() { } } fetchSummary() + return () => controller.abort() }, []) const handleExport = useCallback(async () => { setActionState('exporting') try { - await fetch(`${API_BASE}/api/v1/privacy/export`, { method: 'POST' }) + await fetch(`${API_BASE}/privacy/export`, { + method: 'POST', + headers: getAuthHeaders(), + }) } catch { // Demo mode: export simulated } finally { @@ -57,7 +65,10 @@ export function PrivacyPanel() { } setActionState('deleting') try { - await fetch(`${API_BASE}/api/v1/privacy/delete`, { method: 'DELETE' }) + await fetch(`${API_BASE}/privacy/delete`, { + method: 'DELETE', + headers: getAuthHeaders(), + }) setSummary({ plans: 0, sessions: 0, materials: 0, last_updated: new Date().toISOString() }) } catch { // Demo mode: delete simulated diff --git a/frontend/src/components/setup/setup-wizard.tsx b/frontend/src/components/setup/setup-wizard.tsx index ebfe906..5560bd2 100644 --- a/frontend/src/components/setup/setup-wizard.tsx +++ b/frontend/src/components/setup/setup-wizard.tsx @@ -15,6 +15,7 @@ import { StepSecurity } from './steps/step-security' import { StepReview } from './steps/step-review' import { DEFAULT_CONFIG, TOTAL_STEPS } from './setup-types' import type { SetupConfig, LlmProvider, EmbeddingProvider } from './setup-types' +import { API_BASE } from '@/lib/api' interface SetupWizardProps { locale: string @@ -26,6 +27,35 @@ type ApplyStatus = 'idle' | 'writing' | 'done' | 'error' * Main setup wizard orchestrator. * Manages step navigation, state, validation, and API calls. */ +/** Map camelCase setup config to snake_case for the backend API. */ +function toSetupPayload(config: SetupConfig): Record { + return { + llm_provider: config.llmProvider, + llm_model: '', + embedding_provider: config.embeddingProvider, + embedding_model: config.embeddingModel, + embedding_dimensions: config.embeddingDimensions, + planner_model: config.plannerModel, + executor_model: config.executorModel, + quality_model: config.qualityModel, + tutor_model: config.tutorModel, + anthropic_api_key: config.llmProvider === 'anthropic' ? config.llmApiKey : '', + openai_api_key: config.llmProvider === 'openai' ? config.llmApiKey : '', + google_api_key: config.llmProvider === 'gemini' ? config.llmApiKey : '', + openrouter_api_key: config.llmProvider === 'openrouter' ? config.llmApiKey : '', + db_url: config.databaseUrl, + redis_url: config.redisUrl, + api_host_port: config.apiPort, + frontend_host_port: config.frontendPort, + jwt_secret: config.jwtSecret, + cors_origins: config.corsOrigins, + elevenlabs_api_key: config.elevenlabsKey, + locale: config.language, + dev_mode: false, + demo_mode: false, + } +} + export function SetupWizard({ locale }: SetupWizardProps) { const t = useTranslations('setup') const prefersReducedMotion = useReducedMotion() @@ -100,11 +130,10 @@ export function SetupWizard({ locale }: SetupWizardProps) { setApplying(true) setApplyStatus('writing') try { - const apiBase = process.env.NEXT_PUBLIC_API_URL ?? 'http://localhost:8011' - const res = await fetch(`${apiBase}/api/v1/setup/apply`, { + const res = await fetch(`${API_BASE}/setup/apply`, { method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify(config), + headers: { 'Content-Type': 'application/json', 'X-Setup-Token': '' }, + body: JSON.stringify(toSetupPayload(config)), }) if (res.ok) { setApplyStatus('done') diff --git a/frontend/src/components/tutor/chat-message-bubble.tsx b/frontend/src/components/tutor/chat-message-bubble.tsx index bf2f053..61a18d4 100644 --- a/frontend/src/components/tutor/chat-message-bubble.tsx +++ b/frontend/src/components/tutor/chat-message-bubble.tsx @@ -1,6 +1,6 @@ 'use client' -import { useCallback, useState, useMemo } from 'react' +import { memo, useCallback, useState, useMemo } from 'react' import { useTranslations } from 'next-intl' import { cn } from '@/lib/cn' import { MarkdownWithMermaid } from '@/components/shared/markdown-with-mermaid' @@ -16,7 +16,7 @@ interface ChatMessageBubbleProps { * User messages right-aligned, assistant messages left-aligned. * Supports TTS read-aloud and mermaid diagram rendering. */ -export function ChatMessageBubble({ +function ChatMessageBubbleInner({ message, isStreaming = false, }: ChatMessageBubbleProps) { @@ -170,6 +170,8 @@ export function ChatMessageBubble({ ) } +export const ChatMessageBubble = memo(ChatMessageBubbleInner) + function UserIcon({ className = '' }: { className?: string }) { return (
diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 9f39249..fd03014 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -1,6 +1,6 @@ /** Shared API configuration for all client-side fetch calls. */ -import { useAuthStore } from '../stores/auth-store' +import { useAuthStore, type AuthUser } from '../stores/auth-store' export const API_BASE = '/api' @@ -92,7 +92,7 @@ export async function demoLogin( } // Store in auth store for proper JWT auth flow const { login } = useAuthStore.getState() - login(data.access_token, data.user as unknown as Parameters[1]) + login(data.access_token, data.user as AuthUser) // Also keep demo profile in sessionStorage as fallback sessionStorage.setItem(DEMO_PROFILE_KEY, profileKey) return { token: data.access_token, user: data.user } diff --git a/frontend/src/lib/sse-fetch.ts b/frontend/src/lib/sse-fetch.ts index 39a3ad0..1b8c5ff 100644 --- a/frontend/src/lib/sse-fetch.ts +++ b/frontend/src/lib/sse-fetch.ts @@ -99,54 +99,91 @@ export async function fetchEventSource( ...fetchInit } = opts - const response = await fetch(url, { ...fetchInit, signal }) + const MAX_RETRIES = 5 + const BASE_RETRY_MS = 3000 + let retryMs = BASE_RETRY_MS + let lastEventId = '' + let attempt = 0 + + while (!signal?.aborted) { + try { + const headers: Record = { + ...(fetchInit.headers as Record ?? {}), + } + if (lastEventId) { + headers['Last-Event-ID'] = lastEventId + } - if (onopen) { - await onopen(response) - } + const response = await fetch(url, { ...fetchInit, headers, signal }) + + if (onopen) { + await onopen(response) + } - // If the response isn't OK and onopen didn't throw, bail out - if (!response.ok || !response.body) return + if (!response.ok || !response.body) return - const reader = response.body.getReader() - const decoder = new TextDecoder() - let buffer = '' + // Reset retry count on successful connection + attempt = 0 + retryMs = BASE_RETRY_MS - try { - for (;;) { - const { done, value } = await reader.read() + const reader = response.body.getReader() + const decoder = new TextDecoder() + let buffer = '' - if (done) break + for (;;) { + const { done, value } = await reader.read() + if (done) break - buffer += decoder.decode(value, { stream: true }) + buffer += decoder.decode(value, { stream: true }) - // SSE events are separated by double newlines - const parts = buffer.split('\n\n') - // Last part is either empty (complete event) or partial (keep buffering) - buffer = parts.pop() ?? '' + const parts = buffer.split('\n\n') + buffer = parts.pop() ?? '' - for (const part of parts) { - if (!part.trim()) continue - const msg = parseSseEvent(part) - if (msg && onmessage) { - onmessage(msg) + for (const part of parts) { + if (!part.trim()) continue + const msg = parseSseEvent(part) + if (msg) { + if (msg.id) lastEventId = msg.id + if (msg.retry !== undefined) retryMs = msg.retry + if (onmessage) onmessage(msg) + } } } - } - // Flush any remaining buffer - if (buffer.trim()) { - const msg = parseSseEvent(buffer) - if (msg && onmessage) { - onmessage(msg) + // Flush remaining buffer + if (buffer.trim()) { + const msg = parseSseEvent(buffer) + if (msg) { + if (msg.id) lastEventId = msg.id + if (onmessage) onmessage(msg) + } } - } - } catch (err) { - if (signal?.aborted) return - if (onerror) { - onerror(err) - } else { - throw err + + // Stream ended normally — no retry needed + return + } catch (err) { + if (signal?.aborted) return + + attempt++ + if (onerror) { + onerror(err) + } + + if (attempt >= MAX_RETRIES) { + // Max retries exceeded — give up + if (!onerror) throw err + return + } + + // Exponential backoff: 3s, 6s, 12s, 24s, 48s + const delay = retryMs * Math.pow(2, attempt - 1) + await new Promise((resolve) => { + const timer = setTimeout(resolve, delay) + signal?.addEventListener('abort', () => { + clearTimeout(timer) + resolve() + }, { once: true }) + }) } } } diff --git a/frontend/src/stores/accessibility-store.ts b/frontend/src/stores/accessibility-store.ts index 27ed896..e900fa4 100644 --- a/frontend/src/stores/accessibility-store.ts +++ b/frontend/src/stores/accessibility-store.ts @@ -7,6 +7,8 @@ import { cssTheme } from '@/hooks/use-theme' * Theme switching is done via DOM attribute, not React state (ADR-019). */ +let _mqCleanup: (() => void) | null = null + const STORAGE_KEY = 'ailine-a11y-prefs' interface A11yPrefs { @@ -147,13 +149,17 @@ export const useAccessibilityStore = create((set, get) => ({ } } mql.addEventListener('change', handler) - // Store cleanup function for testing - ;(useAccessibilityStore as unknown as Record) - ._mqCleanup = () => mql.removeEventListener('change', handler) + _mqCleanup = () => mql.removeEventListener('change', handler) } }, })) +/** Remove media query listener (for testing). */ +export function cleanupAccessibilityStore(): void { + _mqCleanup?.() + _mqCleanup = null +} + const fontSizeMap: Record = { small: '14px', medium: '16px', diff --git a/runtime/ailine_runtime/adapters/db/alembic/versions/2026_02_12_0001_initial_schema.py b/runtime/ailine_runtime/adapters/db/alembic/versions/2026_02_12_0001_initial_schema.py index 825ceb1..bb8c653 100644 --- a/runtime/ailine_runtime/adapters/db/alembic/versions/2026_02_12_0001_initial_schema.py +++ b/runtime/ailine_runtime/adapters/db/alembic/versions/2026_02_12_0001_initial_schema.py @@ -66,6 +66,9 @@ def upgrade() -> None: ), ) + # UniqueConstraint needed by composite FK on lessons (PG16 requires it) + op.create_unique_constraint("uq_courses_teacher_id", "courses", ["teacher_id", "id"]) + # --- lessons (composite FK for tenant safety, ADR-053) ----------------- op.create_table( "lessons", @@ -267,5 +270,6 @@ def downgrade() -> None: op.drop_table("chunks") op.drop_table("materials") op.drop_table("lessons") + op.drop_constraint("uq_courses_teacher_id", "courses", type_="unique") op.drop_table("courses") op.drop_table("teachers") diff --git a/runtime/ailine_runtime/adapters/db/alembic/versions/2026_02_27_0006_fix_composite_fk_unique_constraints.py b/runtime/ailine_runtime/adapters/db/alembic/versions/2026_02_27_0006_fix_composite_fk_unique_constraints.py index 6d1d2f4..1713407 100644 --- a/runtime/ailine_runtime/adapters/db/alembic/versions/2026_02_27_0006_fix_composite_fk_unique_constraints.py +++ b/runtime/ailine_runtime/adapters/db/alembic/versions/2026_02_27_0006_fix_composite_fk_unique_constraints.py @@ -24,7 +24,7 @@ # Tables that need UniqueConstraint("teacher_id", "id") for composite FK targets _TABLES = [ - ("courses", "uq_courses_teacher_id"), + # ("courses", "uq_courses_teacher_id") — moved to 0001 (before composite FK) ("lessons", "uq_lessons_teacher_id"), ("materials", "uq_materials_teacher_id"), ("tutor_agents", "uq_tutor_agents_teacher_id"), diff --git a/runtime/ailine_runtime/adapters/db/alembic/versions/2026_03_04_0007_composite_indexes.py b/runtime/ailine_runtime/adapters/db/alembic/versions/2026_03_04_0007_composite_indexes.py new file mode 100644 index 0000000..580c19c --- /dev/null +++ b/runtime/ailine_runtime/adapters/db/alembic/versions/2026_03_04_0007_composite_indexes.py @@ -0,0 +1,30 @@ +"""Add composite indexes for common query patterns. + +Revision ID: 0007 +Revises: 0006 +Create Date: 2026-03-04 +""" +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op + +revision: str = "0007" +down_revision: str = "0006" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_index("ix_materials_teacher_subject", "materials", ["teacher_id", "subject"]) + op.create_index( + "ix_pipeline_runs_teacher_status", + "pipeline_runs", + ["teacher_id", "status", "created_at"], + ) + + +def downgrade() -> None: + op.drop_index("ix_pipeline_runs_teacher_status", table_name="pipeline_runs") + op.drop_index("ix_materials_teacher_subject", table_name="materials") diff --git a/runtime/ailine_runtime/adapters/events/redis_bus.py b/runtime/ailine_runtime/adapters/events/redis_bus.py index ddfa384..c920759 100644 --- a/runtime/ailine_runtime/adapters/events/redis_bus.py +++ b/runtime/ailine_runtime/adapters/events/redis_bus.py @@ -43,12 +43,37 @@ async def publish(self, event_type: str, data: dict[str, Any]) -> None: except Exception: _log.exception("event_handler_failed", event_type=event_type) - def subscribe( + async def subscribe( self, event_type: str, handler: Callable[[dict[str, Any]], Awaitable[None]], ) -> None: self._handlers[event_type].append(handler) + await self._pubsub.subscribe(event_type) + if self._listener_task is None or self._listener_task.done(): + self._listener_task = asyncio.create_task(self._listen()) + + async def _listen(self) -> None: + """Background listener that dispatches PubSub messages to handlers.""" + try: + async for message in self._pubsub.listen(): + if message["type"] != "message": + continue + try: + payload = json.loads(message["data"]) + event_type = payload.get("event_type", "") + data = payload.get("data", {}) + for handler in self._handlers.get(event_type, []): + try: + await handler(data) + except Exception: + _log.exception("event_handler_failed", event_type=event_type) + except (json.JSONDecodeError, KeyError): + _log.warning("invalid_pubsub_message", data=message.get("data")) + except asyncio.CancelledError: + return + except Exception: + _log.exception("pubsub_listener_crashed") async def ping(self) -> bool: """Check Redis connectivity.""" diff --git a/runtime/ailine_runtime/api/middleware/tenant_context.py b/runtime/ailine_runtime/api/middleware/tenant_context.py index b58528d..35dabea 100644 --- a/runtime/ailine_runtime/api/middleware/tenant_context.py +++ b/runtime/ailine_runtime/api/middleware/tenant_context.py @@ -413,30 +413,56 @@ def extract_claims_from_jwt(token: str) -> tuple[_JwtClaims, str | None]: return _extract_teacher_id_from_jwt(token) +# In-memory LRU fallback cache for JTI blacklist when Redis is down +_REVOCATION_CACHE: dict[str, float] = {} +_REVOCATION_CACHE_MAX = 1000 +_REVOCATION_CACHE_TTL = 300.0 # 5 minutes + + async def _is_jti_blacklisted(request: Request, jti: str) -> bool: """Check if a JWT ID (jti) has been revoked via Redis blacklist. - Returns False when Redis is unavailable (fail-open for availability). - The blacklist key format is ``jti_blacklist:{jti}`` with a TTL matching - the token's remaining lifetime (set by POST /auth/logout). + Fail-closed: returns True (treat as blacklisted) when Redis is + unavailable, to prevent accepting potentially revoked tokens. + Uses an in-memory LRU cache as fallback for known-good JTIs. """ - # F-258: Use public EventBus.get_redis_client() instead of private _redis + import time as _time + + # Check in-memory cache first (known-good JTIs) + cached_ts = _REVOCATION_CACHE.get(jti) + if cached_ts is not None: + if _time.monotonic() - cached_ts < _REVOCATION_CACHE_TTL: + return False # Recently verified as not blacklisted + else: + del _REVOCATION_CACHE[jti] + container = getattr(getattr(request.app, "state", None), "container", None) if container is None: - return False + logger.warning("jti_blacklist_check_failed_closed", jti=jti, reason="no_container") + return True # Fail-closed event_bus = getattr(container, "event_bus", None) if event_bus is None: - return False + logger.warning("jti_blacklist_check_failed_closed", jti=jti, reason="no_event_bus") + return True # Fail-closed redis_client = await event_bus.get_redis_client() if redis_client is None: - return False + logger.warning("jti_blacklist_check_failed_closed", jti=jti, reason="no_redis_client") + return True # Fail-closed try: result = await redis_client.get(f"jti_blacklist:{jti}") - return result is not None - except Exception: - # Fail-open: if Redis is down, allow the request through - logger.warning("jti_blacklist_check_failed", jti=jti) + if result is not None: + return True # Token IS blacklisted + + # Cache as known-good + if len(_REVOCATION_CACHE) >= _REVOCATION_CACHE_MAX: + # Evict oldest entry + oldest_key = min(_REVOCATION_CACHE, key=_REVOCATION_CACHE.get) # type: ignore[arg-type] + del _REVOCATION_CACHE[oldest_key] + _REVOCATION_CACHE[jti] = _time.monotonic() return False + except Exception: + logger.warning("jti_blacklist_check_failed_closed", jti=jti, reason="redis_error") + return True # Fail-closed on Redis error class TenantContextMiddleware(BaseHTTPMiddleware): diff --git a/runtime/ailine_runtime/api/routers/auth.py b/runtime/ailine_runtime/api/routers/auth.py index d7900ca..2fd0b7a 100644 --- a/runtime/ailine_runtime/api/routers/auth.py +++ b/runtime/ailine_runtime/api/routers/auth.py @@ -197,6 +197,16 @@ def _verify_password(password: str, stored_hash: str) -> bool: return hmac.compare_digest(new_hash, stored_hash) +async def _hash_password_async(password: str, salt: bytes | None = None) -> str: + """Async wrapper for _hash_password -- offloads to threadpool.""" + return await asyncio.to_thread(_hash_password, password, salt) + + +async def _verify_password_async(password: str, stored_hash: str) -> bool: + """Async wrapper for _verify_password -- offloads to threadpool.""" + return await asyncio.to_thread(_verify_password, password, stored_hash) + + def _create_jwt( user_id: str, role: str, @@ -336,7 +346,7 @@ async def login(body: LoginRequest, request: Request) -> TokenResponse: avatar_url="", accessibility_profile="", is_active=True, - hashed_password=_hash_password(body.password), + hashed_password=await _hash_password_async(body.password), ) await _user_repo.create(user) logger.info("auth.auto_created_user", user_id=user.id, role=validated_role) @@ -344,7 +354,7 @@ async def login(body: LoginRequest, request: Request) -> TokenResponse: # Verify password: if user has a hashed password, require correct password stored_hash = user.hashed_password or "" if stored_hash: - if not _verify_password(body.password, stored_hash): + if not await _verify_password_async(body.password, stored_hash): raise HTTPException(status_code=401, detail="Invalid credentials") else: # User has no password (demo user) — only allow in dev mode. @@ -381,7 +391,7 @@ async def register(body: RegisterRequest) -> TokenResponse: avatar_url="", accessibility_profile="", is_active=True, - hashed_password=_hash_password(body.password), + hashed_password=await _hash_password_async(body.password), ) await _user_repo.create(user) logger.info("auth.registered", user_id=user.id, role=validated_role) @@ -583,7 +593,7 @@ async def demo_login(body: DemoLoginRequest, request: Request) -> TokenResponse: avatar_url="", accessibility_profile=profile.get("accessibility", ""), is_active=True, - hashed_password=_hash_password("demo123"), + hashed_password=await _hash_password_async("demo123"), ) await _user_repo.create(user) logger.info("auth.demo_login_created_user", demo_key=canonical_key) @@ -640,7 +650,7 @@ async def seed_demo_users_async() -> None: """ from .demo_profiles import DEMO_PROFILES - demo_pw_hash = _hash_password("demo123") + demo_pw_hash = await _hash_password_async("demo123") seeded = 0 for key, profile in DEMO_PROFILES.items(): email = f"{key}@ailine-demo.edu" diff --git a/runtime/ailine_runtime/api/routers/observability.py b/runtime/ailine_runtime/api/routers/observability.py index 47c8e88..dd6f484 100644 --- a/runtime/ailine_runtime/api/routers/observability.py +++ b/runtime/ailine_runtime/api/routers/observability.py @@ -206,7 +206,7 @@ async def standards_evidence( # Build standards alignment evidence obs_store = get_observability_store() - evidence = obs_store.get_standards_evidence(run_id) + evidence = obs_store.get_standards_evidence(run_id, teacher_id=teacher_id) return { "run_id": run_id, @@ -240,7 +240,7 @@ async def standards_handout( ) obs_store = get_observability_store() - evidence = obs_store.get_standards_evidence(run_id) + evidence = obs_store.get_standards_evidence(run_id, teacher_id=teacher_id) quality_data: dict[str, Any] = {} for node in trace.nodes: diff --git a/runtime/ailine_runtime/api/routers/plans_stream.py b/runtime/ailine_runtime/api/routers/plans_stream.py index 5996052..3fa4bca 100644 --- a/runtime/ailine_runtime/api/routers/plans_stream.py +++ b/runtime/ailine_runtime/api/routers/plans_stream.py @@ -24,6 +24,8 @@ from pydantic import BaseModel, Field from sse_starlette.sse import EventSourceResponse +from ailine_agents.resilience import IdempotencyGuard + from ...app.authz import require_authenticated from ...shared.review_store import get_review_store from ...shared.sanitize import sanitize_prompt @@ -38,6 +40,9 @@ # Heartbeat interval to keep the connection alive (seconds). _HEARTBEAT_INTERVAL_S = 15.0 +# Idempotency guard -- prevent duplicate concurrent runs with the same run_id. +_idempotency_guard = IdempotencyGuard(ttl_seconds=300.0, max_size=1000) + class PlanStreamIn(BaseModel): """Request body for streaming plan generation.""" @@ -67,15 +72,18 @@ class PlanStreamIn(BaseModel): async def _heartbeat_loop( emitter: SSEEventEmitter, queue: asyncio.Queue[dict[str, str] | None], + done: asyncio.Event, interval: float = _HEARTBEAT_INTERVAL_S, ) -> None: """Push heartbeat events into the queue at a fixed interval. - Stops when ``None`` is placed in the queue (sentinel). + Stops when the ``done`` event is set or the task is cancelled. """ try: - while True: + while not done.is_set(): await asyncio.sleep(interval) + if done.is_set(): + return event = emitter.heartbeat() await queue.put({"data": event.to_sse_data()}) except asyncio.CancelledError: @@ -89,6 +97,8 @@ async def _run_pipeline( container: Any, emitter: SSEEventEmitter, queue: asyncio.Queue[dict[str, str] | None], + idem_key: str, + done: asyncio.Event, ) -> None: """Execute the LangGraph plan workflow, pushing SSE events to the queue.""" trace_store = None @@ -99,6 +109,7 @@ async def _run_pipeline( # Persist run metadata for the /runs resource model (F-237) await trace_store.update_run( body.run_id, + teacher_id=teacher_id, user_prompt=body.user_prompt[:500], subject=body.subject or "", ) @@ -177,6 +188,7 @@ def stream_writer(event: Any) -> None: await trace_store.update_run( body.run_id, + teacher_id=teacher_id, status="completed", final_score=final_payload.get("score"), scorecard=scorecard, @@ -200,9 +212,11 @@ def stream_writer(event: Any) -> None: # Mark trace as failed (guard against trace_store init failure) if trace_store is not None: - await trace_store.update_run(body.run_id, status="failed") + await trace_store.update_run(body.run_id, teacher_id=teacher_id, status="failed") finally: + done.set() + _idempotency_guard.complete(idem_key, None) # Sentinel to signal the generator to stop await queue.put(None) @@ -233,15 +247,21 @@ async def plans_generate_stream( # F-262: pass sanitised copy to pipeline; original body stays immutable safe_body = body.model_copy(update={"user_prompt": sanitized_prompt}) + # Idempotency guard -- prevent duplicate runs + idem_key = f"{teacher_id}:{safe_body.run_id}" + if not _idempotency_guard.try_acquire(idem_key): + raise HTTPException(status_code=409, detail="A run with this ID is already in progress") + emitter = SSEEventEmitter(safe_body.run_id) queue: asyncio.Queue[dict[str, str] | None] = asyncio.Queue(maxsize=500) + done = asyncio.Event() async def event_generator() -> AsyncIterator[dict[str, str]]: # Start the pipeline and heartbeat as background tasks pipeline_task = asyncio.create_task( - _run_pipeline(safe_body, teacher_id, settings, container, emitter, queue) + _run_pipeline(safe_body, teacher_id, settings, container, emitter, queue, idem_key, done) ) - heartbeat_task = asyncio.create_task(_heartbeat_loop(emitter, queue)) + heartbeat_task = asyncio.create_task(_heartbeat_loop(emitter, queue, done)) try: while True: diff --git a/runtime/ailine_runtime/api/routers/setup_service.py b/runtime/ailine_runtime/api/routers/setup_service.py index 6d9f2ef..6ebce8a 100644 --- a/runtime/ailine_runtime/api/routers/setup_service.py +++ b/runtime/ailine_runtime/api/routers/setup_service.py @@ -41,16 +41,10 @@ def _get_or_create_setup_token() -> str: _SETUP_TOKEN = env_token else: _SETUP_TOKEN = secrets.token_urlsafe(32) - import sys - print( - f"\n{'=' * 60}\n" - f" AILINE SETUP TOKEN: {_SETUP_TOKEN}\n" - f" Use this in the X-Setup-Token header for /setup/apply\n" - f"{'=' * 60}\n", - file=sys.stderr, - flush=True, + _log.info( + "setup_token_generated", + msg="Setup token generated. Retrieve via server admin or set AILINE_SETUP_TOKEN env var.", ) - _log.info("setup_token_generated", msg="Setup token printed to stderr") return _SETUP_TOKEN @@ -112,7 +106,7 @@ def require_setup_token(x_setup_token: str | None) -> None: "id": "gemini", "name": "Google Gemini", "models": [ - {"id": "gemini-embedding-001", "max_dims": 3072, "default_dims": 3072}, + {"id": "gemini-embedding-001", "max_dims": 3072, "default_dims": 1536}, ], }, { diff --git a/runtime/ailine_runtime/api/routers/tutors.py b/runtime/ailine_runtime/api/routers/tutors.py index eb66d34..52b660a 100644 --- a/runtime/ailine_runtime/api/routers/tutors.py +++ b/runtime/ailine_runtime/api/routers/tutors.py @@ -8,6 +8,7 @@ from typing import Any, Literal +import anyio from ailine_agents.deps import AgentDepsFactory from ailine_agents.workflows.tutor_workflow import build_tutor_workflow, run_tutor_turn from fastapi import APIRouter, Depends, HTTPException, Request @@ -61,8 +62,8 @@ async def tutors_create( @router.get("/{tutor_id}") -async def tutors_get(tutor_id: str): - spec = load_tutor_spec(tutor_id) +async def tutors_get(tutor_id: str, teacher_id: str = Depends(require_authenticated)): + spec = await anyio.to_thread.run_sync(load_tutor_spec, tutor_id) if not spec: raise HTTPException(status_code=404, detail="Tutor not found") @@ -77,16 +78,16 @@ class TutorSessionCreateOut(BaseModel): @router.post("/{tutor_id}/sessions") -async def tutor_create_session(tutor_id: str): - spec = load_tutor_spec(tutor_id) +async def tutor_create_session(tutor_id: str, teacher_id: str = Depends(require_authenticated)): + spec = await anyio.to_thread.run_sync(load_tutor_spec, tutor_id) if not spec: raise HTTPException(status_code=404, detail="Tutor not found") # Centralized tenant verification (ADR-060) require_tenant_access(spec.teacher_id, action="create", resource="tutor session") - s = create_session(tutor_id) - save_session(s) + s = await anyio.to_thread.run_sync(create_session, tutor_id) + await anyio.to_thread.run_sync(save_session, s) return TutorSessionCreateOut(session_id=s.session_id) @@ -106,14 +107,14 @@ async def tutor_chat(tutor_id: str, body: TutorChatIn, request: Request): status_code=422, detail="message must not be empty after sanitization" ) - spec = load_tutor_spec(tutor_id) + spec = await anyio.to_thread.run_sync(load_tutor_spec, tutor_id) if not spec: raise HTTPException(status_code=404, detail="Tutor not found") # Centralized tenant verification (ADR-060) require_tenant_access(spec.teacher_id, action="chat", resource="tutor") - session = load_session(body.session_id) + session = await anyio.to_thread.run_sync(load_session, body.session_id) if not session: raise HTTPException(status_code=404, detail="Session not found") if session.tutor_id != tutor_id: @@ -148,7 +149,7 @@ async def tutor_chat(tutor_id: str, body: TutorChatIn, request: Request): validated = result.get("validated_output") or {} answer = validated.get("answer_markdown", "") session.append("assistant", answer) - save_session(session) + await anyio.to_thread.run_sync(save_session, session) return { "validated": validated, @@ -165,12 +166,12 @@ async def tutor_chat(tutor_id: str, body: TutorChatIn, request: Request): @router.get("/{tutor_id}/sessions/{session_id}/transcript") async def tutor_session_transcript(tutor_id: str, session_id: str): """Get full conversation transcript for teacher review.""" - spec = load_tutor_spec(tutor_id) + spec = await anyio.to_thread.run_sync(load_tutor_spec, tutor_id) if not spec: raise HTTPException(status_code=404, detail="Tutor not found") require_tenant_access(spec.teacher_id, action="read", resource="tutor transcript") - session = load_session(session_id) + session = await anyio.to_thread.run_sync(load_session, session_id) if not session: raise HTTPException(status_code=404, detail="Session not found") if session.tutor_id != tutor_id: @@ -198,12 +199,12 @@ class TurnFlagIn(BaseModel): @router.post("/{tutor_id}/sessions/{session_id}/flag") async def tutor_flag_turn(tutor_id: str, session_id: str, body: TurnFlagIn): """Flag a specific turn in a tutor conversation for review.""" - spec = load_tutor_spec(tutor_id) + spec = await anyio.to_thread.run_sync(load_tutor_spec, tutor_id) if not spec: raise HTTPException(status_code=404, detail="Tutor not found") ctx = require_tenant_access(spec.teacher_id, action="flag", resource="tutor turn") - session = load_session(session_id) + session = await anyio.to_thread.run_sync(load_session, session_id) if not session: raise HTTPException(status_code=404, detail="Session not found") if session.tutor_id != tutor_id: diff --git a/runtime/ailine_runtime/shared/config.py b/runtime/ailine_runtime/shared/config.py index f0be929..e39687a 100644 --- a/runtime/ailine_runtime/shared/config.py +++ b/runtime/ailine_runtime/shared/config.py @@ -19,7 +19,7 @@ class EmbeddingConfig(BaseSettings): model_config = SettingsConfigDict(env_prefix="AILINE_EMBEDDING_") provider: Literal["gemini", "openai", "local", "openrouter"] = "gemini" model: str = "gemini-embedding-001" - dimensions: int = 3072 + dimensions: int = 1536 api_key: str = "" batch_size: int = 100 """Max embeddings per API call. Controls chunking of large embed requests @@ -63,7 +63,7 @@ class Settings(BaseSettings): "", validation_alias=AliasChoices("OPENAI_API_KEY", "AILINE_OPENAI_API_KEY") ) google_api_key: str = Field( - "", validation_alias=AliasChoices("GOOGLE_API_KEY", "AILINE_GOOGLE_API_KEY") + "", validation_alias=AliasChoices("GOOGLE_API_KEY", "AILINE_GOOGLE_API_KEY", "GEMINI_API_KEY") ) openrouter_api_key: str = Field( "", @@ -184,11 +184,12 @@ def validate_environment(self) -> list[str]: jwt_secret = os.getenv("AILINE_JWT_SECRET", "") jwt_public_key = os.getenv("AILINE_JWT_PUBLIC_KEY", "") - if not jwt_secret and not jwt_public_key: + jwt_private_key = os.getenv("AILINE_JWT_PRIVATE_KEY", "") + if not jwt_secret and not jwt_public_key and not jwt_private_key: errors.append( "JWT key material is required in production. " "Set AILINE_JWT_SECRET (HS256) or " - "AILINE_JWT_PUBLIC_KEY (RS256/ES256)." + "AILINE_JWT_PRIVATE_KEY + AILINE_JWT_PUBLIC_KEY (RS256/ES256)." ) if errors and is_prod: diff --git a/runtime/ailine_runtime/shared/observability_store.py b/runtime/ailine_runtime/shared/observability_store.py index 1c88c25..13dabd4 100644 --- a/runtime/ailine_runtime/shared/observability_store.py +++ b/runtime/ailine_runtime/shared/observability_store.py @@ -61,7 +61,7 @@ def __init__(self) -> None: "last_success": None, } self._circuit_breaker_state: str = "closed" - self._standards_evidence: dict[str, dict[str, Any]] = {} + self._standards_evidence: dict[tuple[str, str], dict[str, Any]] = {} self._cost_model: str = "default" # --- SSE event tracking --- @@ -163,27 +163,45 @@ def record_standards_evidence( standards: list[dict[str, Any]], bloom_level: str | None = None, alignment_explanation: str = "", + *, + teacher_id: str = "", ) -> None: - """Record standards alignment evidence for a run.""" + """Record standards alignment evidence for a run. + + When *teacher_id* is provided, the evidence is keyed by + ``(teacher_id, run_id)`` for tenant isolation. + """ + key = (teacher_id, run_id) if teacher_id else ("", run_id) with self._lock: - self._standards_evidence[run_id] = { + self._standards_evidence[key] = { "standards": standards, "bloom_level": bloom_level, "alignment_explanation": alignment_explanation, "recorded_at": time.time(), } - def get_standards_evidence(self, run_id: str) -> dict[str, Any]: - """Return standards evidence for a run, or empty defaults.""" + def get_standards_evidence( + self, run_id: str, *, teacher_id: str = "" + ) -> dict[str, Any]: + """Return standards evidence for a run, or empty defaults. + + When *teacher_id* is provided, looks up by ``(teacher_id, run_id)`` + for tenant isolation. Falls back to ``("", run_id)`` for + backward compatibility with evidence recorded without a tenant. + """ + key = (teacher_id, run_id) if teacher_id else ("", run_id) with self._lock: - return self._standards_evidence.get( - run_id, - { + result = self._standards_evidence.get(key) + if result is None and teacher_id: + # Fallback: check legacy key without teacher_id + result = self._standards_evidence.get(("", run_id)) + if result is None: + return { "standards": [], "bloom_level": None, "alignment_explanation": "Standards evidence not yet captured for this run.", - }, - ) + } + return result # Module-level singleton diff --git a/runtime/ailine_runtime/shared/trace_store.py b/runtime/ailine_runtime/shared/trace_store.py index 8df8753..d6487c2 100644 --- a/runtime/ailine_runtime/shared/trace_store.py +++ b/runtime/ailine_runtime/shared/trace_store.py @@ -40,19 +40,19 @@ def __init__( self._lock = asyncio.Lock() async def get( - self, run_id: str, *, teacher_id: str | None = None + self, run_id: str, *, teacher_id: str ) -> RunTrace | None: """Get a trace by run_id, or None if not found / expired. - When *teacher_id* is provided, only returns the trace if it - belongs to the given teacher (tenant isolation). + Only returns the trace if it belongs to the given teacher + (tenant isolation). *teacher_id* is required. """ async with self._lock: self._evict_expired() trace = self._traces.get(run_id) if trace is None: return None - if teacher_id is not None and trace.teacher_id != teacher_id: + if trace.teacher_id != teacher_id: return None return trace @@ -76,13 +76,18 @@ async def get_or_create(self, run_id: str, *, teacher_id: str = "") -> RunTrace: self._traces[run_id].teacher_id = teacher_id return self._traces[run_id] - async def append_node(self, run_id: str, node: NodeTrace) -> None: + async def append_node( + self, run_id: str, node: NodeTrace, *, teacher_id: str = "" + ) -> None: """Append a node trace to a run. F-252: Does NOT auto-create a RunTrace. If the run_id does not exist (i.e. was never initialised via ``get_or_create``), the call is silently ignored with a warning log. This prevents tenant-integrity bypass through implicit trace creation. + + When *teacher_id* is provided, validates it matches the stored + trace (tenant isolation). """ async with self._lock: if run_id not in self._traces: @@ -91,14 +96,28 @@ async def append_node(self, run_id: str, node: NodeTrace) -> None: run_id, ) return - self._traces[run_id].nodes.append(node) + trace = self._traces[run_id] + if teacher_id and trace.teacher_id and trace.teacher_id != teacher_id: + logger.warning( + "append_node tenant mismatch run_id=%s expected=%s got=%s — ignored", + run_id, + trace.teacher_id, + teacher_id, + ) + return + trace.nodes.append(node) self._timestamps[run_id] = time.monotonic() - async def update_run(self, run_id: str, **kwargs: Any) -> None: + async def update_run( + self, run_id: str, *, teacher_id: str = "", **kwargs: Any + ) -> None: """Update top-level run fields (status, total_time_ms, etc.). F-252: Does NOT auto-create a RunTrace. If the run_id does not exist, the call is silently ignored with a warning log. + + When *teacher_id* is provided, validates it matches the stored + trace (tenant isolation). """ async with self._lock: if run_id not in self._traces: @@ -108,6 +127,14 @@ async def update_run(self, run_id: str, **kwargs: Any) -> None: ) return trace = self._traces[run_id] + if teacher_id and trace.teacher_id and trace.teacher_id != teacher_id: + logger.warning( + "update_run tenant mismatch run_id=%s expected=%s got=%s — ignored", + run_id, + trace.teacher_id, + teacher_id, + ) + return for key, value in kwargs.items(): if hasattr(trace, key): setattr(trace, key, value) diff --git a/runtime/tests/test_redis_event_bus.py b/runtime/tests/test_redis_event_bus.py index 0eb2e8b..bc4fec7 100644 --- a/runtime/tests/test_redis_event_bus.py +++ b/runtime/tests/test_redis_event_bus.py @@ -48,7 +48,7 @@ async def test_publish_calls_local_handlers(self): async def handler(data: dict) -> None: received.append(data) - bus.subscribe("test.event", handler) + await bus.subscribe("test.event", handler) await bus.publish("test.event", {"key": "value"}) assert len(received) == 1 @@ -92,8 +92,8 @@ async def bad_handler(data: dict) -> None: async def good_handler(data: dict) -> None: calls.append("ok") - bus.subscribe("err", bad_handler) - bus.subscribe("err", good_handler) + await bus.subscribe("err", bad_handler) + await bus.subscribe("err", good_handler) await bus.publish("err", {}) assert calls == ["ok"]