MockBoard

Architecture Design Document

This document outlines the MVP architecture for a multi-tenant B2B SaaS platform that orchestrates real-time AI focus groups. It focuses on the managed concurrency of ECS Fargate tasks and the dual-layer vector store in pgvector, providing a clean, declarative boundary suitable for an AI coding agent practicing Compound Engineering.

TL;DR

This platform lets product teams upload a PRD or pitch deck and watch a panel of AI personas debate it in real time, then receive a structured analyst report when the session ends. The two most important architectural choices are using Amazon ECS on Fargate to run per-session LangGraph orchestration workers (persistent but fully managed), and a dual-layer vector store in pgvector-on-RDS to ground each persona in real ethnographic data while keeping product context sharply separated.

Executive Summary

The platform is a multi-tenant B2B SaaS system built entirely on AWS managed services. The backend is a Python/FastAPI service that handles tenant management, session configuration, document ingestion, and report retrieval. Session orchestration runs as ephemeral ECS Fargate tasks — one task per active focus group session — each hosting a LangGraph graph that coordinates a Moderator Agent, 1–10 Persona Agents, and an Analyst Agent. Real-time debate output streams from the orchestration worker directly to clients: the worker retrieves WebSocket connection IDs from Redis and pushes agent messages to those connections via the API Gateway Management API, with no intermediate relay component.

The dual-layer RAG architecture stores two logically separate namespaces in pgvector: a base ethnographic corpus (loaded at deploy time and supplemented by user-supplied grounding data) and per-session product context chunks derived from uploaded PRDs and decks. LangGraph state checkpoints are persisted directly into RDS PostgreSQL via the native AsyncPostgresSaver — no S3 involvement in the checkpoint path. S3 is reserved strictly for raw uploaded files and final compiled analyst reports.

State persistence uses a combination of RDS (relational session metadata, team/workspace data, message history, and LangGraph checkpoints) and Amazon S3 (raw files and final reports). The 99.9% availability target is met through RDS Multi-AZ, ECS task health checks with automatic restart, and the checkpoint-based resume pattern — if a worker crashes mid-session, the next task picks up from the last committed LangGraph checkpoint in RDS.

The primary architectural tension is managed concurrency: 500 simultaneous ECS Fargate tasks, each making parallel LLM calls for up to 10 persona agents, means the system is I/O-bound and cloud-spend-sensitive. AsyncPostgresSaver checkpoint writes at 500 concurrent sessions add meaningful write pressure to RDS; the instance is sized accordingly and write IOPS is a monitored metric. Cost guardrails are built into the design from the start.

System Overview

graph TD

subgraph CLIENT["Client Layer"]
    nextjs["Next.js Frontend
(Amplify)"]
end

subgraph APIGW["API Gateway Layer"]
    waf["AWS WAF"]
    restApi["REST API
(HTTP)"]
    wsApi["WebSocket API"]
end

subgraph AUTH["Auth"]
    cognito["Amazon Cognito
User Pools"]
end

subgraph BE["Backend Services"]
    alb["Application
Load Balancer"]
    fastapi["FastAPI
Control Plane
(ECS Fargate)"]
end

subgraph ORCH["Orchestration Workers"]
    worker["LangGraph Worker
(ECS Fargate
per session)"]
    moderator["Moderator Agent"]
    persona["Persona Agents
(1–10)"]
    analyst["Analyst Agent"]
end

subgraph INGEST["Ingestion Pipeline"]
    sqs["SQS
Ingestion Queue"]
    ingestTask["Ingestion Worker
(ECS Fargate)"]
end

subgraph DATA["Data Layer"]
    rds["RDS PostgreSQL 16
+ pgvector
+ AsyncPostgresSaver
(Multi-AZ)"]
    redis["ElastiCache Redis 7
(Connection ID Registry
+ Heartbeat)"]
    s3["Amazon S3
(uploads / reports)"]
end

subgraph EXTERNAL["External AWS Services"]
    bedrock["Amazon Bedrock
Claude claude-sonnet-4-5
+ Titan Embeddings v2"]
    mgmtApi["API Gateway
Management API"]
    ses["Amazon SES"]
    secretsMgr["Secrets Manager"]
    stepFn["Step Functions
(GDPR Erasure)"]
end

nextjs -->|"HTTPS REST"| waf
nextjs -->|"WebSocket connect"| wsApi
waf --> restApi
restApi -->|"JWT auth"| cognito
restApi --> alb
alb --> fastapi

wsApi -->|"store connection_id"| redis

fastapi -->|"RunTask"| worker
fastapi --> rds
fastapi --> s3
fastapi --> ses

worker --> moderator
moderator -->|"routes to"| persona
persona -->|"routes back"| moderator
moderator -->|"end session"| analyst

worker -->|"checkpoint write"| rds
worker -->|"write message + turn_sequence"| rds
worker -->|"read connection IDs"| redis
worker -->|"direct push"| mgmtApi
worker -->|"LLM calls"| bedrock
mgmtApi -->|"push messages"| nextjs

persona -->|"RAG: ethnographic
+ product context"| rds

ingestTask -->|"embed + upsert vectors"| rds
ingestTask -->|"parse files"| s3
ingestTask -->|"embed"| bedrock

s3 -->|"event notification"| sqs
sqs --> ingestTask

analyst -->|"write report"| s3

fastapi --> secretsMgr
worker --> secretsMgr

stepFn -->|"purge tenant data
+ checkpoints"| rds
stepFn -->|"delete S3 objects"| s3

Client layer: A Next.js application (React) served from AWS Amplify. Users configure and launch sessions, watch the live debate stream, and browse historical sessions and reports.

API layer: An Amazon API Gateway with two surfaces — a REST API for CRUD operations (sessions, workspaces, uploads, reports) and a WebSocket API for real-time streaming. Both route to the backend services layer.

Backend services: A containerized FastAPI application running on ECS Fargate behind an Application Load Balancer. Handles authentication (via Amazon Cognito), document ingestion pipeline, session lifecycle management, and report retrieval. This is the control plane.

Orchestration workers: Separate ECS Fargate tasks, one spawned per active session. Each task runs the LangGraph graph for the duration of the session. Workers initialize their graph state from the LangGraph checkpoint stored in RDS (via AsyncPostgresSaver) and write a new checkpoint at every node boundary. On each agent turn, the worker retrieves the active WebSocket connection IDs for the session from Redis and pushes agent messages directly to those connections via the API Gateway Management API.

Data plane: RDS PostgreSQL (Multi-AZ) with pgvector extension for vector similarity search and the LangGraph checkpoint tables for session state. Two pgvector namespaces — ethnographic for persona grounding and product_context for per-session PRD chunks. S3 holds raw uploaded files and rendered analyst reports only.

Streaming bus: ElastiCache Redis (cluster mode disabled, Multi-AZ with automatic failover). On $connect, API Gateway stores the WebSocket connection ID in Redis keyed by session_id. Orchestration workers read connection IDs from Redis and push messages directly to API Gateway Management API. Redis also carries the session heartbeat tracking.

Document ingestion: A separate ECS Fargate task (pulled from a dedicated task pool) handles PDF/deck parsing, chunking, embedding via Amazon Bedrock Titan Embeddings v2, and upsert into pgvector.

Data flow for a live session:

1. User uploads PRD → ingestion pipeline chunks, embeds, and stores vectors in the product_context namespace tagged to the session.
2. User configures personas and launches session → API layer spawns an ECS orchestration worker task.
3. Worker initializes LangGraph state from the RDS checkpoint (via AsyncPostgresSaver) or a fresh graph state for a new session.
4. Each agent turn: relevant ethnographic chunks are retrieved from the ethnographic namespace for the active persona; relevant PRD chunks are retrieved from product_context for the session. Both are injected into the agent prompt.
5. Agent output is written to the RDS message history table (with an incremented turn_sequence counter) and LangGraph writes a checkpoint to RDS via AsyncPostgresSaver.
6. The worker reads the session's connection IDs from Redis and pushes the agent message payload to each connected client via the API Gateway Management API.
7. On session completion, the Analyst Agent synthesizes the full message history into a structured report, stored in S3 and indexed in RDS.

Architectural Constraints Summary

AWS lock-in is a feature here, not a limitation. The preference for fully managed services drove the choice of ECS Fargate over EKS (no control plane to manage), RDS over a self-hosted vector DB, Cognito over Auth0/Clerk, and Amplify over a custom S3/CloudFront setup. Every service in the stack has a managed upgrade path and no server to patch.

The resumability requirement is the single hardest constraint. It means orchestration cannot be stateless — every LangGraph node transition must write a checkpoint before the next node begins. Lambda is ruled out for orchestration (15-minute limit, no persistent process). AsyncPostgresSaver writes checkpoints directly into the existing RDS PostgreSQL instance, eliminating the need for any separate checkpoint store. The constraint now drives RDS instance sizing rather than an S3 snapshot strategy.

The dual-layer RAG constraint means the vector store must support logical namespace separation. pgvector satisfies this with a table-per-namespace approach. A single managed RDS instance handles both namespaces and the LangGraph checkpoint tables, avoiding the operational overhead of a separate dedicated vector DB.

GDPR drives tenant data isolation (row-level security on all RDS tables keyed to tenant_id), right-to-erasure support (soft-delete with hard-purge jobs on S3 and RDS), data residency in a single AWS region (eu-west-1 or us-east-1 per tenant agreement), and no cross-tenant vector leakage (product context vectors are always scoped to session_id).

The concurrency target (50–500 sessions) resolved the scaling model: ECS Service Auto Scaling based on active session count, with a pre-warmed minimum task count to absorb burst. At 500 simultaneous tasks, AsyncPostgresSaver checkpoint writes become a meaningful RDS write workload — the instance is sized at db.r7g.large minimum with write IOPS monitored and alerting configured. The LLM API remains the binding constraint for session throughput; rate limit management is built into the Persona Agent layer.

The streaming architecture change (removing the Lambda relay) is driven by the desire to reduce components. The orchestration worker is already a persistent ECS task with network access — it can call the API Gateway Management API directly after reading connection IDs from Redis, making a dedicated relay Lambda redundant and eliminating one failure surface.

Component Architecture

Next.js Frontend (Amplify)

Responsible for all user-facing interactions. Uses React Server Components for static shell pages (workspace dashboard, session history) and client components for the live session view. The live session view opens a WebSocket connection to API Gateway on mount and appends incoming agent messages to a local state buffer rendered as a debate transcript. The session configuration wizard (persona count, archetype selection, topic framing) is a multi-step form that submits to the REST API. File upload uses pre-signed S3 URLs — the frontend requests a URL from the REST API, uploads directly to S3, then notifies the API of upload completion. No file bytes transit the API layer.

API Gateway (REST + WebSocket)

Two separate API Gateway instances. The REST API uses Lambda integrations for lightweight CRUD operations and proxies longer operations (session launch) to the FastAPI ALB. The WebSocket API manages connection lifecycle — $connect, $disconnect, and $default routes. On $connect, a Lambda stores the connection ID in Redis as a member of a set keyed by session_id, along with the API Gateway Management API endpoint URL needed for the push. On $disconnect, a Lambda removes the connection ID from that set. There is no long-running relay Lambda — the orchestration worker handles all outbound pushes directly. Connection IDs are stored in Redis with a TTL matching the maximum session duration.

FastAPI Backend (ECS Fargate — Control Plane)

Handles: tenant onboarding, workspace and team management, session CRUD (create, list, get, delete), persona archetype management, triggering document ingestion, spawning orchestration worker tasks via the ECS API, polling session state, and serving analyst reports. Runs as a persistent ECS Fargate service (not per-request Lambda) because it manages long-lived session state transitions and needs stable outbound connections to RDS and ElastiCache. Scales horizontally behind the ALB; target tracking on CPU utilization at 60%.

Authentication is verified via Cognito JWT on every request. Row-level security is enforced at the application layer — every database query is scoped to the tenant_id extracted from the JWT.

Document Ingestion Pipeline

Triggered by an S3 event notification when a raw file lands in the uploads bucket. An SQS queue decouples the event from processing. An ECS Fargate task (pulled from a dedicated task pool) consumes the SQS message, fetches the file from S3, parses it (PDFs via PyMuPDF, PowerPoint via python-pptx), chunks it into ~512-token segments with 10% overlap, generates embeddings via Bedrock Titan Embeddings v2, and upserts the vectors into the product_context namespace in pgvector, tagged with session_id and tenant_id. On completion, the task updates the session record in RDS with ingestion status and notifies the FastAPI service via an internal SQS completion queue.

LangGraph Orchestration Worker (ECS Fargate — per session)

The most complex component. Spawned by the FastAPI service via the ECS RunTask API call. Each task receives its session configuration (session ID, persona configs, topic, model parameters) via environment variables at task launch.

State persistence: On startup the worker opens an AsyncPostgresSaver connection to RDS and either resumes from the latest checkpoint for the given thread ID (if resuming an interrupted session) or initializes a fresh graph state. At every node boundary, LangGraph commits a checkpoint row to the RDS checkpoint tables before advancing to the next node. The session record in RDS no longer tracks a snapshot S3 key — the checkpoint is always fetched from RDS directly.

Streaming push: After each agent turn is written to the message history table, the worker reads the set of active connection IDs for the session from Redis, then calls the API Gateway Management API for each connection ID to push the message payload. Stale connection IDs (where the Management API returns a gone response) are removed from the Redis set immediately.

Graph topology: The LangGraph graph is cyclic. The Moderator Agent node runs first, selects the next persona to speak (or decides the round is complete), then transitions to the selected Persona Agent node. After each persona speaks, control returns to the Moderator. After a configurable number of rounds, the Moderator transitions to the Analyst Agent node, which produces the final report and transitions to a terminal node. A checkpoint is committed at the exit of every node.

Turn sequencing: The LangGraph graph state includes a global turn counter. The counter is incremented by the Moderator at the start of each agent turn and is passed into the RDS message history insert as the turn_sequence value. This guarantees absolute sequential ordering of messages independent of timestamp resolution or clock skew across retry scenarios.

Moderator Agent: Tracks which personas have spoken, how many rounds have elapsed, and whether the conversation has reached the configured topic depth. Uses a Claude claude-sonnet-4-5 call to generate a structured facilitation message and a routing decision (which persona speaks next, or end the round). The routing decision is the LangGraph conditional edge. The Moderator also injects the current debate context into each persona prompt to maintain coherence.

Persona Agent: Each persona is configured with an archetype (e.g., "skeptical enterprise buyer," "early adopter founder") and optional user-supplied grounding documents. On each turn, the agent queries the ethnographic pgvector namespace for the top-k most relevant chunks given the current debate topic and the persona's archetype embedding. It also queries the product_context namespace for the top-k most relevant PRD chunks given the Moderator's question. Both retrieval results are injected into the system prompt before the Claude claude-sonnet-4-5 call. The agent's output is the persona's spoken response, written to the RDS message history table (with the current turn_sequence value) and pushed to clients via the Management API.

Analyst Agent: Runs once at the end of the session. Receives the full message history from RDS ordered by turn_sequence (not from the graph state, to avoid token limit issues with large sessions). Uses Claude claude-sonnet-4-5 to generate a structured report covering: positioning findings, pricing sensitivity signals, feature prioritization, and dissenting persona views. The report is stored as a structured JSON document in S3 and a summary is written to RDS.

Rate limit management: The worker maintains a token bucket per LLM provider. Persona agents within a session are called sequentially (dictated by the Moderator's routing), not in parallel, so the per-session LLM call rate is bounded. Across 500 sessions this is still high — exponential backoff with jitter is applied on all LLM calls, and Bedrock throughput quotas are requested proactively.

Heartbeat: The worker publishes a heartbeat message to its Redis key every 30 seconds. The FastAPI service monitors heartbeats; if a worker misses three consecutive heartbeats, the session is marked suspended and a new ECS task is spawned to resume from the latest RDS checkpoint.

Ethnographic Corpus Loader

A one-time (and periodically re-run) batch ECS task that loads the base ethnographic corpus into the ethnographic pgvector namespace. Also handles ingestion of user-supplied grounding files (uploaded separately from PRDs). These vectors are tagged with tenant_id for user-supplied data and corpus: base for the shared ethnographic baseline. Persona agents query both during retrieval.

Data Architecture

Storage Systems

• RDS PostgreSQL 16 (Multi-AZ, db.r7g.large minimum) — relational metadata, message history, session state, team/workspace data, pgvector RAG namespaces, and LangGraph AsyncPostgresSaver checkpoint tables. Write IOPS is a monitored metric given checkpoint volume at 500 concurrent sessions; instance class should be upgraded to db.r7g.xlarge if sustained write IOPS exceed 3,000.
• Amazon S3 — raw uploaded files and final compiled analyst reports only. No checkpoint or snapshot data.
• ElastiCache Redis 7.x (cache.r7g.large, Multi-AZ with automatic failover) — WebSocket connection ID registry (sets keyed by session_id), session heartbeat tracking.

Core Entities

Entity	Attribute	Type + Constraints	Business Rule & Access Control
Tenant	id	UUID, PK	Root isolation unit; all other entities reference tenant_id
Tenant	gdpr_region	ENUM('eu', 'us'), NOT NULL	Determines which AWS region data is stored in; enforced at account creation
Tenant	erasure_requested_at	TIMESTAMPTZ, nullable	Triggers scheduled hard-purge job; access restricted to platform admins
Workspace	id	UUID, PK	Belongs to exactly one tenant; teams are workspace-scoped
Workspace	tenant_id	UUID, FK(Tenant), NOT NULL	Row-level security predicate; all queries filtered by this
User	id	UUID, PK (Cognito sub)	Cognito is source of truth for identity; local record is profile/preference store
User	tenant_id	UUID, FK(Tenant), NOT NULL	Must match the tenant of every workspace they belong to
WorkspaceMember	workspace_id + user_id	Composite PK	Junction table; role ENUM('owner', 'editor', 'viewer')
Session	id	UUID, PK	Core orchestration unit
Session	workspace_id	UUID, FK(Workspace), NOT NULL	Inherits tenant isolation from workspace
Session	status	ENUM('configuring', 'ingesting', 'running', 'paused', 'completed', 'failed'), NOT NULL	State machine enforced at the API layer; only valid transitions allowed
Session	current_round	INT, DEFAULT 0	Tracks progress; used by Moderator Agent for round-limit enforcement
Session	persona_count	INT, CHECK(1..10), NOT NULL	Enforced at session creation; drives worker task sizing
PersonaConfig	id	UUID, PK	Belongs to a session
PersonaConfig	archetype	TEXT, NOT NULL	Used as the base retrieval query against ethnographic namespace
PersonaConfig	grounding_doc_s3_key	TEXT, nullable	User-supplied grounding file; ingested into ethnographic namespace with tenant_id tag
Message	id	UUID, PK	Append-only; never updated
Message	session_id	UUID, FK(Session), NOT NULL	All messages for a session retrievable in order
Message	turn_sequence	INT, NOT NULL	Global turn counter incremented by the LangGraph state on each turn; absolute ordering guarantee independent of timestamp; set by the orchestration worker from graph state
Message	agent_role	ENUM('moderator', 'persona', 'analyst'), NOT NULL	Used for display rendering and analyst context assembly
Message	persona_config_id	UUID, FK(PersonaConfig), nullable	Populated for persona messages; null for moderator and analyst
Message	content	TEXT, NOT NULL	Raw agent output; GDPR erasure replaces content with '[erased]'
Message	created_at	TIMESTAMPTZ, DEFAULT NOW()	Secondary timestamp; used for debugging and audit; ordering uses turn_sequence
AnalystReport	id	UUID, PK	One per completed session
AnalystReport	session_id	UUID, FK(Session), UNIQUE	One-to-one with session
AnalystReport	s3_key	TEXT, NOT NULL	Full structured JSON report in S3
AnalystReport	summary	TEXT	Short text summary stored in RDS for listing views
EthnographicChunk	id	UUID, PK	Lives in pgvector table in ethnographic namespace
EthnographicChunk	tenant_id	UUID, nullable	NULL for base corpus chunks; set for user-supplied grounding
EthnographicChunk	embedding	vector(1536), NOT NULL	Bedrock Titan Embeddings v2 output
EthnographicChunk	source_doc_s3_key	TEXT	Traceability back to source; used for GDPR deletion
ProductContextChunk	id	UUID, PK	Lives in pgvector table in product_context namespace
ProductContextChunk	session_id	UUID, FK(Session), NOT NULL	Strictly session-scoped; never shared across sessions
ProductContextChunk	tenant_id	UUID, NOT NULL	Belt-and-suspenders isolation
ProductContextChunk	embedding	vector(1536), NOT NULL	Bedrock Titan Embeddings v2 output

LangGraph Checkpoint Tables

AsyncPostgresSaver creates and manages its own checkpoint tables in RDS (typically named checkpoints and checkpoint_writes). These tables are keyed by thread_id (mapped to session_id) and store the serialized LangGraph graph state at each node boundary. These tables are managed by the LangGraph library and should not be manually modified; however, they must be included in GDPR erasure (rows keyed to a session belonging to the erased tenant must be deleted) and in RDS backup coverage. At 500 concurrent sessions each writing a checkpoint per node transition, these tables are the primary driver of RDS write IOPS.

Data Lifecycle

Raw uploads: Stored in S3 under uploads/{tenant_id}/{session_id}/. Retained for the duration of the tenant account. GDPR erasure deletes the S3 object and all derived product context chunks.

LangGraph checkpoints: Written to RDS AsyncPostgresSaver tables, keyed by session thread ID. The latest checkpoint is always the most recent row for that thread — AsyncPostgresSaver manages its own versioning. No S3 involvement. Old checkpoints for completed sessions are eligible for pruning after 30 days via a scheduled cleanup job that deletes checkpoint rows for sessions in completed or failed status older than the retention window.

Message history: Appended to RDS for the life of the session, ordered by turn_sequence. Retained per tenant data retention policy (default 2 years). GDPR erasure replaces message content with [erased].

Analyst reports: S3 key stored in RDS; the JSON blob lives in S3. Retained alongside the session. GDPR erasure deletes the S3 object.

Product context vectors: Scoped to session. Purged from pgvector when the session is deleted or on GDPR erasure.

Ethnographic vectors (base corpus): Shared across all tenants. Not subject to GDPR erasure (no PII). Re-indexed on corpus updates.

Ethnographic vectors (user-supplied): Tagged with tenant_id. Purged on GDPR erasure.

Integration Architecture

Anthropic Claude via Amazon Bedrock

All LLM calls go through Amazon Bedrock's managed Claude endpoint, not the Anthropic API directly. This keeps API key management inside AWS IAM (no external secrets) and makes throughput quota management a native AWS operation. The Moderator, Persona, and Analyst agents all call Claude claude-sonnet-4-5 via Bedrock with streaming responses enabled — each token streams back to the worker, which assembles the full response and pushes the complete message to clients via the API Gateway Management API. Bedrock's managed retry and throttling behavior handles transient rate limits; the worker additionally applies exponential backoff with jitter for sustained overload. If Bedrock is unavailable for more than 60 seconds, the session is paused, the LangGraph state checkpoint is committed to RDS via AsyncPostgresSaver, and the user receives a "session paused — resumable" notification via the Management API push.

Amazon Bedrock Titan Embeddings v2

Used by both the document ingestion pipeline and the ethnographic corpus loader to produce 1536-dimension embeddings. The ingestion task calls Bedrock Titan synchronously per chunk; on Bedrock unavailability the SQS message is returned to the queue with a visibility timeout backoff (not deleted), and the ingestion task retries up to five times before moving the message to a dead-letter queue and marking the session as ingestion_failed.

Amazon Cognito

Cognito User Pools handle user registration, login, MFA, and token issuance. The Next.js frontend uses the Amplify Auth SDK to manage the token lifecycle (sign-in, refresh, sign-out). Every REST and WebSocket API call passes the Cognito ID token in the Authorization header; API Gateway validates the token against the Cognito JWKS endpoint before invoking any Lambda or backend route. User creation in Cognito triggers a Lambda post-confirmation hook that creates the local User record in RDS and assigns the user to a default workspace. If Cognito is unavailable, API Gateway rejects all requests with an authentication error — there is no fallback auth path.

Amazon S3

Two distinct bucket purposes: (1) raw uploads, accessed via pre-signed URL for upload (frontend → S3 direct) and via IAM role for read (ingestion tasks, orchestration workers reading persona grounding files); (2) reports, accessed by the FastAPI API for serving download URLs via pre-signed GET. S3 event notifications on the uploads bucket trigger the ingestion SQS queue. All buckets have versioning enabled, server-side encryption with AWS KMS, and Block Public Access enforced. A bucket lifecycle policy enforces region-based data residency for GDPR.

API Gateway Management API (WebSocket Push)

Orchestration workers call the API Gateway Management API directly to push messages to connected clients. The worker retrieves the set of active connection IDs for the session from Redis, then issues a push call per connection ID. If a push returns a "gone" response (the client has disconnected without a clean $disconnect event), the worker removes that connection ID from the Redis set and continues to the remaining connections. This integration is authenticated via the worker task's IAM role, which is granted permission to post to the Management API endpoint for the WebSocket API stage.

Amazon SES

Sends transactional emails: workspace invitation, session completion notification, GDPR erasure confirmation. The FastAPI service calls SES when these events occur. If SES is unavailable, the event is logged and queued in an SQS dead-letter queue for retry; email delivery failure is non-blocking — it does not fail the session or the GDPR erasure.

ECS RunTask API (internal)

The FastAPI control plane calls the ECS RunTask API to spawn orchestration worker tasks. This is an internal AWS API call, authenticated via the FastAPI service's ECS task IAM role. If the RunTask call fails (ECS capacity issue), the FastAPI service retries three times with exponential backoff, then marks the session as failed and notifies the user. No partial state is created — the session remains in configuring status until a task is successfully started.

Security Architecture

Authentication and Authorization

Cognito User Pools with email+password and optional TOTP MFA. JWT tokens (ID token) are validated at API Gateway before any backend logic runs. The FastAPI backend extracts tenant_id and user_id from the validated token and applies these as query predicates on every database operation. No query is issued without a tenant scope. Workspace-level authorization (owner/editor/viewer) is checked in the FastAPI service before session mutations.

Tenant Data Isolation

Row-level filtering is enforced at the application layer (all queries include a tenant scope predicate). RDS row-level security policies provide a defense-in-depth backstop — a compromised application-layer query still cannot return cross-tenant data. Product context vectors in pgvector include both tenant_id and session_id on every row; ANN queries always filter by both. Ethnographic vectors are either base corpus (no PII, shared) or tenant-scoped (filtered by tenant_id). LangGraph checkpoint rows are keyed by thread_id (session ID), and GDPR erasure must also purge checkpoint rows for the erased tenant's sessions.

Secrets Management

All secrets (RDS credentials, Redis auth token, internal service keys) are stored in AWS Secrets Manager. ECS tasks retrieve secrets at task startup via the native ECS Secrets integration — secrets are injected as environment variables from Secrets Manager, never hardcoded or stored in ECR images. Secret rotation is enabled for RDS credentials (automatic 30-day rotation). No secrets appear in CloudWatch logs — the logging configuration masks environment variables.

Input Validation

File uploads: validated for MIME type (PDF, PPTX, PPTM, KEY accepted) and maximum file size (50 MB default) before pre-signed URL generation. The ingestion pipeline additionally validates parsed file structure before chunking. Session configuration: persona count is validated against the 1–10 range at API layer. All user-provided text fields (persona archetype names, topic descriptions) are treated as untrusted input — passed through to LLM prompts only after sanitization that strips control characters and enforces maximum lengths. LLM prompt injection is mitigated by placing user-supplied content in clearly delimited sections of the prompt, never in the instruction segment.

Network Security

All ECS tasks run in private subnets. RDS and ElastiCache are in isolated private subnets with no internet-facing routes. Traffic from ECS tasks to S3 and Bedrock travels via VPC endpoints (Gateway endpoint for S3, Interface endpoints for Bedrock and Secrets Manager) — no traffic leaves the AWS network. Orchestration workers access the API Gateway Management API over the public internet endpoint (as is standard for API Gateway) using HTTPS with IAM-signed requests. API Gateway and the ALB are the only internet-facing endpoints. Security groups are least-privilege: the FastAPI service SG allows inbound only from the ALB; worker tasks allow no inbound traffic; RDS SG allows inbound only from the FastAPI and worker SGs; Redis SG allows inbound only from the FastAPI and worker SGs.

Web-Facing Hardening

API Gateway enforces HTTPS-only (HTTP requests are rejected at the gateway). CORS policy on the REST API allows only the Amplify-hosted frontend origin (no wildcard). Response headers include: Strict-Transport-Security (max-age 31536000, includeSubDomains), X-Content-Type-Options: nosniff, X-Frame-Options: DENY, Content-Security-Policy restricting script sources to the application origin and CDN. AWS WAF is attached to API Gateway with rules for rate limiting (per IP, per user), SQL injection detection, and oversized request body rejection.

GDPR Compliance

Data residency: tenant data stored in the region specified at account creation (eu-west-1 for EU tenants). Right to erasure: a GDPR erasure request triggers a workflow that soft-deletes the user record, replaces all message content with [erased], deletes all S3 objects under the user's tenant prefix, purges tenant-scoped pgvector rows, and deletes AsyncPostgresSaver checkpoint rows for all sessions belonging to the erased tenant. The workflow is implemented as a Step Functions state machine to ensure all steps complete (with retry) even if partial failures occur. A completion audit log entry is written to a separate, append-only RDS table. Data minimization: only data required for platform function is collected; no third-party analytics SDKs with cross-site tracking.

Error Handling Policy

Errors returned to clients are always generic (no stack traces, no internal identifiers, no database error messages). Full error details are logged to CloudWatch Logs with a correlation ID. The correlation ID is returned to the client so support can trace it.

Deployment Architecture

AWS Region and Environment Strategy

Two environments: production and staging. Both run in the same AWS account in the same region (us-east-1 as default; can be eu-west-1 for GDPR primary tenants). Resources are separated by name prefix (prod- / staging-). A separate AWS account for staging is the eventual target but out of scope for the initial build.

Infrastructure as Code

All AWS resources defined in AWS CDK (TypeScript). CDK stacks are organized by concern: VPC/networking, data (RDS, ElastiCache, S3), compute (ECS cluster, task definitions), and application (API Gateway, Cognito, Amplify). CDK deployments are idempotent and run via GitHub Actions.

CI/CD Pipeline

GitHub Actions workflows:

• On PR: lint, type-check, unit tests, CDK diff (shows infra changes without deploying).
• On merge to main: build Docker images, push to ECR, CDK deploy to staging, run integration test suite, require manual approval gate, then CDK deploy to production.

Docker images are built for two targets: the FastAPI control plane service and the LangGraph orchestration worker. Images are tagged with the Git commit SHA. ECS services are updated with a rolling deployment (minimum 100% healthy, maximum 200%) so there is no downtime during deploy. Orchestration worker tasks are not updated mid-session — in-flight sessions continue on the old image; new sessions spawn tasks from the new image.

ECS Fargate Configuration

FastAPI control plane: ECS Service with Application Auto Scaling. Min 2 tasks, max 20 tasks. Target tracking on ALB request count per target (target: 1000 requests per minute per task). Task size: 1 vCPU, 2 GB RAM.

Orchestration workers: Launched via RunTask (not a persistent service). Each task: 2 vCPU, 4 GB RAM. No auto-scaling policy — one task per session, bounded by the 500-session concurrency limit enforced at the API layer. ECS capacity provider: Fargate Spot for cost reduction (with Fargate on-demand as fallback — if the Fargate Spot task is interrupted, the heartbeat mechanism detects the missed heartbeat and resumes from the RDS checkpoint on a new on-demand task).

Ingestion workers: ECS Service consuming from SQS. Min 0 tasks, max 10 tasks (scales with SQS queue depth).

RDS Configuration

PostgreSQL 16, Multi-AZ, db.r7g.large (upgrade path to db.r7g.xlarge if write IOPS from AsyncPostgresSaver checkpointing at 500 concurrent sessions sustains above 3,000 IOPS). Storage: 500 GB gp3, auto-scaling enabled up to 2 TB. Automated backups retained 7 days, PITR enabled. Parameter group tuned for pgvector workloads (increased shared_buffers, work_mem). Write IOPS CloudWatch alarm threshold: 3,000 IOPS sustained for more than 5 minutes triggers an alert to notify the team of the need to evaluate an instance class upgrade.

Monitoring and Observability

CloudWatch Container Insights: ECS task-level CPU, memory, and network metrics. Alarms on: FastAPI task count < 2 (service degradation), orchestration worker task failure rate > 5% (LLM or checkpoint issues), SQS dead-letter queue depth > 0 (ingestion failures).

CloudWatch Logs: Structured JSON logs from all services. Log groups per service with 90-day retention. Correlation IDs on all log entries.

CloudWatch Alarms → SNS → PagerDuty (or email): P1 alarms for RDS Multi-AZ failover, ElastiCache node failure, API Gateway 5xx rate > 1%, RDS write IOPS > 3,000 sustained.

AWS X-Ray: Distributed tracing enabled on FastAPI and orchestration workers. Trace IDs propagate through all LLM calls and database queries.

Scaling at 500 Sessions

At 500 concurrent sessions: 500 Fargate tasks × 2 vCPU = 1,000 vCPU demand. Fargate Spot pools are requested with a Fargate on-demand fallback. A concurrency cap of 500 active sessions is enforced at the API layer — session launch returns a capacity_unavailable error above this limit rather than silently degrading. The Bedrock service quota for Claude claude-sonnet-4-5 tokens per minute is the binding constraint and must be requested via AWS quota increase before launch. RDS write capacity for AsyncPostgresSaver checkpoint writes is the second binding constraint and must be verified under load in staging.

Technology Stack

Technology	Version / Model	Role	Why chosen
Next.js	14 (App Router)	Frontend framework	Constraints specify React/Next.js; App Router enables RSC for static shell, client components for live stream
React	18	UI library	Specified in constraints
AWS Amplify (Hosting)	Gen 2	Frontend hosting, Cognito integration	Fully managed, zero-ops, native Cognito SDK support
FastAPI	0.111	REST API / control plane	Fast Python async framework; compatible with LangGraph Python ecosystem
Python	3.12	Backend runtime	LangGraph and LangChain are Python-native; best ecosystem fit
LangGraph	0.2.x	Multi-agent orchestration	Specified in constraints; cyclic graph support and native AsyncPostgresSaver are key capabilities
LangGraph AsyncPostgresSaver	0.2.x (bundled)	LangGraph checkpoint persistence	Native LangGraph persistence backend; writes checkpoints directly to RDS PostgreSQL; eliminates need for separate snapshot store
Claude claude-sonnet-4-5 (via Bedrock)	claude-sonnet-4-5	All LLM agent calls	Strong instruction-following for structured agent roles; available via Bedrock (no external API key)
Amazon Bedrock Titan Embeddings v2	v2	Embedding generation	AWS-native, no external API key, 1536-dim vectors, managed throughput
Amazon ECS Fargate	—	Control plane + worker + ingestion container hosting	Fully managed, no EC2 to patch; Fargate Spot for cost on workers
Amazon RDS PostgreSQL + pgvector	PostgreSQL 16, pgvector 0.7	Relational data + vector store (dual-layer RAG) + LangGraph checkpoints	Single managed service covers relational, vector, and checkpoint storage needs; eliminates Pinecone and S3 snapshot operational overhead
Amazon ElastiCache Redis	Redis 7.x	WebSocket connection ID registry + session heartbeat tracking	Managed Redis with Multi-AZ failover; lightweight fast reads for connection ID lookup by workers
Amazon API Gateway	v2 (HTTP + WebSocket)	REST API surface + WebSocket connection management + Management API for push	Fully managed; WebSocket API native support; Management API enables worker-direct push
Amazon S3	—	File storage (raw uploads and analyst reports only)	Durable, cheap, native ECS/Lambda access via IAM
Amazon Cognito	—	Authentication, JWT issuance, MFA	AWS-native auth; API Gateway native integration; no external auth vendor
Amazon SQS	—	Ingestion job queue, internal async events	Managed queue; decouples ingestion from upload event
Amazon Step Functions	Standard Workflows	GDPR erasure orchestration	Durable, managed workflow for multi-step erasure with retry
AWS Secrets Manager	—	All secrets storage and rotation	Native ECS integration; automatic rotation
AWS CDK	2.x (TypeScript)	Infrastructure as Code	Type-safe; AWS-native; avoids Terraform external dependency
AWS WAF	v2	API Gateway protection	Managed rules for rate limiting and injection detection
AWS X-Ray	—	Distributed tracing	Native ECS + API Gateway integration; no agent to manage
GitHub Actions	—	CI/CD	No-ops, widely used, integrates with CDK
PyMuPDF	1.24.x	PDF parsing in ingestion pipeline	Fast, pure-Python, no external service dependency
python-pptx	0.6.x	PowerPoint parsing	Standard Python library for PPTX extraction
Amazon SES	v2	Transactional email	AWS-native; no external email vendor

Key Architectural Decisions

ADR-1: ECS Fargate over AWS Lambda for orchestration workers Decided to run LangGraph workers as long-running ECS Fargate tasks, not Lambda functions. Alternatives: Lambda (serverless, zero-ops), EKS (more control). Lambda was rejected because of the 15-minute execution limit — focus group sessions are expected to run 30–90 minutes, and LangGraph requires a persistent process to hold in-memory graph state between node evaluations. EKS was rejected because it requires managing the control plane and node groups, conflicting with the small-team, low-ops constraint. Fargate gives container isolation, per-task billing, and no server management.

ADR-2: pgvector on RDS over a dedicated vector database (Pinecone / Qdrant) Decided to use pgvector in the existing RDS PostgreSQL instance for both RAG namespaces instead of a separate vector database. Alternative: Pinecone (managed, purpose-built). Pinecone adds a third managed service, an external API key, and cross-service latency. pgvector on RDS consolidates relational and vector queries into one managed instance, simplifies GDPR erasure (a single DELETE statement removes both relational and vector data for a session), and is sufficient for 500-session concurrency with HNSW indexing on the vector columns.

ADR-3: Worker-direct push to API Gateway Management API over a dedicated relay Lambda Decided that orchestration workers push messages directly to connected WebSocket clients via the API Gateway Management API, after reading connection IDs from Redis. The previous design used a long-running Lambda relay subscriber listening to Redis pub/sub. That relay Lambda was removed because the orchestration worker is already a persistent, network-capable ECS task that can call the Management API directly — the relay was a redundant intermediary with its own failure surface. Redis is now used only as a connection ID registry (sets keyed by session_id) and a heartbeat store, not as a pub/sub message bus.

ADR-4: Bedrock for all LLM calls over direct Anthropic API Decided to route all Claude calls through Amazon Bedrock rather than calling the Anthropic API directly. Alternatives: Anthropic API, Azure OpenAI. Bedrock is AWS-native — IAM handles authentication (no external API key to rotate or leak), throughput quotas are manageable via AWS Service Quotas, and billing is consolidated with the rest of the AWS infrastructure. This also keeps all traffic inside the AWS network via VPC endpoints, satisfying the GDPR data residency requirement without additional configuration.

ADR-5: One ECS task per session for orchestration workers Decided that each active session spawns exactly one dedicated ECS Fargate task, rather than a shared worker pool multiplexing sessions. Alternative: a worker pool where tasks handle multiple sessions. A dedicated task per session means the LangGraph graph state is fully in-memory within the task (no cross-session state sharing bugs), crashes are isolated to a single session, and resumability is clean (kill the task, checkpoint is in RDS, start a new task). The cost tradeoff is accepted: Fargate Spot pricing and the 500-session cap keep the bill predictable.

ADR-6: AsyncPostgresSaver in RDS for LangGraph checkpoints over S3 JSON blobs Decided to use LangGraph's native AsyncPostgresSaver to write state checkpoints directly into RDS PostgreSQL, replacing the prior design that serialized snapshots as JSON blobs to S3. Alternatives: S3 JSON blobs (previous approach), Redis. AsyncPostgresSaver is the idiomatic LangGraph persistence mechanism and is designed for exactly this use case — it creates and manages its own checkpoint tables in PostgreSQL, versioned by thread ID. This eliminates the operational complexity of managing S3 lifecycle policies for checkpoint objects, removes the need to track a latest_snapshot_s3_key field on the Session table, and consolidates all persistent state into the existing managed RDS instance. The tradeoff is that checkpoint writes add to RDS write IOPS at high concurrency; this is accepted and mitigated by sizing the RDS instance at db.r7g.large minimum with IOPS monitoring and a clear upgrade path to db.r7g.xlarge. S3 is now strictly reserved for raw document uploads and final compiled analyst reports.

ADR-7: turn_sequence integer for absolute message ordering Decided to add an integer turn_sequence column to the Message table, populated from a global turn counter maintained in the LangGraph graph state. The Moderator Agent increments the counter at the start of each agent turn and passes it into the database insert. Alternative: rely on created_at timestamp for ordering. Timestamps are insufficient because clock skew, retry scenarios, and sub-millisecond writes from the same worker can produce ambiguous ordering. The turn_sequence value is authoritative for message ordering, transcript replay, and analyst report construction.

Error Handling & Failure Modes

Document Ingestion Failures

When a user uploads a file that cannot be parsed (corrupted PDF, unsupported format variant), the ingestion task catches the parsing error, moves the SQS message to the dead-letter queue after five retries, and marks the session status as ingestion_failed in RDS. The FastAPI service detects this status and sends a notification to the session owner via SES email. The raw file is retained in S3 for debugging. The user is prompted to re-upload.

What happens if Bedrock Titan Embeddings is unavailable during ingestion? The ingestion task catches the Bedrock service error and returns the SQS message to the queue with a backoff visibility timeout (30s, 60s, 120s, 300s, 600s). After five failed attempts, the message goes to the dead-letter queue, the session is marked ingestion_failed, and the user is notified. The SQS message retention period is set to 24 hours, giving the Bedrock service time to recover for non-DLQ retries.

Orchestration Worker Failures

What happens if an orchestration worker ECS task crashes mid-session? The task stops publishing heartbeats to Redis. The FastAPI control plane's heartbeat monitor (running on a 30-second polling loop) detects three consecutive missed heartbeats (90 seconds of silence). It marks the session as suspended, then calls ECS RunTask to launch a new worker task with the same session configuration. The new task opens an AsyncPostgresSaver connection to RDS and fetches the latest checkpoint for the session's thread ID, resuming the LangGraph graph from the last committed node. Any messages written to RDS before the crash are already persisted with their turn_sequence values intact; the resumed session continues the counter from where it left off. Connected clients receive a "session suspended — resuming" notification via the Management API push from the new worker once it starts. The maximum data loss is one in-progress agent turn — the turn that was executing when the crash occurred is not written to message history, and the graph resumes from the prior committed checkpoint.

What happens if an AsyncPostgresSaver checkpoint write fails at a node boundary? The worker retries the checkpoint write three times with exponential backoff. If all retries fail, the worker pauses the session (stops advancing the graph), pushes a "session paused" message to clients via the Management API, and marks the session paused in RDS. The session remains resumable from the last successful checkpoint. The user is notified.

LLM Call Failures (Bedrock Claude)

Transient throttling (rate limit) from Bedrock: the worker retries with exponential backoff and jitter (starting at 2 seconds, up to 60 seconds, maximum 5 retries). If retries are exhausted, the agent turn fails, the session is paused, the current LangGraph state is checkpointed to RDS via AsyncPostgresSaver, and the user is notified via Management API push that the session is paused and can be resumed. The incomplete agent turn is not written to message history.

If Bedrock Claude is unavailable for more than 60 seconds: same pause-and-checkpoint behavior as above. The session is resumable when Bedrock recovers.

WebSocket Connection Failures

If a client's WebSocket connection drops during a live session, the client reconnects (the frontend retries with exponential backoff up to 5 attempts). On reconnect, the client fetches the full message history from the REST API ordered by turn_sequence to rebuild the transcript, then re-subscribes to the WebSocket stream. On $connect, the new connection ID is stored in Redis under the session's key set. The session continues unaffected — the worker pushes to whatever connection IDs are in Redis at the time of each publish.

If a worker's push to a connection ID returns a "gone" response from the Management API (client disconnected without a clean $disconnect event): the worker removes that connection ID from the Redis set immediately and continues pushing to the remaining connection IDs. No session state is affected.

If the Redis connection ID store is unavailable (ElastiCache node failure): ElastiCache Multi-AZ with automatic failover means the Redis primary fails over to the replica within 30–60 seconds. During failover, worker push attempts that cannot read connection IDs will silently skip the push (no connection IDs retrieved). Messages are still written to RDS with correct turn_sequence values. Clients are unaffected in-session — they receive no messages during the failover window, then the worker resumes pushing when Redis recovers. Clients can fetch any missed messages from the REST API by turn_sequence offset.

API Layer Failures

If the FastAPI control plane ECS service has no healthy tasks: the ALB returns a 503 to API Gateway, which returns a generic error to the client. In-flight orchestration worker tasks are unaffected — they do not depend on the control plane to continue running. The ALB health check is tuned to mark tasks healthy quickly after restart (10-second interval, 2 consecutive successes).

If Cognito is unavailable: API Gateway cannot validate JWTs and rejects all requests. There is no fallback authentication path — this is an accepted availability tradeoff given Cognito's own high availability SLA.

RDS Write Pressure from AsyncPostgresSaver

If RDS write IOPS sustains above the 3,000 IOPS alarm threshold due to checkpoint volume at high concurrency: CloudWatch triggers the alert, and the team evaluates an instance class upgrade to db.r7g.xlarge. In the interim, checkpoint write latency may increase, which adds latency to each node transition. Sessions remain functional — checkpoint writes are synchronous before node advancement, so high latency means slower sessions rather than data loss or session failure.

GDPR Erasure Failures

The erasure Step Functions state machine retries each step (message content update, S3 deletion, pgvector purge, AsyncPostgresSaver checkpoint row deletion, user record soft-delete) independently with exponential backoff. If a step fails after all retries, the state machine moves to a failure state and publishes an alert to SNS. An on-call engineer is notified to manually complete the failed step. A completion audit log is written only after all steps succeed. Partial erasure does not generate a confirmation to the user.

Input Validation Failures

File size exceeds 50 MB: the pre-signed URL generation call to FastAPI validates the declared file size from the client request and rejects it before any S3 interaction. The client receives a clear error message.

Invalid session configuration (e.g., persona count outside 1–10): rejected at the FastAPI validation layer with a descriptive error; no database write occurs.

Pre-Launch Verification

• Bedrock throttling behavior: verified by running a load test against staging with a reduced Bedrock quota, confirming the backoff and pause behavior.
• ECS Fargate Spot interruption and resume: verified by manually stopping a worker task in staging and confirming the heartbeat monitor detects it, spawns a replacement, and the replacement successfully resumes from the RDS AsyncPostgresSaver checkpoint.
• AsyncPostgresSaver checkpoint write throughput: verified by running a 500-session concurrent load test in staging and confirming RDS write IOPS remain below 3,000 at db.r7g.large; if they exceed the threshold, the test itself validates the upgrade trigger.
• GDPR erasure workflow: verified in staging by triggering the Step Functions state machine against a test tenant and confirming all data is purged — including AsyncPostgresSaver checkpoint rows — with an audit log entry.
• Redis failover and connection ID recovery: verified by triggering an ElastiCache primary failover in staging and confirming sessions resume push delivery after the 30–60 second failover window, with clients able to fetch missed messages by turn_sequence.
• Worker-direct Management API push: verified by confirming a running worker task can successfully push messages to a connected WebSocket client without any relay Lambda, including stale connection ID cleanup behavior.
• Ingestion DLQ behavior: verified by uploading a deliberately corrupted PDF and confirming the session reaches ingestion_failed status and the user is notified.

Risks and Mitigations

Risk 1: Bedrock Claude throughput quota is the hard ceiling for session concurrency At 500 sessions with 1–10 sequential LLM calls per session active simultaneously, the aggregate token throughput can exceed default Bedrock service quotas. If throttling becomes severe enough that backoff delays push sessions beyond user tolerance, sessions will feel "stuck." Mitigation: request a Bedrock quota increase before the production launch via AWS Service Quotas. Instrument per-session LLM latency in CloudWatch and alert if median turn latency exceeds 30 seconds. Implement a session queue (visible to the user) when concurrency approaches the quota ceiling rather than degrading silently.

Risk 2: RDS write IOPS saturation from AsyncPostgresSaver checkpoint volume At 500 concurrent sessions each writing a checkpoint on every node boundary, AsyncPostgresSaver generates a sustained high-throughput write workload on the primary RDS instance. If write IOPS saturates the instance class, checkpoint write latency increases, which directly slows node transitions across all active sessions simultaneously — a single-point degradation affecting all users. Mitigation: size the instance at db.r7g.large with gp3 storage and a 3,000 IOPS CloudWatch alarm. Validate checkpoint throughput under 500-session load in staging before production launch. Establish a clear runbook for upgrading to db.r7g.xlarge if the alarm fires. As a longer-term option, evaluate AsyncPostgresSaver's ability to write checkpoints to a read replica — if supported, this offloads checkpoint reads to the replica and reduces primary write amplification.

Risk 3: LangGraph state object grows unbounded for long sessions A session with many rounds and 10 personas accumulates context in the LangGraph state object. If the full message history is kept in the graph state, the AsyncPostgresSaver checkpoint payload grows with session length, eventually causing slow checkpoint writes or serialization failures. Mitigation: the LangGraph state is designed to hold only the current round's working context — not the full message history. Full message history is stored in RDS and retrieved by the Analyst Agent at the end of the session ordered by turn_sequence. The graph state checkpoint remains bounded in size regardless of session length.

Risk 4: pgvector query performance degrades at scale with large ethnographic corpus As the ethnographic corpus grows (both base corpus and tenant-supplied grounding), ANN query latency for persona grounding retrieval may increase. At 500 concurrent sessions each querying pgvector per agent turn, this could become a bottleneck on the RDS instance. Mitigation: HNSW indexes are created on both vector columns at table creation time. The RDS instance class provides sufficient RAM to keep the HNSW index in memory for typical corpus sizes. If the corpus grows beyond what a single instance can serve, read replicas can be added to distribute the ANN query load — pgvector queries during sessions are read-only.

Risk 5: Multi-tenant data isolation bug in pgvector queries A coding error that omits the tenant_id or session_id filter from a pgvector ANN query could expose one tenant's product context or grounding data to another tenant's persona agent. At a B2B SaaS level, this would be a critical data breach. Mitigation: RDS row-level security policies enforcing tenant_id filtering are the backstop even when the application filter is missing. Every pgvector query is integration-tested with cross-tenant data present in the test database to confirm isolation. An automated test in the CI pipeline specifically checks that a query using one tenant's credentials cannot retrieve another tenant's rows from either vector table.

Implementation Plan

This document outlines the MVP implementation plan for a multi-tenant B2B SaaS platform that orchestrates real-time AI focus groups. It focuses on the managed concurrency of ECS Fargate tasks and the dual-layer vector store in pgvector, providing a clean, declarative boundary suitable for an AI coding agent practicing Compound Engineering.

Environment Setup

What you'll need

Before anyone writes a single line of code, you'll need a handful of accounts and a few configuration steps done inside them. Getting these right upfront means the coding agent can connect to real services from day one instead of hitting walls mid-build.

Accounts to create

• AWS — aws.amazon.com — Everything runs here: your database, containers, file storage, authentication, AI models, and the API itself. This is the one account that powers the whole platform.
• GitHub — github.com — Stores your code and runs the automated build and deploy pipeline every time changes are merged.
• Anthropic (via AWS Bedrock) — No separate Anthropic account needed. Claude is accessed through AWS Bedrock, which you enable inside your AWS account (see below). All billing goes through AWS.

Before you start

Inside your AWS account:

1. Enable Amazon Bedrock in your chosen region (us-east-1 or eu-west-1). Inside Bedrock, go to Model Access and request access for Claude claude-sonnet-4-5 and Titan Embeddings v2. Approval is usually instant but can take a few minutes.

2. Request a Bedrock service quota increase for Claude claude-sonnet-4-5 tokens per minute before going to production. File this early — it can take a day or two to process.

3. Create a Cognito User Pool. Inside it, enable email+password sign-in and TOTP MFA (optional for users but the pool should support it). Note the User Pool ID and the App Client ID — you'll need both.

4. Create two S3 buckets in your chosen region: one for raw file uploads, one for analyst reports. On both, turn on versioning, enable server-side encryption with AWS KMS, and make sure Block Public Access is fully enabled. Set a bucket lifecycle policy that keeps data in the region you chose.

5. In Secrets Manager, create secrets for your RDS master password and Redis auth token before the database is created — CDK will reference these by ARN during the first deploy.

6. In your GitHub repository settings, go to Environments and create two environments: staging and production. Add a required manual approval reviewer on the production environment (this is the gate between staging and prod deploys).

7. Create an IAM role for GitHub Actions (using OIDC federation — no long-lived keys). Grant it permissions to push to ECR, deploy CDK stacks, and update ECS services.

8. Verify a sending domain or email address in Amazon SES. If your account is still in SES sandbox mode, request production access so you can send to any address.

9. In API Gateway, after the WebSocket API is created by CDK, note the WebSocket stage URL and the Management API endpoint — the orchestration worker needs these at runtime.

For local CDK bootstrapping:

The coding agent will need the AWS CDK bootstrapped in your account and region once (run cdk bootstrap with appropriate credentials). This only needs to happen once per account/region pair.

Environment variables

The coding agent will need these available in the GitHub Actions environment secrets and as ECS task environment variable references:

GitHub Actions secrets (CI/CD pipeline):

• AWS_ACCOUNT_ID — Your 12-digit AWS account number
• AWS_REGION — The region you chose (e.g., us-east-1)
• CDK_DEFAULT_ACCOUNT — Same as AWS_ACCOUNT_ID
• CDK_DEFAULT_REGION — Same as AWS_REGION

FastAPI control plane task (injected via ECS Secrets integration from Secrets Manager):

• DATABASE_URL — Secrets Manager → your RDS secret → assemble as postgresql+asyncpg://user:password@host:5432/dbname
• REDIS_URL — Secrets Manager → your ElastiCache secret → rediss://auth-token@cluster-endpoint:6379
• COGNITO_USER_POOL_ID — Cognito console → User Pools → your pool → Pool ID
• COGNITO_APP_CLIENT_ID — Cognito console → your pool → App clients → Client ID
• COGNITO_REGION — Same as AWS_REGION
• S3_UPLOADS_BUCKET — The name of the uploads S3 bucket you created
• S3_REPORTS_BUCKET — The name of the reports S3 bucket you created
• SQS_INGESTION_QUEUE_URL — SQS console → your ingestion queue → URL
• SQS_INGESTION_DLQ_URL — SQS console → your ingestion dead-letter queue → URL
• SQS_COMPLETION_QUEUE_URL — SQS console → your ingestion completion notification queue → URL
• ECS_CLUSTER_ARN — ECS console → Clusters → your cluster → ARN
• WORKER_TASK_DEFINITION_ARN — ECS console → Task Definitions → orchestration worker → latest revision ARN
• INGESTION_TASK_DEFINITION_ARN — ECS console → Task Definitions → ingestion worker → latest revision ARN
• WORKER_SUBNET_IDS — VPC console → Subnets → your private subnets (comma-separated IDs)
• WORKER_SECURITY_GROUP_ID — EC2 console → Security Groups → worker task SG → ID
• WEBSOCKET_API_ENDPOINT — API Gateway console → your WebSocket API → stage → WebSocket URL (wss://...)
• MANAGEMENT_API_ENDPOINT — API Gateway console → your WebSocket API → stage → Connection URL (https://.../@connections)
• AWS_BEDROCK_REGION — Same as AWS_REGION
• SES_SENDER_ADDRESS — The verified email address or domain identity in SES

Orchestration worker task (injected via ECS Secrets integration, plus passed at RunTask time):

• DATABASE_URL — Same as above
• REDIS_URL — Same as above
• MANAGEMENT_API_ENDPOINT — Same as above
• S3_REPORTS_BUCKET — Same as above
• AWS_BEDROCK_REGION — Same as above
• SESSION_ID — Passed at RunTask launch time per session (not a static secret)
• WORKER_SUBNET_IDS — Same as above

Ingestion worker task:

• DATABASE_URL — Same as above
• S3_UPLOADS_BUCKET — Same as above
• SQS_INGESTION_QUEUE_URL — Same as above
• SQS_COMPLETION_QUEUE_URL — Same as above
• AWS_BEDROCK_REGION — Same as above

---

Phases

Phase 1: Infrastructure Foundation

This phase establishes the complete AWS infrastructure using CDK before any application code is written. Every downstream phase deploys into this foundation, so it must be complete and stable first. It covers networking, data stores, compute cluster definitions, IAM roles, and the CI/CD pipeline skeleton — everything the application needs to exist before it can run.

Contracts

CDK Stack organization

Four CDK stacks, each in its own file under infra/lib/:

NetworkStack (infra/lib/network-stack.ts)

• Creates the VPC with public, private-app, and private-data subnet tiers across two AZs
• Creates VPC endpoints: S3 Gateway endpoint, and Interface endpoints for Bedrock, Secrets Manager, and ECR
• Exports: vpc, privateAppSubnets (array), privateDataSubnets (array), publicSubnets (array)
• Creates security groups: albSg, fastApiSg, workerSg, ingestionSg, rdsSg, redisSg
• Security group rules:
  • albSg: inbound 443 from internet
  • fastApiSg: inbound from albSg only
  • workerSg: no inbound; outbound unrestricted (HTTPS to Management API)
  • ingestionSg: no inbound
  • rdsSg: inbound 5432 from fastApiSg and workerSg and ingestionSg
  • redisSg: inbound 6379 from fastApiSg and workerSg

DataStack (infra/lib/data-stack.ts)

• RDS PostgreSQL 16, Multi-AZ, db.r7g.large, 500 GB gp3 storage with auto-scaling to 2 TB, AsyncPostgresSaver requires the pgvector extension enabled in the parameter group
• ElastiCache Redis 7.x, cache.r7g.large, Multi-AZ with automatic failover, cluster mode disabled, auth token from Secrets Manager
• S3 uploads bucket and reports bucket (versioning on, KMS encryption, Block Public Access on, event notification to SQS ingestion queue on uploads bucket)
• SQS ingestion queue, SQS ingestion DLQ (4 receives before DLQ), SQS completion queue; ingestion queue message retention 24 hours
• CloudWatch alarm: RDS write IOPS > 3,000 sustained 5 minutes → SNS topic
• Exports: rdsEndpoint, rdsSecretArn, redisEndpoint, uploadsBucketName, reportsBucketName, ingestionQueueUrl, ingestionDlqUrl, completionQueueUrl

ComputeStack (infra/lib/compute-stack.ts)

• ECS cluster (Container Insights enabled)
• ECR repositories: fastapi-control-plane, orchestration-worker, ingestion-worker
• ALB (internet-facing, HTTPS listener, forwards to FastAPI target group)
• FastAPI ECS Service task definition: 1 vCPU, 2 GB RAM; IAM task role granting SQS send, ECS RunTask, Secrets Manager read, SES send, S3 read reports bucket, Bedrock invoke
• Orchestration worker task definition: 2 vCPU, 4 GB RAM; IAM task role granting RDS access (via DATABASE_URL secret), ElastiCache access, API Gateway Management API post, S3 read uploads and write reports, Bedrock invoke, Secrets Manager read; Fargate Spot capacity provider with on-demand fallback
• Ingestion worker ECS Service task definition: 1 vCPU, 2 GB RAM; IAM task role granting SQS receive/delete on ingestion queue, SQS send on completion queue, S3 read uploads, Bedrock invoke Titan Embeddings, Secrets Manager read; scales on SQS depth (min 0, max 10 tasks)
• FastAPI service: min 2, max 20 tasks; target tracking on ALB request count per target at 1,000 RPM
• AWS WAF v2 WebACL attached to API Gateway: rate limiting per IP, SQL injection managed rule group, size constraint rule (reject bodies over 50 MB)
• Exports: ecsClusterArn, fastApiServiceName, workerTaskDefinitionArn, ingestionTaskDefinitionArn, albDnsName

AppStack (infra/lib/app-stack.ts)

• API Gateway REST API (HTTP API v2) with Cognito JWT authorizer
• API Gateway WebSocket API with a Cognito Lambda Authorizer attached to the $connect route to validate the JWT and enforce workspace access. Routes include $connect, $disconnect, $default
• $connect Lambda: reads session_id from query string, writes connection ID to Redis set ws:connections:{session_id} with TTL equal to maximum session duration (7,200 seconds); also stores the Management API endpoint in Redis key ws:endpoint on first write
• $disconnect Lambda: removes connection ID from Redis set
• Lambda functions for REST CRUD routes (tenant, workspace, session list/get, report retrieval) that are lightweight enough for Lambda; session launch and long-lived operations proxy to the ALB
• Cognito User Pool and App Client (email+password, TOTP MFA support)
• Cognito post-confirmation Lambda trigger (creates User record in RDS, assigns default workspace)
• Amazon Amplify app pointing at the GitHub repository's frontend/ directory
• SES email identity configuration
• CloudWatch Log groups (90-day retention) for all services
• X-Ray tracing enabled on FastAPI task definition and API Gateway
• CloudWatch alarms → SNS topic: FastAPI task count < 2, worker task failure rate > 5%, DLQ depth > 0, API Gateway 5xx rate > 1%, ElastiCache node failure, RDS Multi-AZ failover

IAM role naming convention

All task roles follow the pattern {env}-{service}-task-role (e.g., prod-orchestration-worker-task-role). The GitHub Actions OIDC role is github-actions-deploy-role.

GitHub Actions workflow files

infra/lib/network-stack.ts, infra/lib/data-stack.ts, infra/lib/compute-stack.ts, infra/lib/app-stack.ts are the CDK source files.

.github/workflows/pr.yml — on pull request: CDK diff, lint, type-check (no deploy) .github/workflows/deploy.yml — on merge to main: build and push Docker images to ECR tagged with Git SHA, CDK deploy to staging, integration test placeholder (passes trivially at this phase), manual approval gate, CDK deploy to production

Deliverables:

• All four CDK stacks deploy cleanly to staging with cdk deploy --all
• RDS instance is reachable from within the VPC (verified via a connectivity check from a temporary task)
• ElastiCache Redis is reachable from within the VPC
• Both S3 buckets exist with correct policies
• All three ECR repositories exist and accept image pushes
• GitHub Actions PR workflow runs and produces a CDK diff on a test PR
• GitHub Actions deploy workflow runs and reaches the manual approval gate on merge to main

---

Phase 2: Database Schema and Shared Data Layer

This phase creates every database table, index, and constraint the application needs, and establishes the shared Python module that all backend services use to connect to RDS and Redis. Getting the schema fully right before any service code is written prevents migration conflicts between the control plane, worker, and ingestion services built in later phases.

Contracts

Database schema

All tables in a single Alembic migration file at backend/migrations/versions/001_initial_schema.py. The migration runs on FastAPI service startup.

Tables and columns (exact names used by all services):

tenants: id UUID PK, name TEXT NOT NULL, gdpr_region ENUM('eu','us') NOT NULL, erasure_requested_at TIMESTAMPTZ nullable, created_at TIMESTAMPTZ DEFAULT NOW()

workspaces: id UUID PK, tenant_id UUID FK(tenants.id) NOT NULL, name TEXT NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW()

users: id UUID PK (Cognito sub), tenant_id UUID FK(tenants.id) NOT NULL, email TEXT NOT NULL, display_name TEXT, created_at TIMESTAMPTZ DEFAULT NOW()

workspace_members: workspace_id UUID FK(workspaces.id), user_id UUID FK(users.id), role ENUM('owner','editor','viewer') NOT NULL, PK(workspace_id, user_id)

sessions: id UUID PK, workspace_id UUID FK(workspaces.id) NOT NULL, tenant_id UUID NOT NULL (denormalized for query efficiency), status ENUM('configuring','ingesting','running','paused','completed','failed','suspended') NOT NULL DEFAULT 'configuring', current_round INT NOT NULL DEFAULT 0, persona_count INT NOT NULL CHECK(persona_count BETWEEN 1 AND 10), topic TEXT NOT NULL, total_rounds INT NOT NULL DEFAULT 5, created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW()

persona_configs: id UUID PK, session_id UUID FK(sessions.id) NOT NULL, tenant_id UUID NOT NULL, archetype TEXT NOT NULL, display_name TEXT NOT NULL, grounding_doc_s3_key TEXT nullable, created_at TIMESTAMPTZ DEFAULT NOW()

messages: id UUID PK, session_id UUID FK(sessions.id) NOT NULL, tenant_id UUID NOT NULL, turn_sequence INT NOT NULL, agent_role ENUM('moderator','persona','analyst') NOT NULL, persona_config_id UUID FK(persona_configs.id) nullable, content TEXT NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW(), UNIQUE(session_id, turn_sequence)

analyst_reports: id UUID PK, session_id UUID FK(sessions.id) UNIQUE NOT NULL, tenant_id UUID NOT NULL, s3_key TEXT NOT NULL, summary TEXT, created_at TIMESTAMPTZ DEFAULT NOW()

ethnographic_chunks: id UUID PK, tenant_id UUID nullable, corpus TEXT NOT NULL DEFAULT 'base', source_doc_s3_key TEXT, content TEXT NOT NULL, embedding vector(1536) NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW()

product_context_chunks: id UUID PK, session_id UUID FK(sessions.id) NOT NULL, tenant_id UUID NOT NULL, content TEXT NOT NULL, embedding vector(1536) NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW()

gdpr_audit_log: id UUID PK, tenant_id UUID NOT NULL, action TEXT NOT NULL, completed_at TIMESTAMPTZ DEFAULT NOW(), details JSONB

Indexes:

• HNSW index on ethnographic_chunks.embedding with vector_cosine_ops
• HNSW index on product_context_chunks.embedding with vector_cosine_ops
• Index on messages(session_id, turn_sequence)
• Index on sessions(tenant_id, status)
• Index on product_context_chunks(session_id, tenant_id)
• Index on ethnographic_chunks(tenant_id) (partial: WHERE tenant_id IS NOT NULL)

RDS Row-Level Security policies: Enabled on sessions, messages, persona_configs, analyst_reports, product_context_chunks, workspace_members, workspaces, users. Each policy enforces tenant_id = current_setting('app.current_tenant_id')::uuid. The application sets this session variable before every query.

AsyncPostgresSaver tables: Created automatically by LangGraph on first worker startup. The migration does not create these — the worker creates them. The migration does create a langgraph_thread_tenant_map table: thread_id TEXT PK, tenant_id UUID NOT NULL, session_id UUID FK(sessions.id) NOT NULL — used exclusively by the GDPR erasure workflow to identify which checkpoint rows to delete.

Shared Python modules

All shared modules live in backend/shared/. Both the FastAPI control plane and the orchestration worker import from this package.

backend/shared/db.py

# Module-level engine and sessionmaker; call init_db() once at service startup

async def init_db(database_url: str) -> None:
    # Creates async SQLAlchemy engine and sessionmaker, stores as module globals
    # Enables pgvector extension registration on the engine

async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
    # Yields an AsyncSession; sets app.current_tenant_id session variable before yielding
    # Caller must set tenant_id on the session via set_tenant_context()

async def set_tenant_context(session: AsyncSession, tenant_id: str) -> None:
    # Executes: SET LOCAL app.current_tenant_id = '{tenant_id}'
    # Must be called at the start of every request handler before any query

async def close_db() -> None:
    # Disposes the engine; called on service shutdown

backend/shared/redis_client.py

async def init_redis(redis_url: str) -> None:
    # Creates an aioredis connection pool; stores as module global

async def get_redis() -> Redis:
    # Returns the module-global Redis client

async def close_redis() -> None:
    # Closes the Redis connection pool

# Redis key conventions (used by all services — must match exactly):
WS_CONNECTIONS_KEY = "ws:connections:{session_id}"   # Redis Set; members are connection IDs
HEARTBEAT_KEY = "ws:heartbeat:{session_id}"           # Redis String; value is ISO timestamp
MANAGEMENT_ENDPOINT_KEY = "ws:endpoint"               # Redis String; API Gateway Management API URL

backend/shared/models.py

SQLAlchemy ORM models matching the schema above exactly. Every model includes a tenant_id column. Model class names: Tenant, Workspace, User, WorkspaceMember, Session, PersonaConfig, Message, AnalystReport, EthnographicChunk, ProductContextChunk, GdprAuditLog, LangGraphThreadTenantMap.

backend/shared/schemas.py

Pydantic v2 models for API request/response shapes. Key schemas:

class SessionStatus(str, Enum):
    configuring = "configuring"
    ingesting = "ingesting"
    running = "running"
    paused = "paused"
    completed = "completed"
    failed = "failed"
    suspended = "suspended"

class AgentRole(str, Enum):
    moderator = "moderator"
    persona = "persona"
    analyst = "analyst"

class PersonaConfigCreate(BaseModel):
    archetype: str        # max 200 chars; control characters stripped
    display_name: str     # max 100 chars

class SessionCreate(BaseModel):
    workspace_id: UUID
    topic: str            # max 1000 chars; control characters stripped
    persona_count: int    # validated 1–10
    total_rounds: int     # validated 1–20; default 5
    personas: list[PersonaConfigCreate]  # length must equal persona_count

class MessageOut(BaseModel):
    id: UUID
    session_id: UUID
    turn_sequence: int
    agent_role: AgentRole
    persona_config_id: UUID | None
    content: str
    created_at: datetime

class SessionOut(BaseModel):
    id: UUID
    workspace_id: UUID
    status: SessionStatus
    current_round: int
    persona_count: int
    topic: str
    total_rounds: int
    created_at: datetime
    updated_at: datetime

class WorkspaceMessagePush(BaseModel):
    # Shape pushed to WebSocket clients by the orchestration worker
    event: str              # "agent_message" | "session_paused" | "session_suspended" | "session_completed" | "session_resuming"
    session_id: str
    message: MessageOut | None       # populated for "agent_message" events
    reason: str | None               # populated for pause/error events

backend/shared/errors.py

class AppError(Exception):
    def __init__(self, message: str, correlation_id: str): ...

class AuthorizationError(AppError): ...        # tenant or workspace access denied
class ValidationError(AppError): ...           # input validation failure
class SessionStateError(AppError): ...         # invalid session status transition
class IngestionError(AppError): ...            # document parsing or embedding failure
class CheckpointError(AppError): ...           # AsyncPostgresSaver write failure
class LLMError(AppError): ...                  # Bedrock call failure after retries exhausted
class CapacityError(AppError): ...             # 500-session limit reached

backend/shared/sanitize.py

def sanitize_user_text(text: str, max_length: int) -> str:
    # Strips control characters (Unicode categories Cc and Cf, except tab and newline)
    # Truncates to max_length
    # Returns sanitized string

Deliverables:

• Alembic migration runs cleanly against a fresh RDS instance and creates all tables, indexes, and RLS policies
• RLS policies verified: a query with one tenant's ID cannot return rows owned by another tenant
• HNSW indexes created on both vector columns
• All shared Python modules importable with no circular dependencies
• AsyncPostgresSaver tables created by a test worker startup and confirmed present in RDS

---

Phase 3: Authentication and Tenant Management

This phase delivers the Cognito integration, the JWT validation middleware, and the full tenant/workspace/user management API surface. Every subsequent phase depends on authenticated, tenant-scoped requests working correctly. No other business logic is built until this layer is solid.

Contracts

FastAPI application entry point

backend/control_plane/main.py

# Lifespan context manager calls: init_db(), init_redis(), run_migrations()
# Mounts routers: auth_router, tenant_router, workspace_router, user_router
# Global exception handler: catches AppError subclasses, returns JSON with correlation_id and generic message; logs full details to CloudWatch
# X-Ray middleware enabled
# CORS: allow only the Amplify frontend origin (from env ALLOWED_ORIGIN)
# Security response headers middleware: HSTS, X-Content-Type-Options, X-Frame-Options, CSP

JWT middleware

backend/control_plane/auth/middleware.py

async def verify_cognito_token(
    authorization: str,         # "Bearer <token>" from Authorization header
    cognito_user_pool_id: str,
    cognito_region: str,
) -> TokenClaims:
    # Fetches Cognito JWKS (cached with 1-hour TTL)
    # Validates JWT signature, expiry, issuer, and audience
    # Returns TokenClaims on success; raises AuthorizationError on failure

@dataclass
class TokenClaims:
    user_id: str          # Cognito sub
    tenant_id: str        # custom:tenant_id claim
    email: str
    workspace_roles: dict[str, str]  # workspace_id → role, from custom claims

FastAPI dependency get_current_user extracts and validates the token, then calls set_tenant_context(session, claims.tenant_id). Every route that touches the database uses this dependency.

Workspace authorization

backend/control_plane/auth/authorization.py

async def require_workspace_role(
    workspace_id: UUID,
    user_id: str,
    minimum_role: Literal["viewer", "editor", "owner"],
    db: AsyncSession,
) -> None:
    # Queries WorkspaceMember for (workspace_id, user_id)
    # Raises AuthorizationError if the member doesn't exist or role is below minimum
    # Role order: viewer < editor < owner

Tenant router

backend/control_plane/routers/tenants.py

POST /tenants
    body: TenantCreate(name: str, gdpr_region: Literal['eu', 'us'])
    returns: TenantOut(id, name, gdpr_region, created_at)
    # Creates Tenant row and a default Workspace; assigns the calling user as workspace owner
    # Only callable before the user has a tenant assigned (bootstrap flow)

GET /tenants/me
    returns: TenantOut
    # Returns the tenant for the calling user's tenant_id

Workspace router

backend/control_plane/routers/workspaces.py

GET /workspaces
    returns: list[WorkspaceOut(id, name, tenant_id, created_at)]
    # All workspaces for the calling user's tenant

POST /workspaces
    body: WorkspaceCreate(name: str)
    returns: WorkspaceOut
    # Caller becomes owner

GET /workspaces/{workspace_id}
    returns: WorkspaceOut
    # Requires viewer role or above

POST /workspaces/{workspace_id}/members
    body: MemberInvite(email: str, role: Literal['editor', 'viewer'])
    returns: MemberOut(workspace_id, user_id, role)
    # Requires owner role; sends SES invitation email; creates WorkspaceMember row
    # If the invited user doesn't exist in RDS yet, creates a pending record

DELETE /workspaces/{workspace_id}/members/{user_id}
    returns: 204
    # Requires owner role

User router

backend/control_plane/routers/users.py

GET /users/me
    returns: UserOut(id, tenant_id, email, display_name, created_at)

PATCH /users/me
    body: UserUpdate(display_name: str | None)
    returns: UserOut

Cognito post-confirmation Lambda

backend/lambdas/cognito_post_confirmation/handler.py

def handler(event: dict, context) -> dict:
    # Triggered by Cognito after email verification
    # Reads user attributes from event: sub, email, custom:tenant_id (may be absent on first user)
    # If tenant_id absent: creates a new Tenant (gdpr_region from event custom:gdpr_region attribute)
    # Creates User row in RDS with id = sub, email, tenant_id
    # Creates default Workspace "Default Workspace" and assigns user as owner
    # Returns event unchanged (Cognito trigger requirement)
    # Uses a direct psycopg2 connection (not the async SQLAlchemy engine) for simplicity in Lambda context

SES email helper

backend/shared/email.py

async def send_workspace_invitation(
    to_address: str,
    workspace_name: str,
    inviter_name: str,
    role: str,
) -> None:
    # Calls SES SendEmail; on SES unavailability logs the failure and enqueues to DLQ — does not raise

async def send_session_completion_notification(
    to_address: str,
    session_id: str,
    workspace_name: str,
) -> None:
    # Same non-blocking behavior

async def send_gdpr_erasure_confirmation(to_address: str) -> None:
    # Same non-blocking behavior

Deliverables:

• POST /tenants creates a tenant, workspace, and owner membership; returns correct TenantOut
• GET /workspaces returns only workspaces belonging to the calling user's tenant (cross-tenant isolation verified with two test tenants)
• An expired or malformed JWT returns 401 with a correlation ID and no stack trace
• Workspace invitation sends a real SES email in staging (SES sandbox: verified addresses only)
• Cognito post-confirmation Lambda creates User and Workspace records in RDS on first sign-up

---

Phase 4: Document Ingestion Pipeline

This phase delivers the full pipeline from file upload through chunking, embedding, and vector storage. It also establishes the error handling contract for all Bedrock calls, which later phases inherit. A session cannot move from configuring to running until ingestion completes successfully.

Contracts

Pre-signed URL endpoint

backend/control_plane/routers/uploads.py

POST /sessions/{session_id}/upload
    body: UploadRequest(filename: str, content_type: str, file_size_bytes: int)
    returns: UploadResponse(presigned_url: str, s3_key: str, upload_id: str)
    # Requires editor role on the session's workspace
    # Validates content_type is one of: application/pdf, application/vnd.openxmlformats-officedocument.presentationml.presentation, application/vnd.ms-powerpoint, application/vnd.apple.keynote
    # Validates file_size_bytes ≤ 52_428_800 (50 MB); raises ValidationError if exceeded
    # S3 key: uploads/{tenant_id}/{session_id}/{upload_id}/{filename}
    # Pre-signed URL expires in 900 seconds
    # Updates session status to 'ingesting'

POST /sessions/{session_id}/upload/{upload_id}/complete
    returns: 204
    # Called by frontend after direct S3 upload succeeds
    # Publishes SQS message to ingestion queue with IngestionJob payload
    # Session status remains 'ingesting' until ingestion worker reports completion

SQS message shapes

@dataclass
class IngestionJob:
    job_id: str           # UUID, unique per upload
    session_id: str
    tenant_id: str
    s3_key: str
    filename: str
    content_type: str

@dataclass
class IngestionComplete:
    job_id: str
    session_id: str
    tenant_id: str
    status: Literal['success', 'failed']
    error: str | None
    chunk_count: int | None

Ingestion worker

backend/ingestion_worker/main.py — entry point; polls SQS ingestion queue, dispatches to process_job()

backend/ingestion_worker/processor.py

async def process_job(job: IngestionJob) -> IngestionComplete:
    # 1. Fetch file from S3 into memory (raises IngestionError if file not found)
    # 2. Parse file: route to parse_pdf() or parse_pptx() based on content_type
    # 3. Chunk the extracted text into ~512-token segments with 10% overlap
    # 4. Generate embeddings for each chunk via embed_chunks()
    # 5. Upsert ProductContextChunk rows into pgvector (session_id + tenant_id tagged)
    # 6. Return IngestionComplete(status='success', chunk_count=N)
    # On IngestionError at any step: return IngestionComplete(status='failed', error=str(e))

backend/ingestion_worker/parsers.py

def parse_pdf(file_bytes: bytes) -> list[str]:
    # Uses PyMuPDF to extract text page by page
    # Returns list of page text strings
    # Raises IngestionError if the file cannot be opened or yields no text

def parse_pptx(file_bytes: bytes) -> list[str]:
    # Uses python-pptx to extract text from each slide's text frames
    # Returns list of slide text strings
    # Raises IngestionError if the file cannot be opened or yields no text

backend/ingestion_worker/chunker.py

def chunk_text(pages: list[str], target_tokens: int = 512, overlap_fraction: float = 0.10) -> list[str]:
    # Concatenates pages with paragraph separators
    # Splits into chunks of ~target_tokens tokens (approximated as chars / 4)
    # Overlap: last overlap_fraction * target_tokens tokens of each chunk prepended to next
    # Returns list of chunk strings; minimum 1 chunk even for short documents

backend/ingestion_worker/embedder.py

async def embed_chunks(chunks: list[str], bedrock_region: str) -> list[list[float]]:
    # Calls Bedrock Titan Embeddings v2 for each chunk sequentially
    # Returns list of 1536-dimension float vectors in the same order as input chunks
    # Error handling (first introduced here — all subsequent Bedrock embedding calls inherit this contract):
    #   - On transient Bedrock service error: retry up to 5 times with increasing delay (30s, 60s, 120s, 300s, 600s)
    #   - After 5 failures: raises IngestionError with the original error message
    #   - On non-retryable error (e.g., invalid input): raises IngestionError immediately without retry

SQS error handling and DLQ behavior

The ingestion worker SQS consumer:

• On IngestionError returned from process_job(): does not delete the SQS message; returns the message to the queue (message becomes visible again after the visibility timeout). The SQS queue's maxReceiveCount is 5; after 5 receives the message moves to the DLQ automatically.
• On IngestionComplete(status='failed') returned after all retries inside process_job(): publishes IngestionComplete to the completion queue, deletes the SQS message (the job is done — it just failed), updates session status to ingestion_failed in RDS.
• On IngestionComplete(status='success'): publishes IngestionComplete to the completion queue, deletes the SQS message, sends session_completion notification via SES.

Completion queue consumer

backend/control_plane/routers/ingestion_events.py — a background task started by the FastAPI lifespan that polls the SQS completion queue and updates session status accordingly:

• status='success': session status → configuring (ready to launch), updated_at refreshed
• status='failed': session status → ingestion_failed; SES notification sent to session owner

Grounding doc ingestion

Same pipeline as above, except:

• S3 key: uploads/{tenant_id}/grounding/{persona_config_id}/{filename}
• Chunks are inserted into ethnographic_chunks (not product_context_chunks), with tenant_id set and corpus = 'user'
• Triggered via POST /personas/{persona_config_id}/grounding-doc/upload (same pre-signed URL pattern)

Deliverables:

• Uploading a valid PDF via the pre-signed URL flow results in ProductContextChunk rows in pgvector with correct session_id and tenant_id
• Uploading a corrupted PDF results in session status ingestion_failed and an SES email sent (to a verified address in staging)
• A Bedrock Titan unavailability simulation (via mocking) triggers the retry sequence and ultimately DLQ placement
• EthnographicChunk rows inserted for a grounding doc are tagged with tenant_id and corpus = 'user'
• Cross-tenant query isolation verified: a pgvector ANN query with one tenant's tenant_id does not return another tenant's product_context_chunks

---

Phase 5: Session Management and Worker Lifecycle

This phase delivers the session CRUD API, the persona configuration endpoints, the ECS RunTask integration that spawns orchestration workers, and the heartbeat monitor that detects and recovers crashed workers. This is the control plane's most critical operational responsibility.

Contracts

Session router

backend/control_plane/routers/sessions.py

POST /workspaces/{workspace_id}/sessions
    body: SessionCreate
    returns: SessionOut
    # Requires editor role
    # Validates persona_count matches len(personas)
    # Creates Session (status='configuring') and PersonaConfig rows in one transaction
    # Returns SessionOut

GET /workspaces/{workspace_id}/sessions
    returns: list[SessionOut]
    # Requires viewer role
    # Returns sessions ordered by created_at DESC

GET /sessions/{session_id}
    returns: SessionOut
    # Requires viewer role on the session's workspace (resolved via session→workspace join)

DELETE /sessions/{session_id}
    returns: 204
    # Requires owner role
    # Soft-deletes: sets status='failed', does not delete rows
    # If session has an active worker task: calls ECS StopTask before marking failed
    # Purges ProductContextChunk rows for the session from pgvector

POST /sessions/{session_id}/launch
    returns: SessionOut (status='running')
    # Requires editor role
    # Valid only when session status is 'configuring' (raises SessionStateError otherwise)
    # Enforces the 500-session concurrency cap: counts active sessions (status IN ('running','paused','suspended')) across all tenants; raises CapacityError if ≥ 500
    # Calls spawn_worker_task(); on failure retries 3 times with increasing delay; after 3 failures sets session status='failed' and raises AppError
    # On success: sets session status='running', records worker ECS task ARN in a new column sessions.worker_task_arn TEXT nullable

POST /sessions/{session_id}/resume
    returns: SessionOut
    # Requires editor role
    # Valid when status is 'paused' or 'suspended'
    # Calls spawn_worker_task() with resume=True; same retry and failure behavior as launch

GET /sessions/{session_id}/messages
    query params: after_sequence: int | None (default None — returns all messages)
    returns: list[MessageOut] ordered by turn_sequence ASC
    # Requires viewer role
    # If after_sequence provided: returns messages WHERE turn_sequence > after_sequence
    # Used by frontend for transcript replay and missed-message recovery

Worker lifecycle module

backend/control_plane/worker_lifecycle.py

async def spawn_worker_task(
    session_id: str,
    tenant_id: str,
    resume: bool,
    ecs_client,
    cluster_arn: str,
    task_definition_arn: str,
    subnet_ids: list[str],
    security_group_id: str,
) -> str:
    # Calls ECS RunTask with:
    #   - cluster: cluster_arn
    #   - task definition: task_definition_arn
    #   - launch type: FARGATE (capacity provider: Spot with on-demand fallback)
    #   - network config: subnet_ids, security_group_id
    #   - environment overrides: SESSION_ID={session_id}, TENANT_ID={tenant_id}, RESUME={resume}
    # Returns the task ARN on success
    # Raises AppError on ECS API failure (after 3 retries with increasing delay)

async def stop_worker_task(task_arn: str, ecs_client, cluster_arn: str) -> None:
    # Calls ECS StopTask; ignores "task not found" errors (idempotent)
    # Raises AppError on other ECS API failures

Heartbeat monitor

backend/control_plane/heartbeat_monitor.py

async def run_heartbeat_monitor(
    poll_interval_seconds: int = 30,
    missed_threshold: int = 3,
) -> None:
    # Background asyncio task started by FastAPI lifespan
    # Every poll_interval_seconds:
    #   1. Attempt to acquire Redis distributed lock: SET ws:lock:heartbeat_monitor true NX EX 25
    #   2. If lock acquired, query RDS for sessions with status IN ('running', 'suspended') and worker_task_arn IS NOT NULL
    #   3. For each: read HEARTBEAT_KEY from Redis
    #   4. If key missing or last heartbeat older than (missed_threshold × poll_interval_seconds) seconds:
    #      a. Set session status = 'suspended' in RDS
    #      b. Call spawn_worker_task(resume=True) for the session
    #      c. Update sessions.worker_task_arn with the new task ARN
    #      d. Push WorkspaceMessagePush(event='session_suspended') to all session WebSocket connections via Management API
    #   5. Log all state transitions with session_id and correlation_id

async def push_to_session_connections(
    session_id: str,
    payload: WorkspaceMessagePush,
    redis,
    management_api_endpoint: str,
    apigw_management_client,
) -> None:
    # Reads Redis set WS_CONNECTIONS_KEY for the session
    # For each connection ID: posts payload JSON to Management API
    # On "gone" response: removes connection ID from Redis set
    # On other errors: logs with correlation_id and continues (does not raise)
    # This function is defined here and reused by the orchestration worker (same contract)

Valid session status transitions

Enforced in backend/control_plane/session_state.py:

VALID_TRANSITIONS: dict[SessionStatus, set[SessionStatus]] = {
    SessionStatus.configuring:  {SessionStatus.ingesting, SessionStatus.running, SessionStatus.failed},
    SessionStatus.ingesting:    {SessionStatus.configuring, SessionStatus.ingestion_failed, SessionStatus.failed},
    SessionStatus.running:      {SessionStatus.paused, SessionStatus.suspended, SessionStatus.completed, SessionStatus.failed},
    SessionStatus.paused:       {SessionStatus.running, SessionStatus.failed},
    SessionStatus.suspended:    {SessionStatus.running, SessionStatus.failed},
    SessionStatus.completed:    set(),
    SessionStatus.failed:       set(),
}

def validate_transition(current: SessionStatus, next: SessionStatus) -> None:
    # Raises SessionStateError if the transition is not in VALID_TRANSITIONS[current]

Deliverables:

• POST /sessions/{session_id}/launch with a valid session in configuring status spawns an ECS Fargate task and sets status to running
• POST /sessions/{session_id}/launch when 500 sessions are active returns a CapacityError response with a correlation ID
• An invalid status transition (e.g., launching a completed session) returns SessionStateError
• Manually stopping a worker ECS task in staging causes the heartbeat monitor to detect the missed heartbeats, mark the session suspended, and spawn a replacement task within 90 seconds
• GET /sessions/{session_id}/messages?after_sequence=5 returns only messages with turn_sequence > 5

---

Phase 6: LangGraph Orchestration Worker

This phase implements the orchestration worker — the most complex component in the system. It delivers the full LangGraph graph with Moderator, Persona, and Analyst agents; RAG retrieval from both pgvector namespaces; checkpoint persistence via AsyncPostgresSaver; real-time WebSocket push; and the heartbeat publisher. This phase has no external API surface — its output is message rows in RDS, checkpoint rows in RDS, and WebSocket pushes to clients.

Contracts

Worker entry point

backend/orchestration_worker/main.py

async def main() -> None:
    # Reads SESSION_ID, TENANT_ID, RESUME from environment variables
    # Calls init_db(), init_redis()
    # Creates AsyncPostgresSaver connection to RDS
    # Calls run_session_graph(session_id, tenant_id, resume)
    # On unhandled exception: logs with correlation_id, pushes WorkspaceMessagePush(event='session_suspended') to clients, exits non-zero

LangGraph state

backend/orchestration_worker/graph_state.py

class FocusGroupState(TypedDict):
    session_id: str
    tenant_id: str
    topic: str
    total_rounds: int
    current_round: int                    # incremented by Moderator at round start
    turn_sequence: int                    # global counter; incremented before each agent turn
    persona_configs: list[dict]           # list of {id, archetype, display_name}
    personas_spoken_this_round: list[str] # persona_config_id strings
    current_persona_id: str | None        # set by Moderator routing decision
    moderator_message: str | None         # Moderator's facilitation text for the current turn
    last_agent_role: AgentRole | None
    session_complete: bool                # set True by Moderator when all rounds done

The graph state holds only current-round working context. It does not accumulate message history. Full history is retrieved from RDS by the Analyst Agent at the end.

Graph definition

backend/orchestration_worker/graph.py

def build_graph(checkpointer: AsyncPostgresSaver) -> CompiledGraph:
    # Builds a StateGraph[FocusGroupState] with nodes:
    #   - "moderator": calls moderator_node()
    #   - "persona": calls persona_node()
    #   - "analyst": calls analyst_node()
    #   - "end": terminal node
    # Edges:
    #   - Entry: "moderator"
    #   - Conditional edge from "moderator": if session_complete → "analyst", elif current_persona_id set → "persona"
    #   - Edge from "persona" → "moderator"
    #   - Edge from "analyst" → "end"
    # Compiled with checkpointer=checkpointer
    # Returns compiled graph

async def run_session_graph(
    session_id: str,
    tenant_id: str,
    resume: bool,
    checkpointer: AsyncPostgresSaver,
) -> None:
    # thread_config = {"configurable": {"thread_id": session_id}}
    # If not resume: initializes FocusGroupState from session and persona_config rows in RDS
    # Inserts a LangGraphThreadTenantMap row (thread_id=session_id, tenant_id=tenant_id, session_id=session_id)
    # Calls graph.astream(initial_state or None, config=thread_config) — LangGraph resumes from checkpoint if resume=True
    # Publishes heartbeat every 30 seconds during graph execution (asyncio background task)
    # On graph completion (reaches "end"): sets session status='completed' in RDS

Moderator node

backend/orchestration_worker/agents/moderator.py

async def moderator_node(state: FocusGroupState, config: RunnableConfig) -> FocusGroupState:
# Dependencies (db, bedrock) are resolved from backend.shared globals or config["configurable"]    
    # 1. Increment turn_sequence by 1
    # 2. If all personas have spoken this round and current_round < total_rounds:
    #       increment current_round, reset personas_spoken_this_round
    #    If current_round >= total_rounds:
    #       set session_complete=True, return state (routes to analyst)
    # 3. Select the next persona: the first persona_config not in personas_spoken_this_round
    # 4. Call Claude via Bedrock to generate facilitation text:
    #       System prompt: Moderator role description with topic, current round, total rounds
    #       User prompt: "Generate a focused question or prompt for {persona.display_name} ({persona.archetype}) given the current debate state."
    #       Response: moderator_message string
    # 5. Write moderator_message to RDS messages table (agent_role='moderator', turn_sequence=state.turn_sequence, persona_config_id=None)
    # 6. Push WorkspaceMessagePush(event='agent_message', message=MessageOut(...)) to clients
    # 7. Return updated state with current_persona_id, moderator_message, turn_sequence, current_round
    # LLM error handling (first introduced here for LLM agent calls — persona and analyst nodes inherit this contract):
    #   - On transient Bedrock throttling: retry up to 5 times with increasing delay starting at 2 seconds, up to 60 seconds
    #   - If Bedrock is unresponsive for more than 60 seconds total: pause session (status='paused'), commit checkpoint, push WorkspaceMessagePush(event='session_paused'), raise LLMError
    #   - On CheckpointError from AsyncPostgresSaver: retry up to 3 times with increasing delay; on all failures: pause session, push event, raise CheckpointError

Persona node

backend/orchestration_worker/agents/persona.py

async def persona_node(state: FocusGroupState, config: RunnableConfig) -> FocusGroupState:
# Dependencies (db, bedrock, redis) are resolved from backend.shared globals or config["configurable"]
    # 1. Load PersonaConfig for state.current_persona_id from RDS
    # 2. Retrieve top-5 ethnographic chunks for the persona's archetype from pgvector:
    #       ANN query on ethnographic_chunks WHERE (tenant_id = state.tenant_id OR tenant_id IS NULL)
    #       ordered by cosine similarity to embed(persona.archetype)
    # 3. Retrieve top-5 product context chunks for the current topic from pgvector:
    #       ANN query on product_context_chunks WHERE session_id = state.session_id AND tenant_id = state.tenant_id
    #       ordered by cosine similarity to embed(state.moderator_message)
    # 4. Increment turn_sequence by 1
    # 5. Assemble system prompt:
    #       [ROLE] You are {persona.display_name}, a {persona.archetype}.
    #       [GROUNDING] {ethnographic chunks concatenated}
    #       [PRODUCT CONTEXT] {product context chunks concatenated}
    #       [DEBATE CONTEXT] {state.moderator_message}
    #       [INSTRUCTION] Respond in character. Max 300 words.
    # 6. Call Claude via Bedrock (same LLM error handling contract as moderator_node)
    # 7. Sanitize response via sanitize_user_text() — strip control chars, max 2000 chars
    # 8. Write persona response to RDS messages table (agent_role='persona', persona_config_id=current_persona_id, turn_sequence=state.turn_sequence)
    # 9. Push WorkspaceMessagePush(event='agent_message') to clients
    # 10. Add current_persona_id to personas_spoken_this_round; clear current_persona_id
    # 11. Return updated state

async def embed_query(text: str, bedrock_client) -> list[float]:
    # Calls Bedrock Titan Embeddings v2 for a single string
    # Returns 1536-dimension float vector
    # Same error handling contract as Phase 4's embed_chunks (ingestion worker)

Analyst node

backend/orchestration_worker/agents/analyst.py

async def analyst_node(state: FocusGroupState, config: RunnableConfig) -> FocusGroupState:
# Dependencies (db, bedrock, s3_client) are resolved from backend.shared globals or config["configurable"]
    # 1. Fetch all messages for state.session_id from RDS ordered by turn_sequence ASC
    # 2. Format messages as a transcript string (role: content, one per line)
    # 3. Call Claude via Bedrock with the full transcript:
    #       System prompt: Analyst role; instructs generation of a structured JSON report
    #       Report schema:
    #         {
    #           "positioning_findings": str,
    #           "pricing_sensitivity": str,
    #           "feature_prioritization": list[str],
    #           "dissenting_views": list[{"persona": str, "view": str}],
    #           "summary": str
    #         }
    # 4. Parse Claude's response as JSON (on parse failure: retry the Claude call once; on second failure: store raw text under "raw_output" key)
    # 5. Write the JSON blob to S3 at reports/{tenant_id}/{session_id}/report.json
    # 6. Write AnalystReport row to RDS with s3_key and summary (report["summary"])
    # 7. Set session status='completed' in RDS
    # 8. Push WorkspaceMessagePush(event='session_completed') to clients
    # 9. Send SES session completion notification to workspace owner
    # 10. Return state (session_complete already True)
    # Same LLM error handling contract as moderator_node

Heartbeat publisher

backend/orchestration_worker/heartbeat.py

async def publish_heartbeat(session_id: str, redis, interval_seconds: int = 30) -> None:
    # Asyncio background task; loops until cancelled
    # Every interval_seconds: writes current UTC ISO timestamp to HEARTBEAT_KEY in Redis with TTL of interval_seconds × 5

RAG retrieval module

backend/orchestration_worker/retrieval.py

async def retrieve_ethnographic_chunks(
    archetype_embedding: list[float],
    tenant_id: str,
    top_k: int,
    db: AsyncSession,
) -> list[str]:
    # ANN query: SELECT content FROM ethnographic_chunks
    #   WHERE (tenant_id = :tenant_id OR tenant_id IS NULL)
    #   ORDER BY embedding <=> :query_vector LIMIT :top_k
    # Returns list of content strings

async def retrieve_product_context_chunks(
    query_embedding: list[float],
    session_id: str,
    tenant_id: str,
    top_k: int,
    db: AsyncSession,
) -> list[str]:
    # ANN query: SELECT content FROM product_context_chunks
    #   WHERE session_id = :session_id AND tenant_id = :tenant_id
    #   ORDER BY embedding <=> :query_vector LIMIT :top_k
    # Returns list of content strings

Deliverables:

• A fully configured session with a 2-persona, 2-round configuration runs to completion in staging: all messages written to RDS with correct turn_sequence values, analyst report written to S3 and indexed in RDS, session status set to completed
• Manually stopping the ECS task mid-session and re-launching it results in the graph resuming from the last committed checkpoint, with turn_sequence continuing from where it left off
• WebSocket clients connected during the session receive all agent messages in order
• A simulated Bedrock outage lasting >60 seconds causes the session to pause; the user receives a session_paused WebSocket event; the session is resumable
• Cross-tenant pgvector retrieval isolation verified: a persona agent for Tenant A does not retrieve product_context_chunks belonging to Tenant B's session

---

Phase 7: Frontend Application

This phase delivers the complete Next.js frontend — workspace dashboard, session configuration wizard, live debate view, and session history/report viewer. All data comes from the REST API and WebSocket API built in earlier phases. The frontend makes no direct calls to AWS services except the pre-signed S3 upload.

Contracts

Application structure

frontend/
  app/
    (auth)/
      login/page.tsx
      signup/page.tsx
    (app)/
      layout.tsx                  # auth guard; redirects to /login if no valid token
      dashboard/page.tsx          # workspace overview (RSC)
      sessions/
        new/page.tsx              # session configuration wizard (client component)
        [sessionId]/
          page.tsx                # live debate view (client component)
          report/page.tsx         # analyst report viewer (RSC)
      settings/
        workspace/page.tsx        # member management
        profile/page.tsx
  components/
    debate/
      TranscriptPanel.tsx         # renders message list; scrolls to bottom on new message
      AgentMessageCard.tsx        # single message bubble; styled by agent_role
      ConnectionStatus.tsx        # WebSocket connection state indicator
    session/
      ConfigWizard.tsx            # multi-step form: topic → personas → upload → launch
      PersonaCard.tsx
      FileUploader.tsx            # handles pre-signed URL upload flow
    workspace/
      MemberTable.tsx
      InviteMemberModal.tsx
    shared/
      ErrorBoundary.tsx
      LoadingSpinner.tsx
      CorrelationErrorDisplay.tsx # shows correlation_id to user on API errors
  lib/
    api-client.ts
    websocket-client.ts
    auth.ts
  hooks/
    useSession.ts
    useWebSocket.ts
    useTranscript.ts

API client

frontend/lib/api-client.ts

// Base client that attaches the Cognito ID token to every request
// All methods return typed responses; on API error, extracts correlation_id and throws ApiError

interface ApiError {
  message: string;
  correlationId: string;
}

// Methods (all async, all return typed shapes matching backend schemas):
createSession(workspaceId: string, body: SessionCreate): Promise<SessionOut>
launchSession(sessionId: string): Promise<SessionOut>
resumeSession(sessionId: string): Promise<SessionOut>
getSession(sessionId: string): Promise<SessionOut>
listSessions(workspaceId: string): Promise<SessionOut[]>
getMessages(sessionId: string, afterSequence?: number): Promise<MessageOut[]>
requestUploadUrl(sessionId: string, body: UploadRequest): Promise<UploadResponse>
confirmUpload(sessionId: string, uploadId: string): Promise<void>
uploadFileToS3(presignedUrl: string, file: File): Promise<void>
getReport(sessionId: string): Promise<AnalystReportOut>
listWorkspaceMembers(workspaceId: string): Promise<MemberOut[]>
inviteMember(workspaceId: string, body: MemberInvite): Promise<MemberOut>
removeMember(workspaceId: string, userId: string): Promise<void>

WebSocket client

frontend/lib/websocket-client.ts

class FocusGroupWebSocket {
  constructor(sessionId: string, wsEndpoint: string, onMessage: (msg: WorkspaceMessagePush) => void)
  
  connect(): void
  // Opens WebSocket to wsEndpoint?session_id={sessionId} with Cognito token in query string
  // On open: sets status to 'connected'
  // On message: parses JSON as WorkspaceMessagePush, calls onMessage
  // On close or error: sets status to 'disconnected'; retries up to 5 times with increasing delay
  // After 5 failed retries: sets status to 'failed'; stops retrying
  
  disconnect(): void
  // Closes WebSocket cleanly
  
  status: 'connecting' | 'connected' | 'disconnected' | 'failed'
}

Live debate hook

frontend/hooks/useTranscript.ts

function useTranscript(sessionId: string): {
  messages: MessageOut[];
  connectionStatus: string;
  sessionStatus: SessionStatus;
  pauseReason: string | null;
}
// On mount:
//   1. Fetches all existing messages via getMessages(sessionId) ordered by turn_sequence
//   2. Opens FocusGroupWebSocket
//   3. On 'agent_message' events: appends message to local state if turn_sequence > last seen
//   4. On 'session_paused' / 'session_suspended' / 'session_completed': updates sessionStatus and pauseReason
//   5. On WebSocket reconnect: fetches messages with afterSequence=last_seen_turn_sequence to fill the gap
// Deduplicates by turn_sequence (a message already in state is never added twice)

Session config wizard

frontend/components/session/ConfigWizard.tsx

Multi-step client component. Steps in order:

1. Topic — text input for focus group topic (max 1,000 chars, client-side enforced)
2. Personas — add 1–10 personas; each has display_name and archetype text fields; archetype max 200 chars
3. Upload — optional PRD/deck upload using FileUploader; validates MIME type and file size < 50 MB client-side before requesting pre-signed URL; shows ingestion progress by polling getSession() every 3 seconds until status is no longer ingesting
4. Review & Launch — summary of configuration; "Launch Session" button calls launchSession() then navigates to /sessions/{sessionId}

File uploader flow:

1. Call requestUploadUrl() → get presigned_url and upload_id
2. PUT file bytes directly to presigned_url via uploadFileToS3()
3. Call confirmUpload(sessionId, upload_id)
4. Begin polling session status

Auth module

frontend/lib/auth.ts

// Wraps Amplify Auth SDK
// signIn(email, password): Promise<void>
// signOut(): Promise<void>
// getIdToken(): Promise<string>   — used by api-client.ts and websocket-client.ts
// getCurrentUser(): Promise<{ userId: string; email: string } | null>

Deliverables:

• A user can sign up, confirm email, and land on the workspace dashboard with no errors
• A user can complete the 4-step wizard, upload a PDF, wait for ingestion, and launch a session
• The live debate view shows agent messages appearing in real time as the orchestration worker publishes them
• On WebSocket disconnect and reconnect, the transcript is rebuilt from the REST API with no duplicates or gaps
• The report viewer renders the analyst report's structured JSON fields (positioning, pricing, features, dissenting views) after session completion
• A workspace owner can invite a member by email; the invited user receives an SES email (in staging: verified address)

---

Phase 8: GDPR Erasure Workflow and Data Lifecycle

This phase implements the GDPR right-to-erasure Step Functions state machine, the checkpoint pruning job for completed sessions, and the product context purge on session deletion. These are all correctness-critical operations for compliance — each step must be independently retryable and auditable.

Contracts

GDPR erasure state machine

Implemented as an AWS Step Functions Standard Workflow. Defined in CDK. The state machine ARN is stored in an environment variable on the FastAPI service: ERASURE_STATE_MACHINE_ARN.

Steps in order (each step is a Lambda function):

1. soft_delete_user — Sets users.erasure_requested_at = NOW() and marks the user inactive in Cognito (admin disable user); writes a gdpr_audit_log row with action 'erasure_started'

2. erase_message_content — Updates all messages.content = '[erased]' WHERE tenant_id = :tenant_id; this is a bulk UPDATE, not row-by-row deletion

3. delete_s3_uploads — Lists and deletes all S3 objects under prefix uploads/{tenant_id}/; lists and deletes all S3 objects under reports/{tenant_id}/; handles paginated listing

4. purge_product_context_vectors — Deletes all product_context_chunks WHERE tenant_id = :tenant_id

5. purge_ethnographic_vectors — Deletes all ethnographic_chunks WHERE tenant_id = :tenant_id (user-supplied grounding only; base corpus rows have NULL tenant_id and are unaffected)

6. purge_langgraph_checkpoints — Queries langgraph_thread_tenant_map for all thread_id values WHERE tenant_id = :tenant_id; deletes corresponding rows from AsyncPostgresSaver checkpoint tables (checkpoints and checkpoint_writes) by thread_id; deletes rows from langgraph_thread_tenant_map

7. write_audit_completion — Writes a gdpr_audit_log row with action 'erasure_completed'; calls send_gdpr_erasure_confirmation() via SES (non-blocking — email failure does not fail this step)

Each Lambda step uses the same error handling contract: retries up to 3 times with increasing delay before the Step Functions state machine moves to a failure state and publishes to SNS. Steps are idempotent — re-running a step that already completed leaves state unchanged.

GDPR erasure trigger endpoint

backend/control_plane/routers/gdpr.py

POST /gdpr/erasure
    body: ErasureRequest(user_id: str)
    returns: ErasureStarted(execution_arn: str)
    # Restricted to platform admin role (checked via a separate Cognito group claim)
    # Starts the Step Functions execution with input: {"tenant_id": ..., "user_id": ...}
    # Returns the execution ARN immediately; completion is async

Checkpoint pruning job

backend/orchestration_worker/cleanup/prune_checkpoints.py — runs as a scheduled ECS task (EventBridge Scheduler, daily):

async def prune_old_checkpoints() -> None:
    # Queries sessions WHERE status IN ('completed', 'failed') AND updated_at < NOW() - INTERVAL '30 days'
    # For each: deletes rows from AsyncPostgresSaver checkpoint tables keyed to the session's thread_id
    # Deletes the langgraph_thread_tenant_map row
    # Logs count of pruned sessions

Session deletion purge

Called from DELETE /sessions/{session_id} in Phase 5's session router (not a separate service). Inline in the delete handler:

async def purge_session_vectors(session_id: str, tenant_id: str, db: AsyncSession) -> None:
    # Deletes all product_context_chunks WHERE session_id = :session_id AND tenant_id = :tenant_id
    # Does not touch ethnographic_chunks (grounding docs are tenant-scoped, not session-scoped)

Deliverables:

• Triggering the erasure state machine against a test tenant in staging results in all seven steps completing successfully: message content replaced with [erased], S3 objects deleted, pgvector rows purged, checkpoint rows deleted, audit log entry written
• A partial failure in step 3 (S3 deletion) causes that step to retry up to 3 times; on repeated failure the state machine enters a failure state and publishes to SNS (verified by simulating an S3 permission error)
• The checkpoint pruning job runs in staging and removes checkpoint rows for sessions older than 30 days in completed or failed status
• DELETE /sessions/{session_id} removes all product_context_chunks for that session from pgvector
• A pgvector ANN query after erasure returns zero rows for the erased tenant's data

---

Phase 9: Observability, Hardening, and Production Readiness

This phase adds structured logging with correlation IDs across all services, X-Ray trace propagation, the remaining CloudWatch alarms, WAF configuration, and the load and failover tests that verify the system's behavior at scale before going to production.

Contracts

Structured logging

backend/shared/logging_config.py

def configure_logging(service_name: str) -> None:
    # Configures Python logging to output newline-delimited JSON to stdout
    # Every log record includes: service, environment, correlation_id (from context var), session_id (if set), timestamp, level, message
    # No environment variable values are included in log output (secrets masked)

# Context variables (Python contextvars):
correlation_id_var: ContextVar[str]   # set at request start; propagated to all log calls within the request
session_id_var: ContextVar[str | None]  # set by worker on startup

Every FastAPI request generates a UUID correlation ID at middleware entry. It is returned in the X-Correlation-ID response header and included in every log line for the request. Orchestration workers generate a correlation ID per node transition.

X-Ray trace propagation

• FastAPI: X-Ray SDK middleware applied at app startup; patches asyncpg, aiobotocore, and aioredis automatically
• Orchestration worker: X-Ray SDK applied; patches Bedrock client and asyncpg; trace IDs logged alongside correlation IDs
• API Gateway: X-Ray active tracing enabled (configured in CDK AppStack)

CloudWatch alarms (additions to Phase 1 baseline)

All defined in CDK AppStack:

• Worker task session_paused events (custom metric from CloudWatch Logs metric filter): > 10 per minute → warning alarm
• Median turn_latency_seconds (custom metric from worker logs) > 30 seconds sustained 5 minutes → P2 alarm
• Ingestion DLQ depth > 0 → P2 alarm (already in Phase 1; confirmed wired to SNS)
• RDS write IOPS > 3,000 sustained 5 minutes → P1 alarm (already in Phase 1)
• ElastiCache connection count > 80% of maxclients → P2 alarm

Load test suite

tests/load/ — uses locust or direct asyncio scripts; not a production runtime dependency:

• 500_sessions_concurrency.py: spawns 500 simulated sessions against staging, each running a 2-persona, 2-round graph; asserts RDS write IOPS remain below 3,000; asserts median turn latency < 30 seconds
• bedrock_throttle_recovery.py: simulates sustained Bedrock throttling (via mocked Bedrock endpoint in staging); asserts sessions enter paused status and are resumable
• fargate_spot_interruption.py: manually stops 10 worker tasks; asserts all 10 are detected by heartbeat monitor within 90 seconds and resume successfully from RDS checkpoints
• redis_failover.py: triggers ElastiCache primary failover; asserts sessions resume pushing after the failover window; asserts clients can retrieve missed messages by turn_sequence
• gdpr_erasure_integrity.py: runs erasure against a test tenant with active sessions; asserts zero data remains across RDS, S3, and pgvector after completion

WAF rules (finalized in CDK)

• Rate limit: 100 requests per 5-minute window per IP to the REST API
• Rate limit: 10 WebSocket $connect attempts per 5-minute window per IP
• AWS Managed Rules: AWSManagedRulesCommonRuleSet, AWSManagedRulesSQLiRuleSet
• Body size limit: reject requests with body > 52,428,800 bytes (50 MB) — matches the upload validation

Pre-launch checklist (verified in staging before production deploy)

Each item maps to a load test or manual verification:

• Bedrock quota increase confirmed received for Claude claude-sonnet-4-5 tokens per minute
• 500-session concurrent load test passes (RDS IOPS < 3,000, turn latency < 30s)
• Fargate Spot interruption recovery verified (10 sessions, all resume within 90s)
• GDPR erasure verified against a test tenant (all data purged, audit log written)
• Redis failover recovery verified (sessions resume push, clients recover missed messages)
• WAF rate limiting verified (IP blocked after 100 requests in 5 minutes returns 429)
• CloudWatch alarms verified: each alarm fires as expected when its threshold is breached in staging

Deliverables:

• Every API response includes an X-Correlation-ID header; every log line for that request includes the same ID
• CloudWatch Logs Insights query on correlation_id returns all log lines for a given request across FastAPI, worker, and Lambda services
• X-Ray service map in the AWS console shows a complete trace from API Gateway through FastAPI through RDS for a session launch request
• The 500-session load test completes with RDS write IOPS staying below 3,000 and no sessions failing due to checkpoint errors
• WAF blocks an IP that exceeds the rate limit and returns a generic 429 with no internal details
• All five load test scripts pass in staging; results documented in a staging sign-off artifact before the manual production approval gate is opened

Tickets Document

Here is your step-by-step ticket backlog to build MockBoard. Whether you're pasting these sequentially into Claude Code or importing them into Linear for your team, these tickets provide the exact implementation steps to bring this architecture to life.

Phase 1: Infrastructure Foundation

Ticket 1.1: Create Network Foundation Stack

Objective: Provision the foundational AWS networking infrastructure, including the VPC, subnets across multiple availability zones, security groups, and VPC endpoints.

Context: This ticket establishes the secure and isolated network environment required by all other services, as specified in the Architecture Design Document. Creating the NetworkStack first allows subsequent data and compute resources to be placed into the correct, pre-configured private subnets with appropriate firewall rules.

Technical Directives:

• Implement a CDK stack named NetworkStack in infra/lib/network-stack.ts.
• Create a new VPC spanning two Availability Zones.
• Configure three subnet tiers within the VPC: public, private-app, and private-data.
• Create a gateway VPC endpoint for S3.
• Create interface VPC endpoints for Bedrock, Secrets Manager, and ECR.
• Define security groups for the ALB, FastAPI service, orchestration/ingestion workers, RDS, and Redis.
• Configure security group ingress and egress rules precisely as defined in the phase contract, allowing traffic only from specified sources (e.g., RDS only accepts traffic from the application and worker security groups).
• The stack must export its created VPC, subnet, and security group resources for consumption by other stacks.

Scope Boundaries:

• AWS CDK: infra/lib/network-stack.ts

Acceptance Criteria:

• The NetworkStack CDK stack deploys successfully.
• A VPC with public, private-app, and private-data subnets exists in the AWS account.
• All specified security groups are created with the correct ingress/egress rules.
• All specified VPC endpoints are created and associated with the correct VPC route tables.
• The stack's outputs (VPC, subnets, security groups) are available for import by other CDK stacks.

Dependencies: None

---

Ticket 1.2: Create Data and Storage Stack

Objective: Provision all stateful data stores, including the PostgreSQL database, Redis cache, S3 buckets, SQS queues, and associated monitoring.

Context: This ticket creates the persistence and messaging layer defined in the DataStack contract, using the network foundation from the previous ticket. The Architecture Design Document specifies using managed AWS services (RDS, ElastiCache, S3, SQS) to minimize operational overhead for the team.

Technical Directives:

• Implement a CDK stack named DataStack in infra/lib/data-stack.ts.
• Provision an RDS PostgreSQL 16 instance (db.r7g.large) in the private-data subnets, configured for Multi-AZ.
• Enable the pgvector extension on the RDS instance.
• Configure RDS storage as 500 GB gp3 with auto-scaling up to 2 TB.
• Provision an ElastiCache for Redis 7.x cluster (cache.r7g.large) in the private-data subnets, configured for Multi-AZ with automatic failover and cluster mode disabled.
• Retrieve the Redis authentication token from AWS Secrets Manager.
• Create two S3 buckets: one for uploads and one for reports, both with versioning, server-side KMS encryption, and Block Public Access enabled.
• Configure an S3 event notification on the uploads bucket to send messages to an SQS ingestion queue.
• Create three SQS queues: ingestion, ingestion-dlq, and completion. The ingestion queue must be configured to move messages to the DLQ after 4 failed receives and retain messages for 24 hours.
• Create a CloudWatch alarm that triggers an SNS topic if the RDS instance's write IOPS exceeds 3,000 for a sustained 5-minute period.
• The stack must export the names, ARNs, and endpoints of its created resources.

Scope Boundaries:

• AWS CDK: infra/lib/data-stack.ts

Acceptance Criteria:

• The DataStack CDK stack deploys successfully.
• The RDS instance is running, accessible from the private-app subnets, and has the pgvector extension available.
• The ElastiCache for Redis cluster is running and accessible from the private-app subnets.
• Both S3 buckets and all three SQS queues are created with the specified configurations.
• The CloudWatch alarm for RDS write IOPS is created and configured.
• The created RDS and Redis instances are ready to be configured with the schema and data models defined in the Following Phase.

Dependencies: Ticket 1

---

Ticket 1.3: Create Compute and Container Stack

Objective: Provision the AWS ECS Fargate compute infrastructure, including the cluster, ECR repositories, Application Load Balancer, and task definitions for all backend services.

Context: This ticket defines the runtime environments for the application's containerized services, as outlined in the Architecture Design Document. It implements the core "one task per session" strategy for orchestration workers and sets up the auto-scaling control plane service.

Technical Directives:

• Implement a CDK stack named ComputeStack in infra/lib/compute-stack.ts.
• Create an ECS cluster with Container Insights enabled.
• Create three ECR repositories: fastapi-control-plane, orchestration-worker, and ingestion-worker.
• Provision an internet-facing Application Load Balancer in the public subnets.
• Define an ECS task definition for the FastAPI control plane (1 vCPU, 2GB RAM) with an IAM role granting permissions for SQS, ECS RunTask, Secrets Manager, SES, S3, and Bedrock.
• Define an ECS task definition for the orchestration worker (2 vCPU, 4GB RAM) with an IAM role granting permissions for RDS, ElastiCache, API Gateway Management API, S3, Bedrock, and Secrets Manager.
• Define an ECS task definition for the ingestion worker (1 vCPU, 2GB RAM) with an IAM role granting permissions for SQS, S3, Bedrock, and Secrets Manager.
• Configure an ECS Service for the FastAPI control plane with a minimum of 2 and a maximum of 20 tasks, scaling based on an ALB request count target of 1,000 RPM.
• Configure an ECS Service for the ingestion worker with a minimum of 0 and a maximum of 10 tasks, scaling based on the depth of the SQS ingestion queue.
• Attach an AWS WAF WebACL to the ALB with rules for rate limiting, SQL injection, and request size constraints.
• Configure the orchestration worker task definition to use a Fargate Spot capacity provider with an on-demand fallback.
• The stack must export the ARNs and names of its created resources.

Scope Boundaries:

• AWS CDK: infra/lib/compute-stack.ts

Acceptance Criteria:

• The ComputeStack CDK stack deploys successfully.
• The ECS cluster and all three ECR repositories are created.
• The ALB is created and correctly forwards traffic to the FastAPI service's target group.
• All three ECS task definitions and the two ECS services are created with their specified configurations and IAM roles.
• The ECR repositories are ready to receive Docker images from the CI/CD pipeline.
• The task definitions are ready to be used by the application code deployments in subsequent phases.

Dependencies: Ticket 1, Ticket 2

---

Ticket 1.4: Create Application Integration Stack

Objective: Provision the application-facing services, including API Gateways, Lambda authorizers and handlers, the Cognito User Pool, and the Amplify frontend application.

Context: This ticket creates the primary user entry points and authentication layer as defined in the AppStack contract. It integrates serverless components like API Gateway and Lambda with the stateful data and compute backends to complete the infrastructure setup.

Technical Directives:

• Implement a CDK stack named AppStack in infra/lib/app-stack.ts.
• Create an API Gateway REST API (HTTP API v2) with a default proxy route to the ALB created in the ComputeStack.
• Create an API Gateway WebSocket API with $connect, $disconnect, and $default routes.
• Create a Cognito User Pool and App Client configured for email/password sign-in and TOTP MFA support.
• Configure a Cognito JWT authorizer on the REST API and a Lambda authorizer on the WebSocket $connect route.
• Implement the $connect Lambda to read a session_id from the query string and store the connection ID in the Redis set ws:connections:{session_id}. The Lambda must also write the Management API endpoint URL to the Redis key ws:endpoint.
• Implement the $disconnect Lambda to remove the connection ID from the corresponding Redis set.
• Implement a Cognito post-confirmation Lambda trigger that will create a new user record in RDS. (Note: The database connection logic will be a placeholder until Phase 2).
• Create an AWS Amplify application configured to point to the project's GitHub repository.
• Configure an SES email identity for sending transactional emails.
• Enable AWS X-Ray tracing on the FastAPI task definition and the API Gateways.
• Create CloudWatch alarms for FastAPI task count, worker failure rate, DLQ depth, API Gateway 5xx rate, ElastiCache node failure, and RDS failover, all reporting to an SNS topic.

Scope Boundaries:

• AWS CDK: infra/lib/app-stack.ts

Acceptance Criteria:

• The AppStack CDK stack deploys successfully.
• The Cognito User Pool is created and allows user sign-up and sign-in.
• The REST API routes requests to the ALB, and the WebSocket API manages connections.
• Connecting to the WebSocket API successfully triggers the $connect Lambda, which writes the connection ID to the correct key in Redis.
• Disconnecting from the WebSocket API triggers the $disconnect Lambda, which removes the connection ID from Redis.
• All specified CloudWatch alarms are created and configured.

Dependencies: Ticket 1, Ticket 2, Ticket 3

---

Ticket 1.5: Implement CI/CD Pipeline Workflows

Objective: Create the GitHub Actions workflow files to automate infrastructure and application linting, testing, and deployment.

Context: This ticket establishes the automated CI/CD pipeline specified in the Deployment Architecture. This pipeline is crucial for ensuring code quality and enabling repeatable, safe deployments to both staging and production environments, with a manual gate for production releases.

Technical Directives:

• Create a GitHub Actions workflow file at .github/workflows/pr.yml.
• Configure the pr.yml workflow to trigger on every pull request to the main branch.
• The pr.yml workflow must execute linting, type-checking, and a cdk diff command to preview infrastructure changes.
• Create a GitHub Actions workflow file at .github/workflows/deploy.yml.
• Configure the deploy.yml workflow to trigger on every merge to the main branch.
• The deploy.yml workflow must build placeholder Docker images for all services, push them to the ECR repositories tagged with the Git SHA, and deploy all CDK stacks to the staging environment.
• After the staging deployment, the workflow must include a placeholder for integration tests.
• After the tests, the workflow must use a GitHub Actions environment to require manual approval before proceeding.
• Upon approval, the workflow must deploy all CDK stacks to the production environment.

Scope Boundaries:

• GitHub Actions configuration files: .github/workflows/

Acceptance Criteria:

• Creating a pull request against the main branch successfully triggers the pr.yml workflow, and the cdk diff step completes.
• Merging a pull request to the main branch successfully triggers the deploy.yml workflow.
• The deploy.yml workflow successfully pushes placeholder images to all ECR repositories.
• The deploy.yml workflow successfully completes the cdk deploy --all step for the staging environment.
• The deploy.yml workflow correctly pauses and waits for manual approval before the production deployment step.

Dependencies: Ticket 1, Ticket 2, Ticket 3, Ticket 4

Phase 2: Database Schema and Shared Data Layer

Ticket 2.1: Create Database Schema with Alembic Migration

Objective: This ticket creates the initial Alembic migration file that defines all required database tables, columns, constraints, and indexes for the entire application.

Context: This foundational schema supports all core entities described in the Architecture Design Document, including multi-tenancy, session management, message history, and the dual-layer RAG storage (ethnographic_chunks, product_context_chunks). It also includes the langgraph_thread_tenant_map table required for the GDPR erasure workflow to correctly identify and purge LangGraph checkpoint data from RDS.

Technical Directives:

• Create a new Alembic migration script.
• The script must define all tables and columns exactly as specified in the Phase 2: Database Schema contract, including tenants, workspaces, users, workspace_members, sessions, persona_configs, messages, analyst_reports, ethnographic_chunks, product_context_chunks, gdpr_audit_log, and langgraph_thread_tenant_map.
• All primary keys, foreign keys, data types (including ENUMs and vector(1536)), NOT NULL constraints, and default values must match the contract.
• The script must create HNSW indexes on the embedding columns of ethnographic_chunks and product_context_chunks using cosine distance.
• The script must create all other specified b-tree indexes on foreign keys and commonly queried columns.
• The migration must not create the AsyncPostgresSaver checkpoint tables; these are managed by the LangGraph library at runtime.

Scope Boundaries:

• Database migration script (backend/migrations/versions/)

Acceptance Criteria:

• The Alembic migration runs successfully from a clean state against the RDS PostgreSQL instance.
• All specified tables, columns, constraints, and indexes are present in the database after the migration completes.
• The data types for all columns, including ENUMs and vector dimensions, match the specification.
• The HNSW indexes are configured for cosine similarity search.

Dependencies: None

---

Ticket 2.2: Implement SQLAlchemy ORM Models

Objective: This ticket creates the SQLAlchemy ORM model classes that map directly to the database schema established in the previous ticket.

Context: These ORM models provide a type-safe, object-oriented interface to the database for all backend services (FastAPI control plane, orchestration worker, ingestion worker). As per the Architecture Design Document, every model must support tenant isolation by including a tenant_id attribute, which is fundamental to the system's security architecture.

Technical Directives:

• Create SQLAlchemy declarative model classes for every table defined in the 001_initial_schema.py migration.
• The class names must be Tenant, Workspace, User, WorkspaceMember, Session, PersonaConfig, Message, AnalystReport, EthnographicChunk, ProductContextChunk, GdprAuditLog, and LangGraphThreadTenantMap.
• All model attributes must correctly map to their corresponding database columns, including data types and relationships.
• All models intended for tenant-scoped data must include the tenant_id attribute.

Scope Boundaries:

• Shared ORM models module (backend/shared/models.py)

Acceptance Criteria:

• The backend/shared/models.py module can be imported without errors.
• Instantiating a model object and accessing its attributes does not raise an error.
• When used with a SQLAlchemy session, the models can be used to successfully insert and retrieve data from their corresponding tables.

Dependencies: Ticket 1

---

Ticket 2.3: Define Pydantic API and Message Schemas

Objective: This ticket defines the Pydantic models that serve as the data transfer objects (DTOs) for API requests, responses, and internal WebSocket messages.

Context: These schemas establish the formal data contracts between the frontend client, the API layer, and the orchestration workers, as outlined in the Architecture Design Document. Defining these upfront ensures consistency and provides a clear handoff to the Following Phase (Phase 3), which will build the API endpoints that consume and produce these shapes.

Technical Directives:

• Implement all Pydantic V2 models as specified in the Phase 2: Contracts section.
• Create the SessionStatus and AgentRole string enums.
• Implement the PersonaConfigCreate, SessionCreate, MessageOut, SessionOut, and WorkspaceMessagePush models with the exact field names and types specified.
• Use Python's UUID and datetime types where appropriate.
• Ensure the personas list in SessionCreate is correctly typed as list[PersonaConfigCreate].
• The WorkspaceMessagePush schema must accommodate different event types and optional payloads as defined in the contract.

Scope Boundaries:

• Shared Pydantic schemas module (backend/shared/schemas.py)

Acceptance Criteria:

• The backend/shared/schemas.py module can be imported without errors.
• Pydantic models can be instantiated with valid data and raise validation errors for invalid data (e.g., wrong types, missing required fields).
• The SessionCreate model correctly validates that the length of the personas list matches the persona_count value in consuming code.
• The WorkspaceMessagePush schema can be used to serialize a payload for the agent_message event, which will be used by the orchestration worker in a later phase.

Dependencies: None

---

Ticket 2.4: Create Shared Database Connection Module

Objective: This ticket creates a reusable Python module for managing the asynchronous database connection pool and sessions.

Context: This module centralizes database connection logic, ensuring that all backend services connect to RDS PostgreSQL consistently. It provides the foundation for application-level Row-Level Security by including the function to set the tenant context on a per-session basis, a key security control from the Architecture Design Document.

Technical Directives:

• Implement the backend/shared/db.py module with the specified asynchronous functions: init_db, get_db_session, set_tenant_context, and close_db.
• init_db must create a module-level async SQLAlchemy engine and sessionmaker from a provided database URL and register the pgvector extension.
• get_db_session must be an async generator that yields a new AsyncSession.
• set_tenant_context must execute the SET LOCAL app.current_tenant_id command on a given session.
• close_db must properly dispose of the engine resources.

Scope Boundaries:

• Shared database connection module (backend/shared/db.py)

Acceptance Criteria:

• Calling init_db successfully establishes a connection pool to the RDS instance.
• get_db_session yields a functional AsyncSession object that can be used to execute queries.
• set_tenant_context successfully sets the session variable in PostgreSQL, verifiable by executing SHOW app.current_tenant_id within the same transaction.
• Calling close_db terminates the connection pool gracefully.

Dependencies: Ticket 1

---

Ticket 2.5: Implement Row-Level Security Policies and Tenant Context Management

Objective: This ticket applies Row-Level Security (RLS) policies to all tenant-scoped tables in the database to enforce data isolation at the database layer.

Context: RLS is a defense-in-depth security measure mandated by the Architecture Design Document to prevent cross-tenant data leakage, even in the event of an application-layer bug. This ticket implements the PostgreSQL RLS policies and integrates them with the application's tenant context management.

Technical Directives:

• Augment the Alembic migration from Ticket 1 to enable Row-Level Security on the specified tables: sessions, messages, persona_configs, analyst_reports, product_context_chunks, workspace_members, workspaces, and users.
• For each table, create a policy that permits access only when the row's tenant_id column matches the value of the app.current_tenant_id session setting.
• The set_tenant_context function in backend/shared/db.py will be used by the application layer in a later phase to set this session variable for each request.

Scope Boundaries:

• Database migration script (backend/migrations/versions/)
• Shared database connection module (backend/shared/db.py)

Acceptance Criteria:

• After running the migration, RLS is enabled on all specified tables.
• A test connecting to the database confirms that a SELECT query on a protected table returns no rows if app.current_tenant_id is not set.
• After setting app.current_tenant_id to Tenant A's UUID, a SELECT query returns only rows belonging to Tenant A and excludes rows from Tenant B.
• The set_tenant_context function correctly sets the required session variable for the RLS policies to function.

Dependencies: Ticket 1, Ticket 4

---

Ticket 2.6: Create Shared Redis Client Module

Objective: This ticket creates a reusable Python module for managing the asynchronous connection to the ElastiCache Redis cluster.

Context: This module provides a centralized client for all Redis interactions, as specified in the Architecture Design Document. It will be used by the API layer to manage WebSocket connection IDs and by the orchestration worker for session heartbeating and retrieving connection IDs for pushing messages.

Technical Directives:

• Implement the backend/shared/redis_client.py module with the specified asynchronous functions: init_redis, get_redis, and close_redis.
• init_redis must create a module-level aioredis connection pool from a provided Redis URL.
• get_redis must return a Redis client instance from the pool.
• close_redis must properly close the connection pool.
• The module must define the exact string constants for Redis keys as specified in the contract: WS_CONNECTIONS_KEY, HEARTBEAT_KEY, and MANAGEMENT_ENDPOINT_KEY.

Scope Boundaries:

• Shared Redis client module (backend/shared/redis_client.py)

Acceptance Criteria:

• Calling init_redis successfully connects to the ElastiCache Redis instance.
• The client returned by get_redis can successfully execute Redis commands (e.g., SET, GET, SADD).
• The defined key constants (WS_CONNECTIONS_KEY, etc.) are importable and correctly formatted for use with string substitution.
• Calling close_redis closes the connection pool without errors.

Dependencies: None

---

Ticket 2.7: Define Shared Custom Error Classes

Objective: This ticket establishes a standardized set of custom exception classes for handling application-specific errors consistently across all backend services.

Context: As per the Error Handling Policy in the Architecture Design Document, the system must return generic error messages to clients while logging detailed internal information. These custom exceptions provide the mechanism for services to signal specific failure modes (e.g., authorization failure, invalid state transition) which can then be caught by a global exception handler in the API layer.

Technical Directives:

• Create the backend/shared/errors.py module.
• Define the base AppError class which accepts a message and a correlation ID.
• Define the specific exception classes that inherit from AppError: AuthorizationError, ValidationError, SessionStateError, IngestionError, CheckpointError, LLMError, and CapacityError.

Scope Boundaries:

• Shared errors module (backend/shared/errors.py)

Acceptance Criteria:

• The backend/shared/errors.py module can be imported without errors.
• Each custom exception class can be raised and caught correctly.
• An instance of AppError or its subclasses correctly stores the message and correlation ID passed to its constructor.

Dependencies: None

---

Ticket 2.8: Implement Shared Text Sanitization Utility

Objective: This ticket creates a reusable function for sanitizing user-provided text inputs to mitigate security risks like prompt injection.

Context: The Architecture Design Document's Input Validation section requires that all user-supplied text be sanitized before being used in LLM prompts or stored in the database. This centralized utility ensures that sanitization is applied consistently by all services that process user input.

Technical Directives:

• Create the backend/shared/sanitize.py module.
• Implement the function sanitize_user_text which accepts a string and a maximum length integer.
• The function must remove Unicode control characters in the Cc and Cf categories, while preserving essential whitespace like tabs and newlines.
• The function must truncate the input string to the specified maximum length.

Scope Boundaries:

• Shared sanitization module (backend/shared/sanitize.py)

Acceptance Criteria:

• The sanitize_user_text function returns a string with control characters (e.g., \x00, \x08) removed.
• The function correctly truncates a string that is longer than the max_length argument.
• The function preserves standard characters, numbers, punctuation, and essential whitespace.
• A string shorter than max_length with no control characters is returned unchanged.

Dependencies: None

Phase 3: Authentication and Tenant Management

Ticket 3.1: FastAPI Application Entry Point and Core Middleware

Objective: Establish the main FastAPI application shell with essential middleware for security, error handling, and lifecycle management.

Context: This ticket creates the foundational structure of the control_plane service as described in the Architecture Design Document. It implements the application entry point, including startup and shutdown events, global exception handling, and security headers, which are required before any API routes can be added.

Technical Directives:

• Implement a FastAPI application in backend/control_plane/main.py.
• Create a lifespan context manager that calls init_db() and init_redis() from the shared modules on startup, and close_db() and close_redis() on shutdown.
• Configure CORS middleware to allow only the origin specified in an environment variable.
• Implement middleware to add the following security headers to all responses: Strict-Transport-Security, X-Content-Type-Options, X-Frame-Options, and Content-Security-Policy.
• Implement a global exception handler that catches all AppError subclasses defined in backend/shared/errors.py.
• The exception handler must log the full error details and return a generic JSON error response to the client containing a correlation ID, without exposing internal details like stack traces.
• Enable AWS X-Ray middleware for distributed tracing.

Scope Boundaries:

• FastAPI application entry point (main.py)
• Core application middleware configuration

Acceptance Criteria:

• The application starts without errors, successfully initializing database and Redis connections.
• A request from an allowed origin receives a response with the correct CORS headers.
• A request from a disallowed origin is blocked by CORS policy.
• All API responses include the specified security headers (HSTS, X-Content-Type-Options, etc.).
• When a route handler raises a custom AppError, the client receives a JSON response with a generic message and a correlation ID, and the full error is logged.

Dependencies: None

Ticket 3.2: Cognito JWT Verification Middleware

Objective: Implement middleware to validate Cognito JWTs and extract user claims for authenticated requests.

Context: This ticket provides the core authentication mechanism for the entire API, as outlined in the Architecture Design Document. It creates a reusable FastAPI dependency that all protected endpoints will use to ensure requests are from authenticated users and to establish tenant context for database queries.

Technical Directives:

• In backend/control_plane/auth/middleware.py, create the verify_cognito_token function as specified in the phase contract.
• The function must fetch the JSON Web Key Set (JWKS) from the Cognito User Pool endpoint.
• The JWKS response must be cached in memory with a time-to-live (TTL) to avoid excessive requests.
• The function must validate the JWT's signature, expiry, issuer, and audience against the configured Cognito User Pool details.
• On successful validation, the function must return a TokenClaims object populated with claims from the token.
• If validation fails for any reason, the function must raise an AuthorizationError.
• Create a FastAPI dependency (get_current_user) that injects the get_db_session dependency, retrieves and validates the token via verify_cognito_token, and executes set_tenant_context(db, tenant_id) on the yielded AsyncSession before returning the user claims.

Scope Boundaries:

• Authentication middleware module

Acceptance Criteria:

• A request to an endpoint using the dependency with a valid JWT succeeds.
• A request with an expired JWT is rejected with an AuthorizationError.
• A request with an invalid signature is rejected with an AuthorizationError.
• A request with a missing Authorization header is rejected with an AuthorizationError.
• On a successful authenticated request, the app.current_tenant_id session variable is correctly set in the database connection before the route handler executes.

Dependencies: 1

Ticket 3.3: Cognito Post-Confirmation Lambda Function

Objective: Create a Lambda function to provision a new user, tenant, and workspace in the database when a user confirms their account in Cognito.

Context: This function automates the user onboarding process, linking the Cognito identity to the application's internal data model as described in the ADD. It ensures that every new user who signs up is correctly provisioned with their own tenant and default workspace, enabling the bootstrap flow.

Technical Directives:

• Create a Lambda function defined in backend/lambdas/cognito_post_confirmation/handler.py.
• The handler must parse the user's sub, email, and custom attributes from the Cognito PostConfirmation trigger event.
• If the custom:tenant_id attribute is absent, the function must create a new Tenant record in the database, using the custom:gdpr_region attribute.
• The function must create a User record in the database, linking the Cognito sub to the newly created or existing tenant ID.
• The function must create a default Workspace for the new tenant.
• The function must create a WorkspaceMember record, making the new user an owner of the default workspace.
• The function must use a direct psycopg2 connection, not the shared SQLAlchemy engine, as specified in the contract.
• The handler must return the event object unchanged to satisfy Cognito's trigger requirements.
• The Lambda function must be explicitly configured (via CDK in the AppStack) to run inside the VPC's private-app subnets and attached to a security group that allows outbound traffic to the RDS instance's security group, otherwise it will silently time out connecting to the database.

Scope Boundaries:

• Cognito Lambda function module

Acceptance Criteria:

• When a new user confirms their email in Cognito, the Lambda is triggered.
• A Tenant, User, Workspace, and WorkspaceMember record are successfully created in the RDS database.
• The User.id in the database matches the Cognito user's sub.
• The Tenant.id is correctly associated with the new User.
• The Lambda function completes successfully and does not cause the Cognito confirmation flow to fail.

Dependencies: None

Ticket 3.4: Tenant Management API Router

Objective: Implement the API endpoints for creating and retrieving tenant information.

Context: This ticket builds the first set of business logic endpoints, allowing new users to bootstrap their accounts and retrieve their tenant details. This is a foundational part of the multi-tenant architecture and depends on the JWT middleware to identify the user.

Technical Directives:

• Create the tenant router in backend/control_plane/routers/tenants.py.
• Implement the POST /tenants endpoint, which is only callable by a user who does not yet have a tenant_id claim.
• The POST /tenants endpoint must create a Tenant, a default Workspace, and a WorkspaceMember record making the caller the owner. It should return a TenantOut schema.
• Implement the GET /tenants/me endpoint.
• This endpoint must use the get_current_user dependency to get the authenticated user's claims.
• It must query and return the Tenant record corresponding to the tenant_id from the user's JWT claims.

Scope Boundaries:

• FastAPI router module (tenants.py)

Acceptance Criteria:

• A POST request to /tenants with valid input creates the tenant, workspace, and workspace member records in the database and returns a 200 response with the tenant data.
• A GET request to /tenants/me by an authenticated user returns the correct tenant information for that user.
• An authenticated user cannot call POST /tenants a second time.
• A request to /tenants/me without a valid JWT token fails with an authentication error.

Dependencies: 2

Ticket 3.5: User Profile API Router

Objective: Implement API endpoints for retrieving and updating the current user's profile.

Context: This provides the basic user profile management functionality. It allows users to view their own information and update mutable fields like their display name, which is a standard feature for any multi-user application.

Technical Directives:

• Create the user router in backend/control_plane/routers/users.py.
• Both endpoints must use the get_current_user dependency for authentication and to identify the user.
• Implement the GET /users/me endpoint to query and return the User record matching the user_id from the JWT claims.
• Implement the PATCH /users/me endpoint to update the display_name of the current user.

Scope Boundaries:

• FastAPI router module (users.py)

Acceptance Criteria:

• A GET request to /users/me by an authenticated user returns their own user profile data.
• A PATCH request to /users/me with a new display_name updates the corresponding record in the database and returns the updated user profile.
• A user cannot view or modify another user's profile through these endpoints.
• A request to either endpoint without a valid JWT fails with an authentication error.

Dependencies: 2

Ticket 3.6: Workspace Authorization Logic

Objective: Create a reusable authorization function to enforce role-based access control within workspaces.

Context: This ticket implements the core authorization logic that determines if a user has sufficient permissions to perform an action on a workspace. This function will be used as a dependency in the workspace router to protect endpoints according to the business rules defined in the ADD.

Technical Directives:

• In backend/control_plane/auth/authorization.py, implement the require_workspace_role function as specified in the phase contract.
• The function must accept a workspace_id, user_id, minimum_role, and a database session.
• It must query the workspace_members table to find the user's role for the given workspace.
• It must raise an AuthorizationError if the user is not a member of the workspace or if their role is lower than the minimum_role.
• The role hierarchy is viewer < editor < owner.

Scope Boundaries:

• Authorization logic module

Acceptance Criteria:

• The function does not raise an error if the user's role meets or exceeds the minimum requirement.
• The function raises an AuthorizationError if the user's role is below the minimum requirement (e.g., a viewer attempting an editor action).
• The function raises an AuthorizationError if the user is not a member of the specified workspace.

Dependencies: 2

Ticket 3.7: Workspace Management API Router

Objective: Implement the API endpoints for creating, listing, and retrieving workspaces.

Context: This ticket delivers the core functionality for managing workspaces, which are the primary containers for sessions. It uses the authentication middleware from Ticket 2 and the authorization logic from Ticket 6 to ensure all operations are secure and correctly scoped to the user's tenant and permissions.

Technical Directives:

• Create the workspace router in backend/control_plane/routers/workspaces.py.
• Implement GET /workspaces to list all workspaces within the user's tenant.
• Implement POST /workspaces to create a new workspace within the user's tenant. The user who creates the workspace must be made its owner.
• Implement GET /workspaces/{workspace_id} to retrieve a single workspace by its ID.
• The GET /workspaces/{workspace_id} endpoint must use the require_workspace_role dependency to ensure the user has at least viewer permissions.
• All endpoints must use the get_current_user dependency.

Scope Boundaries:

• FastAPI router module (workspaces.py)

Acceptance Criteria:

• A GET request to /workspaces returns a list of workspaces belonging only to the user's tenant.
• A POST request to /workspaces creates a new Workspace and WorkspaceMember record and returns the new workspace data.
• A GET request to /workspaces/{workspace_id} by a user with viewer rights succeeds.
• A GET request to /workspaces/{workspace_id} by a user who is not a member of that workspace is rejected with an AuthorizationError.
• A request to any endpoint that attempts to access a workspace in another tenant fails.

Dependencies: 2, 6

Ticket 3.8: SES Email Helper Module

Objective: Create a shared module for sending transactional emails via Amazon SES.

Context: This ticket centralizes the logic for sending emails, as required by various parts of the application. The first consumer will be the workspace member invitation flow, but this module is designed to be reused for other notifications as specified in the ADD.

Technical Directives:

• Create the email helper module in backend/shared/email.py.
• Implement the send_workspace_invitation function.
• This function must construct and send an email using the AWS SES API.
• If the SES API call fails, the function must log the error and enqueue a message to a dead-letter queue for later inspection/retry. It must not raise an exception that would fail the parent API request.
• Use the SES_SENDER_ADDRESS environment variable for the "from" address.

Scope Boundaries:

• Shared email module

Acceptance Criteria:

• Calling send_workspace_invitation with a valid recipient address results in an email being sent via SES.
• If the SES service is unavailable (simulated via mocking), the function logs an error, enqueues a message, and returns successfully without raising an exception.

Dependencies: None

Ticket 3.9: Workspace Member Management Endpoints

Objective: Implement API endpoints for inviting and removing workspace members.

Context: This ticket completes the workspace management functionality by allowing workspace owners to manage their team's access. It integrates the authorization logic and the SES email helper to create a secure and user-friendly invitation flow.

Technical Directives:

• Add new endpoints to the workspace router in backend/control_plane/routers/workspaces.py.
• Implement POST /workspaces/{workspace_id}/members.
• This endpoint must use the require_workspace_role dependency to ensure the caller is an owner.
• It must create a WorkspaceMember record for the invited user.
• It must call the send_workspace_invitation function from the email helper module.
• Implement DELETE /workspaces/{workspace_id}/members/{user_id}.
• This endpoint must use the require_workspace_role dependency to ensure the caller is an owner.
• It must delete the specified WorkspaceMember record.

Scope Boundaries:

• FastAPI router module (workspaces.py)

Acceptance Criteria:

• A POST request to /workspaces/{workspace_id}/members by a workspace owner creates a WorkspaceMember record and sends an invitation email.
• An attempt to invite a member by a non-owner is rejected with an AuthorizationError.
• A DELETE request to /workspaces/{workspace_id}/members/{user_id} by an owner removes the corresponding WorkspaceMember record.
• An attempt to remove a member by a non-owner is rejected with an AuthorizationError.

Dependencies: 7, 8

Phase 4: Document Ingestion Pipeline

Phase 5: Session Management and Worker Lifecycle

Phase 6: LangGraph Orchestration Worker

Phase 7: Frontend Application

Phase 8: GDPR Erasure Workflow and Data Lifecycle

Phase 9: Observability, Hardening, and Production Readiness