diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0c34fbf4..2db7b230 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -68,6 +68,11 @@ jobs: # Run tests for affected projects run: bun nx affected --target=test --parallel=3 + - name: API End-to-End + # Boots the real API (startApi) in local mode on an embedded MongoDB and + # drives it over HTTP; catches wiring a mocked unit test cannot. + run: bun nx affected --target=test:e2e --parallel=1 + - name: Build # Run build for affected projects run: bun nx affected --target=build --parallel=3 diff --git a/CLAUDE.md b/CLAUDE.md index 6aa78bd2..c69a78ae 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -15,20 +15,25 @@ Key features: - **Thinking Canvas** — visual problem-solving with Seed (problem), Soil (constraints), and Root (assumptions) nodes - **Socratic Conversations** — text-based dialogue using 6 techniques: elenchus, dialectic, maieutics, definitional, analogical, counterfactual - **Node Dialogues** — per-node Socratic chat within canvases (technique auto-selected by node type) -- **Knowledge Gap Detection** (RFC-0001, in progress) — Bayesian Knowledge Tracing to detect when Socratic questioning fails -- **Adaptive Scaffolding** (RFC-0002, in progress) — 5-level scaffold system that never gives direct answers +- **Knowledge Gap Detection** (RFC-0001) — Bayesian Knowledge Tracing to detect when Socratic questioning fails; runs on every chat, canvas and map turn +- **Adaptive Scaffolding** (RFC-0002) — 5-level scaffold system that never gives direct answers; runs on every turn +- **Thought Maps** (RFC-0003) — free-form maps with per-node chat, suggested branches, and extraction from a conversation ## Repo Structure ``` -├── packages/mukti-api/ # @mukti/api — NestJS 11 backend -├── packages/mukti-web/ # @mukti/web — Next.js 15 frontend -├── docs/reference/ # Technical documentation assets -├── mukti-mcp-server/ # Standalone MCP server (not in Nx workspace) -├── docs/rfcs/ # Design documents (RFC-0001, RFC-0002) -└── scripts/ # AI commit, deploy scripts +├── packages/mukti-api/ # @mukti/api — NestJS 11 backend (npm: @muktiai/api) +├── packages/mukti-web/ # @mukti/web — Next.js 15 frontend (npm: @muktiai/web) +├── packages/mukti-cli/ # `npx muktiai` local launcher (npm: muktiai) +├── packages/mukti-contracts/ # @muktiai/contracts — private, type-only web/API wire contracts +├── packages/mukti-mcp/ # Standalone MCP server (Bun workspace, not an Nx project) +├── docs/reference/ # Architecture overview and technical docs +├── docs/rfcs/ # Design documents (RFC-0001 to RFC-0007) +└── scripts/ # AI commit, deploy, release, local start scripts ``` +Architecture overview (current): `docs/reference/architecture/overview.md`. + ## Package Manager & Runtime Use **Bun** as the primary package manager and runtime: @@ -54,7 +59,7 @@ bun run affected:lint / affected:test / affected:build ## Infrastructure -Docker Compose provides: **MongoDB 7** (port 27017, db: `mukti`), **Redis 7** (port 6379). API runs on port 3000, Web on port 3001. +Docker Compose provides: **MongoDB 7** (port 27017, db: `mukti`), **Redis 7** (port 6379, BullMQ queues and the cross-instance SSE stream bus). API runs on port 3000, Web on port 3001. Local mode (`MUKTI_LOCAL=1`) needs neither: embedded MongoDB, inline jobs, in-process streams. Key env vars: `JWT_SECRET`, `JWT_REFRESH_SECRET`, `OPENROUTER_API_KEY`, `SESSION_SECRET`, `CORS_ORIGINS`, `COOKIE_DOMAIN` (defaults to `mukti.chat` in production). @@ -78,21 +83,35 @@ Key env vars: `JWT_SECRET`, `JWT_REFRESH_SECRET`, `OPENROUTER_API_KEY`, `SESSION ``` AppModule -├── DatabaseModule — MongoDB connection, registers ALL_SCHEMAS from src/schemas/index.ts -├── AuthModule — JWT + Google OAuth, email verification, refresh token rotation -├── AiModule — OpenRouter/Gemini clients, BYOK key encryption, model resolution -├── CanvasModule — Thinking Canvas CRUD (Seed/Soil/Root/Insight nodes, relationships) -├── ConversationsModule — Text-based Socratic sessions, BullMQ queue, SSE streaming -├── DialogueModule — Per-node canvas dialogue, BullMQ queue, SSE streaming -├── KnowledgeTracingModule — Bayesian Knowledge Tracing (RFC-0001, partially implemented) -├── HealthModule — Health check endpoint -└── WaitlistModule — Pre-launch waitlist +├── CoreModule — MongoDB (ALL_SCHEMAS), BullMQ (QueueModule), StreamBus (Redis pub/sub or in-process) +├── AuthModule — JWT + Google OAuth, email verification, refresh tokens +├── AiModule — provider clients, BYOK key encryption, model policy, AiRequestGate, free quota +├── CanvasModule — Thinking Canvas CRUD (Seed/Soil/Root/Insight nodes, relationships) +├── ConversationsModule — Text-based Socratic sessions, BullMQ queue, SSE streaming +├── DialogueModule — Per-node canvas dialogue, BullMQ queue, SSE streaming +├── ThoughtMapModule — Thought maps, node dialogue, branch suggestions, conversation extraction +├── KnowledgeTracingModule — BKT; KnowledgeStateStore is the only writer of knowledge_states +├── LocalModeModule — MUKTI_LOCAL runtime wiring +├── HealthModule — Health check endpoint +└── WaitlistModule — Pre-launch waitlist + +Imported by the surfaces (not in AppModule directly): +SocraticTurnModule → ScaffoldingModule → KnowledgeTracingModule, DialogueQualityModule ``` +### Architecture Rules + +- **Every AI request passes `AiRequestGate.authorize()`** before it is enqueued. It resolves key, model and plan, and consumes a free message on the server key. New AI endpoints must call it. +- **Every Socratic turn runs `SocraticTurnEngine`** (`prepare` → model call → `inspectReply` → `settle`). Do not reimplement gap detection, quality checks or scaffold fade in a processor. +- **Only `KnowledgeStateStore` writes `knowledge_states`** (compare-and-set on `attempts`). Scaffold writes pass the `guard` from `settle` so concurrent turns cannot overwrite each other. +- **Emit SSE events through the stream services**, never to a connection directly; they publish via `StreamBus` so jobs on any instance reach the client. Each stream has its own Redis channel, and a stream service must call `bus.track` on its first local connection for a key and `bus.untrack` when the last one goes, or that instance never receives the stream. +- **Read another domain's data through its service**, not its model (e.g. thought maps use `ConversationService`, `CanvasService`). +- **SSE event shapes live in `@muktiai/contracts`** (`packages/mukti-contracts`, private, type-only). Import with `import type`. The web type-checks against it; change both sides together. + ### Key Patterns -- **Queue-based AI**: POST message → `{ jobId, position }` (202 Accepted) → SSE stream for real-time response. Both ConversationsModule and DialogueModule use this pattern. -- **SSE events**: `processing → message → complete | error`. Connections keyed by `sessionId:nodeId`. +- **Queue-based AI**: POST message → `{ jobId, position }` (202 Accepted) → SSE stream for real-time response. Chat, canvas node dialogue and thought-map node dialogue all use this pattern. +- **SSE events**: `processing → message → complete | error`. Chat streams are keyed by `conversationId`; node dialogue by `sessionId:nodeId` (thought maps use `map:{mapId}` as the session). - **BYOK**: Users store encrypted OpenRouter/Gemini API keys. `AiPolicyService.resolveEffectiveModel()` resolves which key/model to use. Default model: `openai/gpt-5-mini`. - **Canvas node IDs**: `seed` (singleton), `soil-{index}`, `root-{index}`, `insight-{index}`. Services parse these to determine node type. - **Dialogue technique mapping**: seed → maieutics, root → elenchus, soil → counterfactual, insight → dialectic. @@ -116,7 +135,7 @@ All schemas registered via `ALL_SCHEMAS` array in `src/schemas/index.ts`. ### Auth Flow -Email/password + Google OAuth. Access tokens in memory, refresh tokens in httpOnly cookies + DB. Refresh token rotation on use. Email verification on signup. +Email/password + Google OAuth. Access tokens in memory, refresh tokens in httpOnly cookies + DB. Refreshing returns a new access token; the refresh token itself is not rotated yet. Email verification on signup. Guards: `JwtAuthGuard` (global), `EmailVerifiedGuard`, `LoginRateLimitGuard`, `RolesGuard` Decorators: `@CurrentUser()`, `@Public()`, `@Roles()` @@ -133,26 +152,25 @@ Roles: `user`, `moderator`, `admin` ``` app/ ├── (auth)/auth/ # Login, forgot/reset password, verify email, OAuth callbacks -├── dashboard/ # Authenticated app (catch-all [[...slug]]) -│ ├── canvas/[id]/ # Thinking Canvas session -│ ├── conversations/[id]/ # Text conversation -│ ├── community/ # Technique community -│ ├── settings/ # User settings -│ └── ... # help, messages, reports, resources, security, sessions -├── canvas/[id]/ # Public canvas viewer -├── chat/[id]/ # Public chat viewer -└── api/waitlist/ # Next.js API route +├── chat/ # Text conversations +├── canvas/ # Thinking Canvas sessions +├── maps/ # Thought maps (list, new, [id]) +├── map/[token]/ # Public read-only shared thought map +├── settings/ # User settings (model, API key) +└── dashboard/[[...slug]] # Redirects old /dashboard/* links ``` ### State Management (Zustand Stores) -| Store | Purpose | -| -------------- | ---------------------------------------------------------------------------- | -| `auth-store` | User (persisted to localStorage), accessToken (memory-only), isAuthenticated | -| `canvas-store` | React Flow nodes/edges, **optimistic updates with rollback** on API failure | -| `ai-store` | Active model, AI settings | -| `chat-store` | Chat/conversation state | -| `wizard-store` | Canvas setup wizard state | +| Store | Purpose | +| ------------------- | ---------------------------------------------------------------------------- | +| `auth-store` | User (persisted to localStorage), accessToken (memory-only), isAuthenticated | +| `canvas-store` | React Flow nodes/edges, **optimistic updates with rollback** on API failure | +| `ai-store` | Active model, AI settings | +| `chat-store` | Chat/conversation state | +| `wizard-store` | Canvas setup wizard state | +| `thought-map-store` | Thought map nodes/edges and selection | +| `onboarding-store` | First-run local onboarding state | **Auth pattern:** Access token in memory only (cleared on reload). User object in localStorage. On hydration, `useAuth` hook restores session via refresh token cookie. @@ -177,7 +195,7 @@ app/ ## MCP Server -**Location:** `mukti-mcp-server/` (standalone, not in Nx workspace) +**Location:** `packages/mukti-mcp/` (Bun workspace package, not an Nx project, so CI and releases skip it) Exposes Socratic reasoning as MCP tools via stdio transport: diff --git a/README.md b/README.md index 561cb907..01770884 100644 --- a/README.md +++ b/README.md @@ -25,7 +25,7 @@ Development | Architecture | API | - MCP Server + MCP Server

@@ -222,7 +222,7 @@ Mukti uses dialogue to push thinking forward without taking control of your work ## MCP Server Mukti's Socratic reasoning is also available as MCP tools (`socratic_inquiry`, `explore_paths`, -`explain_approach`) for any MCP-compatible client. See [mukti-mcp-server](mukti-mcp-server/README.md). +`explain_approach`) for any MCP-compatible client. See [packages/mukti-mcp](packages/mukti-mcp/README.md). ## Repo Structure @@ -231,10 +231,12 @@ Mukti's Socratic reasoning is also available as MCP tools (`socratic_inquiry`, ` ├── packages/mukti-cli # `npx muktiai` launcher ├── packages/mukti-web # Next.js frontend ├── packages/mukti-api # NestJS backend -├── mukti-mcp-server # MCP server (standalone) +├── packages/mukti-mcp # MCP server (standalone) └── docs/ # Local mode, RFCs, and technical docs ``` +How the pieces fit together: [docs/reference/architecture/overview.md](docs/reference/architecture/overview.md). + ## Contributing Contributions are welcome. Start with [DEVELOPMENT.md](DEVELOPMENT.md) for setup, and diff --git a/bun.lock b/bun.lock index 1e07fc4f..c6eb4188 100644 --- a/bun.lock +++ b/bun.lock @@ -37,7 +37,7 @@ }, "packages/mukti-api": { "name": "@muktiai/api", - "version": "0.1.1-beta.1", + "version": "0.1.1", "dependencies": { "@google/generative-ai": "^0.24.1", "@nestjs/bullmq": "^11.0.4", @@ -63,6 +63,7 @@ "csurf": "^1.11.0", "fast-check": "^4.3.0", "helmet": "^8.1.0", + "ioredis": "^5.8.2", "mongoose": "^8.19.1", "passport": "^0.7.0", "passport-google-oauth20": "^2.0.0", @@ -113,7 +114,7 @@ }, "packages/mukti-cli": { "name": "muktiai", - "version": "0.1.1-beta.0", + "version": "0.1.1", "bin": { "muktiai": "./dist/index.mjs", }, @@ -138,6 +139,10 @@ "typescript-eslint": "^8.20.0", }, }, + "packages/mukti-contracts": { + "name": "@muktiai/contracts", + "version": "0.0.0", + }, "packages/mukti-mcp": { "name": "@mukti/mcp", "version": "1.0.0", @@ -154,7 +159,7 @@ }, "packages/mukti-web": { "name": "@muktiai/web", - "version": "0.2.1-beta.1", + "version": "0.2.1", "dependencies": { "@hookform/resolvers": "^5.1.1", "@radix-ui/react-accordion": "^1.2.12", @@ -889,6 +894,8 @@ "@muktiai/api": ["@muktiai/api@workspace:packages/mukti-api"], + "@muktiai/contracts": ["@muktiai/contracts@workspace:packages/mukti-contracts"], + "@muktiai/web": ["@muktiai/web@workspace:packages/mukti-web"], "@napi-rs/nice": ["@napi-rs/nice@1.1.1", "", { "optionalDependencies": { "@napi-rs/nice-android-arm-eabi": "1.1.1", "@napi-rs/nice-android-arm64": "1.1.1", "@napi-rs/nice-darwin-arm64": "1.1.1", "@napi-rs/nice-darwin-x64": "1.1.1", "@napi-rs/nice-freebsd-x64": "1.1.1", "@napi-rs/nice-linux-arm-gnueabihf": "1.1.1", "@napi-rs/nice-linux-arm64-gnu": "1.1.1", "@napi-rs/nice-linux-arm64-musl": "1.1.1", "@napi-rs/nice-linux-ppc64-gnu": "1.1.1", "@napi-rs/nice-linux-riscv64-gnu": "1.1.1", "@napi-rs/nice-linux-s390x-gnu": "1.1.1", "@napi-rs/nice-linux-x64-gnu": "1.1.1", "@napi-rs/nice-linux-x64-musl": "1.1.1", "@napi-rs/nice-openharmony-arm64": "1.1.1", "@napi-rs/nice-win32-arm64-msvc": "1.1.1", "@napi-rs/nice-win32-ia32-msvc": "1.1.1", "@napi-rs/nice-win32-x64-msvc": "1.1.1" } }, "sha512-xJIPs+bYuc9ASBl+cvGsKbGrJmS6fAKaSZCnT0lhahT5rhA2VVy9/EcIgd2JhtEuFOJNx7UHNn/qiTPTY4nrQw=="], diff --git a/docs/designs/rfc-0007-learner-readings-judge-port.md b/docs/designs/rfc-0007-learner-readings-judge-port.md index 521c2158..a78babd3 100644 --- a/docs/designs/rfc-0007-learner-readings-judge-port.md +++ b/docs/designs/rfc-0007-learner-readings-judge-port.md @@ -6,6 +6,24 @@ Repo: shettydev/mukti Status: APPROVED Mode: Builder +> **Code has moved since this was approved (2026-10-14 to 2026-10-18).** The file:line +> citations below describe the code as reviewed and are kept as the record. Before +> implementing, map them onto the current layout: +> +> - The turn block the plan moves into `TurnSettlementService` no longer lives in three +> processors. All three call `SocraticTurnEngine` (`modules/socratic-turn/`): +> `prepare` (gap detection, quality, scaffold context), `inspectReply`, and `settle` +> (grading, fade, knowledge write). `TurnSettlementService` should extend or replace +> `settle`, and the "all three processors" milestones now touch one call site each. +> - Every `KnowledgeState` write goes through `KnowledgeStateStore.applyObservation` +> (compare-and-set on `attempts`). The `KNOWLEDGE_WRITES` flag has one place to gate +> both writers (the turn and `POST /knowledge-tracing/update`). +> - Thought-map node chat now runs the RFC-0004 quality checks and reply monitor like +> the other two surfaces, which removes one of this plan's per-surface differences. +> - Scaffold-state writes are guarded on the state the turn read, and a failed +> knowledge write no longer fails the job (relevant to R2-24 and the M2 snapshot +> failure question). + ## Problem Statement Mukti reads each learner message with regex and substring lists, and the verdicts are @@ -868,7 +886,7 @@ B) Leave it writable and document the exception (human: ~10 min / CC: ~2 min) ❌ P1 becomes "nothing writes, except this route", which is easy to forget later C) Restrict it to admins and leave it writable (human: ~1 hour / CC: ~5 min) ✅ Closes the any-user write path while keeping an operator escape hatch -✅ Matches the admin-only pattern the cache-stats route in the same controller already uses +✅ Matches the admin-only pattern the struggling/mastered routes in the same controller already use ❌ The freeze still has a hole, and the flag no longer tells the whole story Net: make the freeze complete, or keep a manual write path and accept the exception. Header: Tracing endpoint @@ -878,7 +896,7 @@ While `KNOWLEDGE_WRITES=off`, `POST /knowledge-tracing/update` returns 409 with B) Leave writable, document The endpoint keeps writing as today. P1's text gains a note naming this route as the one exception. ✅ No behaviour change. ✅ Keeps a manual seeding path. ❌ The freeze has an exception that is easy to forget. C) Admin only, still writable -The endpoint requires the admin role and keeps writing regardless of the flag. ✅ Closes the any-user path. ✅ Matches the cache-stats route's pattern. ❌ The freeze still has a hole. +The endpoint requires the admin role and keeps writing regardless of the flag. ✅ Closes the any-user path. ✅ Matches the struggling/mastered routes' pattern. ❌ The freeze still has a hole. State: approved Actual answer: A) Reject with 409 (recommended), answer to D4 diff --git a/docs/reference/architecture/overview.md b/docs/reference/architecture/overview.md index b024bd0b..49f15e19 100644 --- a/docs/reference/architecture/overview.md +++ b/docs/reference/architecture/overview.md @@ -1,46 +1,132 @@ # Mukti Architecture -## Overview - -Mukti follows a modern cloud-native architecture built on NestJS, designed for scalability and performance. - -

- Mukti Architecture -

- -## Architecture Layers - -### Client Layer - -Web and mobile applications serve as entry points for end users, providing responsive interfaces across platforms. - -### CDN Layer - -CloudFlare CDN handles content delivery, caching static assets and reducing latency globally. - -### Backend Core - -NestJS powers the core API layer, orchestrating business logic and request handling. - -### External Services - -MongoDB serves as the primary operational database, and authentication is handled directly within NestJS (JWT + refresh tokens backed by MongoDB user records). - -### Application Layer - -The main application logic runs on NestJS deployed within Kubernetes clusters, ensuring high availability and horizontal scaling. - -### Infrastructure Services - -- **MongoDB**: Primary document database for persistent data storage -- **Redis**: In-memory cache for session management and performance optimization -- **Object Storage (S3)**: Handles file uploads and media storage - -### Third-Party Integrations - -- **Payment Gateway**: Processes transactions and payment operations -- **Email Service**: Manages transactional and notification emails - -## Data Flow - -Requests flow from clients through the CDN into the NestJS core, which performs authentication itself and coordinates with MongoDB for persistence, Redis for caching, and the Kubernetes-deployed application layer. External integrations handle specialized operations like payments and email delivery. +How Mukti is put together, as it is in the code. For feature status see the RFCs in +[`docs/rfcs`](../../rfcs/README.md); for how to run it see [`DEVELOPMENT.md`](../../../DEVELOPMENT.md) +and [`docs/local-mode.md`](../../local-mode.md). + +> The diagram image [`images/mukti-architecture.png`](./images/mukti-architecture.png) +> predates this page and shows an aspirational cloud stack (CDN, Kubernetes, S3, +> payments) that Mukti does not use. The Mermaid diagrams below are current. + +## Packages + +| Package | What it is | +| -------------------- | ---------------------------------------------------------------------------------------------- | +| `packages/mukti-api` | NestJS 11 API. All business logic, AI calls, queues and live streams. | +| `packages/mukti-web` | Next.js 15 app (App Router, TanStack Query, Zustand). Talks to the API over REST and SSE. | +| `packages/mukti-cli` | `npx muktiai` launcher. Starts the prebuilt API and web locally with an embedded MongoDB. | +| `packages/mukti-mcp` | Standalone MCP server over stdio. Shares the questioning-technique vocabulary, no shared code. | + +## Two ways it runs + +```mermaid +flowchart LR + subgraph Hosted["Hosted (Docker Compose / Dokploy)"] + W1[web] -->|REST + SSE| A1[api instance 1] + W1 -->|REST + SSE| A2[api instance N] + A1 & A2 --> M1[(MongoDB)] + A1 & A2 <--> R1[(Redis)] + A1 & A2 --> OR[OpenRouter / Gemini] + end + subgraph Local["Local (npx muktiai, MUKTI_LOCAL=1)"] + W2[web] -->|REST + SSE| A3[api, one process] + A3 --> M2[(embedded MongoDB)] + A3 --> CLI[Claude Code or Antigravity CLI] + end +``` + +- **Hosted.** MongoDB for data, Redis for BullMQ job queues and the cross-instance stream + bus. AI runs on OpenRouter with the server's key (counted against a daily free + allowance) or on the user's own encrypted key. Email is SendGrid. Auth is JWT access + tokens plus refresh tokens stored in MongoDB, with Google OAuth. +- **Local.** One process, no Redis: queue jobs run inline and stream events stay + in-process. Sign-in is skipped. AI runs on a CLI the user already has signed in. + +## Inside the API + +```mermaid +flowchart TD + subgraph Surfaces + Chat[ConversationsModule
chat] + Canvas[CanvasModule + DialogueModule
guided canvas, node chat] + Map[ThoughtMapModule
thought maps, node chat,
suggestions, extraction] + end + Gate[AiRequestGate
key, model, plan, free quota] + Turn[SocraticTurnEngine
one Socratic turn] + Scaf[ScaffoldingModule
gap detection, help level] + Qual[DialogueQualityModule
RFC-0004 guardrails] + KT[KnowledgeTracingModule
BKT, KnowledgeStateStore] + AI[AiModule
provider clients, keys, policy] + Core[CoreModule
MongoDB, BullMQ, StreamBus] + + Chat & Canvas & Map --> Gate --> AI + Chat & Canvas & Map --> Turn + Turn --> Scaf & Qual + Scaf --> KT + Map -->|reads via services| Chat & Canvas + Surfaces -.-> Core +``` + +The rules that hold this together: + +- **One gate before every AI request.** Controllers call `AiRequestGate.authorize()` + before enqueuing any job that calls a model. It resolves key ownership, plan and + model, and consumes a free message when the server's key will be used. +- **One Socratic turn.** Chat, canvas node chat and thought-map node chat all run + `SocraticTurnEngine`: `prepare` (gap detection, quality directives, scaffold context), + `inspectReply`, then `settle` (grade the reply, move the help level, update + knowledge). Each surface keeps only its own prompt, storage and streaming. +- **One writer for learner knowledge.** `KnowledgeStateStore.applyObservation` is the + only code that writes `knowledge_states`, as a compare-and-set on `attempts`. Scaffold + level and counters are written only if no other turn changed them first. +- **Modules read other domains through services.** Thought maps read conversations + through `ConversationService` and canvas sessions through `CanvasService`, not their + models. Module dependencies are one-way; there is no `forwardRef`. + +## A reply, end to end + +```mermaid +sequenceDiagram + participant B as Browser + participant API as API instance (holds the SSE) + participant Q as BullMQ (Redis) + participant Wk as Worker (any instance) + participant Bus as StreamBus (Redis pub/sub) + + B->>API: GET .../stream (SSE, stays open) + API->>Bus: subscribe mukti:stream:conversation: + B->>API: POST .../messages + API->>API: AiRequestGate.authorize + API->>Q: enqueue job + API-->>B: 202 { jobId } + Q->>Wk: job + Wk->>Wk: SocraticTurnEngine.prepare, model call, settle + Wk->>Bus: processing / message / complete + Bus->>API: same events + API-->>B: SSE events +``` + +Jobs may run on a different process than the one holding the browser's stream, so +stream services publish through the global `StreamBus` (Redis pub/sub when hosted, +in-process when local). Each stream has its own Redis channel +(`mukti:stream:{namespace}:{key}`, e.g. `mukti:stream:conversation:`). An instance +subscribes to a channel when it gains its first connection for that stream and +unsubscribes when it loses its last, so events reach only the instances that hold +the stream. Order within a job is preserved because Redis keeps per-publisher, +per-channel order. + +Pub/sub does not buffer, so a stream only receives what is published after its +subscription lands. `StreamBus.track` resolves at that point. Map extraction is the +one stream whose job is enqueued before the client connects, so its SSE endpoint +waits for `track` and then re-checks the job. If the job finished during that gap, +the endpoint replays the stored result and skips anything already sent live. + +## Contracts between web and API + +Every SSE stream's event shapes (chat, canvas and map node chat, branch suggestions, +conversation-to-map extraction) are defined once in the private, type-only workspace +package `@muktiai/contracts` (`packages/mukti-contracts`). The API's stream services +alias those types, and the web compiles a two-way equality check against them +(`packages/mukti-web/src/types/stream-contract.check.ts`), so drift fails `tsc` and +`next build`. The package is never published: imports are `import type`, so they +vanish from the API's runtime JavaScript. diff --git a/docs/rfcs/active/rfc-0001-knowledge-gap-detection/index.md b/docs/rfcs/active/rfc-0001-knowledge-gap-detection/index.md index f0d06366..60bfc051 100644 --- a/docs/rfcs/active/rfc-0001-knowledge-gap-detection/index.md +++ b/docs/rfcs/active/rfc-0001-knowledge-gap-detection/index.md @@ -74,7 +74,7 @@ Mukti's MCP server already has partial detection (`stuckIndicators` in `response | Reference | Relevance | | --------------------------------------------------- | -------------------------------------------------------------------------- | -| `mukti-mcp-server/src/response-strategies.ts` | Existing `stuckIndicators` detection (disconnected from API) | +| `packages/mukti-mcp/src/response-strategies.ts` | Existing `stuckIndicators` detection (disconnected from API) | | `docs/reference/planning/unified-thinking-model.md` | Documents "Anti-Resolution Signals" and frustration patterns | | GitHub Issue #3 | "Implement dynamic resource suggestion engine" — related but unimplemented | | Carnegie Learning BKT | Production-validated probabilistic knowledge tracking | @@ -420,7 +420,6 @@ Implemented base route: `knowledge-tracing` - `DELETE /knowledge-tracing/state/:userId/:conceptId/reset` - `GET /knowledge-tracing/concept/:conceptId/struggling` - `GET /knowledge-tracing/concept/:conceptId/mastered` -- `GET /knowledge-tracing/admin/cache-stats` - `POST /knowledge-tracing/concepts` - `GET /knowledge-tracing/concepts` - `GET /knowledge-tracing/concepts/:conceptId` @@ -692,7 +691,7 @@ Ask users to rate their own knowledge before each topic. - [Carnegie Learning BKT Paper](https://www.cs.cmu.edu/~ggordon/yudelson-koedinger-gordon-individualized-bkt.pdf) - [ALEKS Knowledge Space Theory](https://www.aleks.com/about_aleks/knowledge_space_theory) - [RPKT: Recursive Prerequisite Knowledge Tracing (2025)](https://arxiv.org/abs/2501.xxxxx) -- [Mukti MCP Server Response Strategies](../../../../mukti-mcp-server/src/response-strategies.ts) +- [Mukti MCP Server Response Strategies](../../../../packages/mukti-mcp/src/response-strategies.ts) - [Unified Thinking Model](../../../reference/planning/unified-thinking-model.md) - [GitHub Issue #3: Dynamic Resource Suggestion](https://github.com/mukti/mukti/issues/3) diff --git a/packages/ARCHITECTURE.md b/packages/ARCHITECTURE.md index 3b98c300..6ae74963 100644 --- a/packages/ARCHITECTURE.md +++ b/packages/ARCHITECTURE.md @@ -1,46 +1,13 @@ # Mukti Architecture -## Overview - -Mukti follows a modern cloud-native architecture built on NestJS, designed for scalability and performance. +The architecture overview lives in +[`docs/reference/architecture/overview.md`](../docs/reference/architecture/overview.md): +the packages, how hosted and local mode differ, the module map of the API, and how a +reply travels from the browser to a worker and back.

- Mukti Architecture + Original Mukti architecture sketch (aspirational, see overview.md for the current design)

-## Architecture Layers - -### Client Layer - -Web and mobile applications serve as entry points for end users, providing responsive interfaces across platforms. - -### CDN Layer - -CloudFlare CDN handles content delivery, caching static assets and reducing latency globally. - -### Backend Core - -NestJS powers the core API layer, orchestrating business logic and request handling. - -### External Services - -MongoDB serves as the primary operational database, and authentication is handled directly within NestJS (JWT + refresh tokens backed by MongoDB user records). - -### Application Layer - -The main application logic runs on NestJS deployed within Kubernetes clusters, ensuring high availability and horizontal scaling. - -### Infrastructure Services - -- **MongoDB**: Primary document database for persistent data storage -- **Redis**: In-memory cache for session management and performance optimization -- **Object Storage (S3)**: Handles file uploads and media storage - -### Third-Party Integrations - -- **Payment Gateway**: Processes transactions and payment operations -- **Email Service**: Manages transactional and notification emails - -## Data Flow - -Requests flow from clients through the CDN into the NestJS core, which performs authentication itself and coordinates with MongoDB for persistence, Redis for caching, and the Kubernetes-deployed application layer. External integrations handle specialized operations like payments and email delivery. +The image above is the original sketch and shows infrastructure Mukti does not use +(CDN, Kubernetes, S3, payments). The overview's diagrams are current. diff --git a/packages/mukti-api/Dockerfile b/packages/mukti-api/Dockerfile index 72110dba..9fe7ff98 100644 --- a/packages/mukti-api/Dockerfile +++ b/packages/mukti-api/Dockerfile @@ -6,6 +6,7 @@ FROM base AS deps COPY package.json bun.lock ./ COPY packages/mukti-api/package.json packages/mukti-api/ COPY packages/mukti-web/package.json packages/mukti-web/ +COPY packages/mukti-contracts/package.json packages/mukti-contracts/ RUN bun install # Stage 2: Build @@ -21,6 +22,7 @@ FROM base AS prod-deps COPY package.json bun.lock ./ COPY packages/mukti-api/package.json packages/mukti-api/ COPY packages/mukti-web/package.json packages/mukti-web/ +COPY packages/mukti-contracts/package.json packages/mukti-contracts/ RUN bun install # Stage 4: Runner diff --git a/packages/mukti-api/package.json b/packages/mukti-api/package.json index 0e3695e1..5cba9686 100644 --- a/packages/mukti-api/package.json +++ b/packages/mukti-api/package.json @@ -26,7 +26,9 @@ "./package.json": "./package.json" }, "files": [ - "dist" + "dist", + "!dist/**/*.d.ts", + "dist/start.d.ts" ], "publishConfig": { "access": "public" diff --git a/packages/mukti-api/project.json b/packages/mukti-api/project.json index e3dcea7c..6f83fa3d 100644 --- a/packages/mukti-api/project.json +++ b/packages/mukti-api/project.json @@ -87,7 +87,7 @@ "test:e2e": { "executor": "nx:run-commands", "options": { - "command": "jest --config ./test/jest-e2e.json", + "command": "jest --config ./test/jest-e2e.json --runInBand --forceExit", "cwd": "packages/mukti-api" } }, diff --git a/packages/mukti-api/src/core/stream-bus/__tests__/stream-bus.spec.ts b/packages/mukti-api/src/core/stream-bus/__tests__/stream-bus.spec.ts index c9c254f7..ccefc97d 100644 --- a/packages/mukti-api/src/core/stream-bus/__tests__/stream-bus.spec.ts +++ b/packages/mukti-api/src/core/stream-bus/__tests__/stream-bus.spec.ts @@ -1,41 +1,72 @@ +jest.mock('@openrouter/sdk', () => ({ + OpenRouter: jest.fn(() => ({})), +})); + import { StreamService } from '../../../modules/conversations/services/stream.service'; import { DialogueStreamService } from '../../../modules/dialogue/services/dialogue-stream.service'; +import { MapExtractionService } from '../../../modules/thought-map/services/map-extraction.service'; import { InProcessStreamBus, RedisStreamBus, - STREAM_BUS_CHANNEL, type StreamBusRedisClients, + streamChannel, } from '../stream-bus'; /** - * A stand-in for one Redis server: every subscriber registered on it gets - * every message any publisher sends, like real pub/sub across processes. + * A stand-in for one Redis server with real channel semantics: a published + * message reaches only subscribers that subscribed to that channel. */ class FakeRedisServer { - private readonly listeners: ((channel: string, raw: string) => void)[] = []; + /** Every publish, in order, for asserting where traffic went. */ + readonly published: string[] = []; + + private readonly subscribers: { + channels: Set; + listener?: (channel: string, raw: string) => void; + }[] = []; clients(): StreamBusRedisClients { + const sub: (typeof this.subscribers)[number] = { channels: new Set() }; + this.subscribers.push(sub); + return { publisher: { publish: jest.fn((channel: string, raw: string) => { - for (const listener of this.listeners) { - listener(channel, raw); + this.published.push(channel); + let receivers = 0; + for (const s of this.subscribers) { + if (s.channels.has(channel) && s.listener) { + s.listener(channel, raw); + receivers++; + } } - return Promise.resolve(this.listeners.length); + return Promise.resolve(receivers); }) as never, quit: jest.fn(() => Promise.resolve('OK')) as never, }, subscriber: { on: jest.fn((event: string, listener: never) => { if (event === 'message') { - this.listeners.push(listener); + sub.listener = listener; } }) as never, quit: jest.fn(() => Promise.resolve('OK')) as never, - subscribe: jest.fn(() => Promise.resolve(1)) as never, + subscribe: jest.fn((channel: string) => { + sub.channels.add(channel); + return Promise.resolve(sub.channels.size); + }) as never, + unsubscribe: jest.fn((channel: string) => { + sub.channels.delete(channel); + return Promise.resolve(sub.channels.size); + }) as never, }, }; } + + /** How many instances are subscribed to a channel right now. */ + subscriberCount(channel: string): number { + return this.subscribers.filter((s) => s.channels.has(channel)).length; + } } /** Lets fire-and-forget publishes settle. */ @@ -163,18 +194,208 @@ describe('StreamBus', () => { expect(handler).not.toHaveBeenCalled(); }); - it('subscribes on the shared channel and closes both connections on shutdown', async () => { + it('subscribes to nothing until a connection arrives, and closes both connections on shutdown', async () => { const clients = server.clients(); const bus = new RedisStreamBus(clients); - expect(clients.subscriber.subscribe).toHaveBeenCalledWith( - STREAM_BUS_CHANNEL, - ); + expect(clients.subscriber.subscribe).not.toHaveBeenCalled(); await bus.onModuleDestroy(); expect(clients.publisher.quit).toHaveBeenCalled(); expect(clients.subscriber.quit).toHaveBeenCalled(); }); + it('publishes each stream on its own channel and only holders receive it', async () => { + const otherInstance = new RedisStreamBus(server.clients()); + const api = new StreamService(apiInstance); + const other = new StreamService(otherInstance); + const worker = new StreamService(workerInstance); + const apiEmit = jest.fn(); + const otherEmit = jest.fn(); + api.addConnection('conv-1', 'user-1', 'conn-1', apiEmit); + other.addConnection('conv-2', 'user-2', 'conn-2', otherEmit); + const otherDispatch = jest.spyOn( + otherInstance as unknown as { dispatch: (raw: string) => void }, + 'dispatch', + ); + + worker.emitToConversation('conv-1', { + data: { jobId: 'j', status: 'started' }, + type: 'processing', + }); + await flush(); + + expect(server.published).toEqual([ + streamChannel('conversation', 'conv-1'), + ]); + expect(apiEmit).toHaveBeenCalledTimes(1); + // The instance holding only conv-2 never even sees conv-1's traffic. + expect(otherDispatch).not.toHaveBeenCalled(); + expect(otherEmit).not.toHaveBeenCalled(); + }); + + it('keeps one subscription per stream across connections and drops it with the last one', async () => { + const api = new StreamService(apiInstance); + const channel = streamChannel('conversation', 'conv-1'); + api.addConnection('conv-1', 'user-1', 'tab-1', jest.fn()); + api.addConnection('conv-1', 'user-1', 'tab-2', jest.fn()); + await flush(); + + expect(server.subscriberCount(channel)).toBe(1); + + api.removeConnection('conv-1', 'tab-1'); + await flush(); + expect(server.subscriberCount(channel)).toBe(1); + + api.removeConnection('conv-1', 'tab-2'); + await flush(); + expect(server.subscriberCount(channel)).toBe(0); + + // Removing again, or cleaning up an already-empty stream, is harmless. + api.removeConnection('conv-1', 'tab-2'); + api.cleanupConversation('conv-1'); + api.addConnection('conv-1', 'user-1', 'tab-3', jest.fn()); + await flush(); + expect(server.subscriberCount(channel)).toBe(1); + }); + + it('delivers extraction events from the worker to the instance holding the job stream', async () => { + const deps = Array.from({ length: 8 }, () => ({}) as never); + const api = new MapExtractionService( + ...(deps as [never, never, never, never, never, never, never, never]), + apiInstance, + ); + const worker = new MapExtractionService( + ...(deps as [never, never, never, never, never, never, never, never]), + workerInstance, + ); + const emitFn = jest.fn(); + const connection = api.addConnection('job-1', 'conn-1', emitFn); + await connection.ready; + + ( + worker as unknown as { + emit: (jobId: string, event: { type: string }) => void; + } + ).emit('job-1', { type: 'complete' }); + await flush(); + + expect(emitFn).toHaveBeenCalledWith( + expect.objectContaining({ type: 'complete' }), + ); + connection.close(); + await flush(); + expect(server.subscriberCount(streamChannel('extraction', 'job-1'))).toBe( + 0, + ); + }); + + it('resolves track only once the channel is subscribed, sharing one subscribe per stream', async () => { + const clients = server.clients(); + let finishSubscribe!: () => void; + (clients.subscriber.subscribe as jest.Mock).mockImplementationOnce( + () => + new Promise((resolve) => { + finishSubscribe = () => resolve(1); + }), + ); + const bus = new RedisStreamBus(clients); + + let firstReady = false; + let secondReady = false; + void bus.track('extraction', 'job-1').then(() => (firstReady = true)); + void bus.track('extraction', 'job-1').then(() => (secondReady = true)); + await flush(); + + expect(clients.subscriber.subscribe).toHaveBeenCalledTimes(1); + expect([firstReady, secondReady]).toEqual([false, false]); + + finishSubscribe(); + await flush(); + expect([firstReady, secondReady]).toEqual([true, true]); + }); + + it('resolves track even when the subscribe fails, so callers can fall back', async () => { + const clients = server.clients(); + (clients.subscriber.subscribe as jest.Mock).mockRejectedValueOnce( + new Error('NOPERM'), + ); + const bus = new RedisStreamBus(clients); + const errorSpy = jest + .spyOn( + (bus as unknown as { logger: { error: () => void } }).logger, + 'error', + ) + .mockImplementation(() => undefined); + + await expect(bus.track('extraction', 'job-1')).resolves.toBeUndefined(); + expect(errorSpy).toHaveBeenCalledWith( + expect.stringContaining('could not subscribe'), + ); + }); + + it('holds the subscription while any of several extraction connections is open', async () => { + const deps = Array.from({ length: 8 }, () => ({}) as never); + const api = new MapExtractionService( + ...(deps as [never, never, never, never, never, never, never, never]), + apiInstance, + ); + const channel = streamChannel('extraction', 'job-1'); + const a = api.addConnection('job-1', 'a', jest.fn()); + const b = api.addConnection('job-1', 'b', jest.fn()); + await Promise.all([a.ready, b.ready]); + expect(server.subscriberCount(channel)).toBe(1); + + a.close(); + a.close(); + await flush(); + expect(server.subscriberCount(channel)).toBe(1); + + b.close(); + await flush(); + expect(server.subscriberCount(channel)).toBe(0); + }); + + it('releases the subscription when a stream is cleaned up wholesale', async () => { + const dialogue = new DialogueStreamService(apiInstance); + const channel = streamChannel('dialogue', 'session-1:node-1'); + dialogue.addConnection('session-1', 'node-1', 'user-1', 'a', jest.fn()); + dialogue.addConnection('session-1', 'node-1', 'user-1', 'b', jest.fn()); + await flush(); + expect(server.subscriberCount(channel)).toBe(1); + + dialogue.cleanupNodeDialogue('session-1', 'node-1'); + await flush(); + expect(server.subscriberCount(channel)).toBe(0); + }); + + it('keeps a job’s events in order on the receiving instance', async () => { + const api = new StreamService(apiInstance); + const worker = new StreamService(workerInstance); + const emitFn = jest.fn(); + api.addConnection('conv-1', 'user-1', 'conn-1', emitFn); + await flush(); + + worker.emitToConversation('conv-1', { + data: { jobId: 'j', status: 'started' }, + type: 'processing', + }); + worker.emitToConversation('conv-1', { + data: { jobId: 'j', status: 'completed' }, + type: 'progress', + } as never); + worker.emitToConversation('conv-1', { + data: { jobId: 'j' }, + type: 'complete', + } as never); + await flush(); + + expect(emitFn.mock.calls.map(([event]) => event.type)).toEqual([ + 'processing', + 'progress', + 'complete', + ]); + }); + it('logs instead of throwing when Redis rejects a publish', async () => { const clients = server.clients(); (clients.publisher.publish as jest.Mock).mockRejectedValueOnce( diff --git a/packages/mukti-api/src/core/stream-bus/stream-bus.ts b/packages/mukti-api/src/core/stream-bus/stream-bus.ts index 3cfdec1c..3565b7ad 100644 --- a/packages/mukti-api/src/core/stream-bus/stream-bus.ts +++ b/packages/mukti-api/src/core/stream-bus/stream-bus.ts @@ -25,13 +25,29 @@ export type StreamBusHandler = (key: string, event: unknown) => void; * Namespaces separate surfaces (`conversation`, `dialogue`, `extraction`); * the key identifies the stream within one (a conversation id, a * `session:node` pair, a job id). + * + * Stream services call {@link track} when an instance gains its first local + * connection for a stream and {@link untrack} when it loses its last, so a + * transport can deliver each stream only to instances that hold it. */ export abstract class StreamBus { /** Subscribes to a namespace. Returns a function that unsubscribes. */ abstract listen(namespace: string, handler: StreamBusHandler): () => void; - /** Sends an event to every listener of the namespace on every instance. */ + /** Sends an event to the instances holding this stream. */ abstract publish(namespace: string, key: string, event: unknown): void; + + /** + * This instance now holds a connection for the stream. Reference counted. + * + * @returns Resolves once events published from now on will be delivered + * here. Never rejects; a failed subscribe is logged and resolves, so callers + * that re-check state afterwards still make progress. + */ + abstract track(namespace: string, key: string): Promise; + + /** Releases one {@link track}. */ + abstract untrack(namespace: string, key: string): void; } /** @@ -53,17 +69,32 @@ export class InProcessStreamBus extends StreamBus { handler(key, event); } } + + /** One process sees every stream, so there is nothing to subscribe to. */ + track(): Promise { + return Promise.resolve(); + } + + /** See {@link track}. */ + untrack(): void { + return; + } } -/** Redis channel every instance publishes to and subscribes on. */ -export const STREAM_BUS_CHANNEL = 'mukti:stream-events'; +/** Prefix of every per-stream Redis channel: `mukti:stream:{namespace}:{key}`. */ +export const STREAM_BUS_CHANNEL_PREFIX = 'mukti:stream:'; + +/** Builds the Redis channel name for one stream. */ +export function streamChannel(namespace: string, key: string): string { + return `${STREAM_BUS_CHANNEL_PREFIX}${namespace}:${key}`; +} /** Token for the Redis clients, so tests can inject fakes. */ export const STREAM_BUS_REDIS = Symbol('STREAM_BUS_REDIS'); export interface StreamBusRedisClients { publisher: Pick; - subscriber: Pick; + subscriber: Pick; } interface Envelope { @@ -73,18 +104,25 @@ interface Envelope { } /** - * Fans events out across instances with Redis pub/sub on one channel. + * Fans events out across instances with one Redis pub/sub channel per stream. * * @remarks - * Every instance receives every event and drops the ones it holds no - * connection for, which is fine at this scale and keeps ordering simple: one - * publisher connection per process, and Redis preserves order per publisher, - * so a job's `processing → message → complete` sequence arrives in order. An - * instance's own events also come back through Redis, so nothing is - * delivered twice. + * An instance subscribes to a stream's channel only while it holds at least + * one connection for that stream, so each event reaches only the instances + * that can deliver it and Redis traffic grows with active streams, not with + * instances times total traffic. + * + * Ordering: each process has one publisher connection, and Redis preserves + * order per publisher and channel, so a job's `processing → message → + * complete` sequence arrives in order. An instance's own events reach it + * through Redis like everyone else's, so nothing is delivered twice. * - * Pub/sub is fire-and-forget like the in-memory emit it replaces: a client - * that is not connected when an event is published does not get it. + * Pub/sub is fire-and-forget: a client gets only events published after its + * instance has subscribed to the stream, which is one Redis round trip after + * the connection opens. {@link track} resolves at that point, so a stream + * whose job may already be running can re-check the job afterwards (the + * extraction stream does). Ioredis re-subscribes tracked channels on + * reconnect. */ @Injectable() export class RedisStreamBus extends StreamBus implements OnModuleDestroy { @@ -92,22 +130,21 @@ export class RedisStreamBus extends StreamBus implements OnModuleDestroy { private readonly logger = new Logger(RedisStreamBus.name); + /** The subscribe in flight (or done) for each tracked channel. */ + private readonly subscribed = new Map>(); + + /** Local connection count per subscribed channel. */ + private readonly tracked = new Map(); + constructor( @Inject(STREAM_BUS_REDIS) private readonly redis: StreamBusRedisClients, ) { super(); this.redis.subscriber.on('message', (channel: string, raw: string) => { - if (channel === STREAM_BUS_CHANNEL) { + if (channel.startsWith(STREAM_BUS_CHANNEL_PREFIX)) { this.dispatch(raw); } }); - void this.redis.subscriber - .subscribe(STREAM_BUS_CHANNEL) - .catch((error: unknown) => - this.logger.error( - `Stream bus could not subscribe; live updates from other instances will be missed: ${String(error)}`, - ), - ); } listen(namespace: string, handler: StreamBusHandler): () => void { @@ -127,7 +164,7 @@ export class RedisStreamBus extends StreamBus implements OnModuleDestroy { publish(namespace: string, key: string, event: unknown): void { const envelope: Envelope = { event, key, namespace }; void this.redis.publisher - .publish(STREAM_BUS_CHANNEL, JSON.stringify(envelope)) + .publish(streamChannel(namespace, key), JSON.stringify(envelope)) .catch((error: unknown) => this.logger.error( `Stream event ${namespace}/${key} not published: ${String(error)}`, @@ -135,6 +172,51 @@ export class RedisStreamBus extends StreamBus implements OnModuleDestroy { ); } + track(namespace: string, key: string): Promise { + const channel = streamChannel(namespace, key); + const count = this.tracked.get(channel) ?? 0; + this.tracked.set(channel, count + 1); + + const pending = this.subscribed.get(channel); + if (count > 0 && pending) { + return pending; + } + + const ready = this.redis.subscriber.subscribe(channel).then( + () => undefined, + (error: unknown) => { + this.logger.error( + `Stream bus could not subscribe to ${channel}; this client will miss live updates: ${String(error)}`, + ); + }, + ); + this.subscribed.set(channel, ready); + return ready; + } + + untrack(namespace: string, key: string): void { + const channel = streamChannel(namespace, key); + const count = this.tracked.get(channel) ?? 0; + + if (count > 1) { + this.tracked.set(channel, count - 1); + return; + } + if (count === 0) { + return; + } + + this.tracked.delete(channel); + this.subscribed.delete(channel); + void this.redis.subscriber + .unsubscribe(channel) + .catch((error: unknown) => + this.logger.warn( + `Stream bus could not unsubscribe from ${channel}: ${String(error)}`, + ), + ); + } + private dispatch(raw: string): void { let envelope: Envelope; try { diff --git a/packages/mukti-api/src/modules/conversations/services/stream.service.ts b/packages/mukti-api/src/modules/conversations/services/stream.service.ts index afdc8db5..6c866610 100644 --- a/packages/mukti-api/src/modules/conversations/services/stream.service.ts +++ b/packages/mukti-api/src/modules/conversations/services/stream.service.ts @@ -1,6 +1,6 @@ -import { Injectable, Logger, Optional } from '@nestjs/common'; +import type { ConversationStreamEventContract } from '@muktiai/contracts'; -import type { ConversationStreamEventContract } from '../../../contracts/stream-events'; +import { Injectable, Logger, Optional } from '@nestjs/common'; import { InProcessStreamBus, @@ -123,6 +123,11 @@ export class StreamService { // Get existing connections for this conversation or create new array const existingConnections = this.connections.get(conversationId) ?? []; + // First local connection for this stream: start receiving its events. + if (existingConnections.length === 0) { + void this.bus.track(BUS_NAMESPACE, conversationId); + } + // Add new connection to the array existingConnections.push(connection); @@ -174,6 +179,7 @@ export class StreamService { // Remove all connections for this conversation this.connections.delete(conversationId); + this.bus.untrack(BUS_NAMESPACE, conversationId); this.logger.log( `Cleaned up ${connectionCount} connection(s) for conversation ${conversationId}`, @@ -342,6 +348,7 @@ export class StreamService { // Update or remove the conversation entry if (updatedConnections.length === 0) { this.connections.delete(conversationId); + this.bus.untrack(BUS_NAMESPACE, conversationId); this.logger.log( `All connections removed for conversation ${conversationId}. Entry deleted from map.`, ); diff --git a/packages/mukti-api/src/modules/dialogue/services/dialogue-stream.service.ts b/packages/mukti-api/src/modules/dialogue/services/dialogue-stream.service.ts index 45193cd9..5289c226 100644 --- a/packages/mukti-api/src/modules/dialogue/services/dialogue-stream.service.ts +++ b/packages/mukti-api/src/modules/dialogue/services/dialogue-stream.service.ts @@ -1,6 +1,6 @@ -import { Injectable, Logger, Optional } from '@nestjs/common'; +import type { NodeDialogueStreamEventContract } from '@muktiai/contracts'; -import type { NodeDialogueStreamEventContract } from '../../../contracts/stream-events'; +import { Injectable, Logger, Optional } from '@nestjs/common'; import { InProcessStreamBus, @@ -77,6 +77,10 @@ export class DialogueStreamService { }; const existingConnections = this.connections.get(key) ?? []; + // First local connection for this stream: start receiving its events. + if (existingConnections.length === 0) { + void this.bus.track(BUS_NAMESPACE, key); + } existingConnections.push(connection); this.connections.set(key, existingConnections); @@ -98,6 +102,7 @@ export class DialogueStreamService { } this.connections.delete(key); + this.bus.untrack(BUS_NAMESPACE, key); this.logger.log( `Cleaned up ${connections.length} connection(s) for ${key}`, ); @@ -171,6 +176,7 @@ export class DialogueStreamService { if (updatedConnections.length === 0) { this.connections.delete(key); + this.bus.untrack(BUS_NAMESPACE, key); this.logger.log(`All connections removed for ${key}`); } else { this.connections.set(key, updatedConnections); diff --git a/packages/mukti-api/src/modules/knowledge-tracing/README.md b/packages/mukti-api/src/modules/knowledge-tracing/README.md index 7b791870..205d48fd 100644 --- a/packages/mukti-api/src/modules/knowledge-tracing/README.md +++ b/packages/mukti-api/src/modules/knowledge-tracing/README.md @@ -8,7 +8,7 @@ This module tracks user mastery of concepts using the 4-parameter BKT probabilis - **Real-time knowledge state updates** based on user responses - **Intelligent recommendations** for next learning actions -- **Performance-optimized architecture** with in-memory caching +- **Concurrency-safe writes**: overlapping turns, job retries and multiple API instances never lose an observation - **Event-driven integration** for analytics and adaptive learning systems ## BKT Model @@ -38,41 +38,23 @@ P(L_n + 1) = posterior + (1 - posterior) * P(T); ## Architecture -``` -┌─────────────────────────────────────────────────────────┐ -│ REST API Layer │ -│ (KnowledgeTracingController) │ -└────────────────────┬────────────────────────────────────┘ - │ - ▼ -┌─────────────────────────────────────────────────────────┐ -│ Business Logic Layer │ -│ (KnowledgeStateTrackerService) │ -│ ┌──────────────────────────────────────────────────┐ │ -│ │ In-Memory Cache (Map) │ │ -│ └──────────────────────────────────────────────────┘ │ -│ │ │ -│ ┌───────────┴───────────┐ │ -│ ▼ ▼ │ -│ ┌──────────────┐ ┌──────────────┐ │ -│ │ BKT Algorithm│ │ Event Emitter│ │ -│ │ Service │ │ (Updates) │ │ -│ └──────────────┘ └──────────────┘ │ -└────────────────────┬────────────────────────────────────┘ - │ - ▼ -┌─────────────────────────────────────────────────────────┐ -│ Persistence Layer │ -│ MongoDB (KnowledgeState Collection) │ -└─────────────────────────────────────────────────────────┘ +```mermaid +flowchart TD + API["REST API
KnowledgeTracingController"] --> Tracker["KnowledgeStateTrackerService
reads + knowledge-state.updated events"] + Turn["Socratic turn
(every chat, canvas and map reply)"] --> Gap["KnowledgeGapDetectorService"] + Tracker -->|writes| Store["KnowledgeStateStore
the only writer"] + Gap -->|writes| Store + Store -->|"read, BKT update,
compare-and-set on attempts"| DB[("MongoDB knowledge_states
unique { conceptId, userId }")] + Tracker -->|reads| DB ``` -### Performance Strategy +### Consistency Strategy -- **Hot Path**: In-memory Map cache for active knowledge states -- **Cold Path**: MongoDB persistence for long-term storage -- **Cache Key**: `${userId}:${conceptId}` -- **Cache Invalidation**: TTL-based (1 hour for inactive states) +- **No in-process cache.** Reads go to MongoDB, so every API instance and the reply pipeline see the same value. An earlier per-process cache served stale P(L) across instances and was removed. +- **One writer.** `KnowledgeStateStore.applyObservation` reads the record, applies BKT, and writes only if `attempts` is unchanged. If another turn wrote first it waits a few random milliseconds, re-reads and re-applies, up to 20 times. +- **Creation race.** When two first observations arrive together, the loser hits the unique index and then updates the winner's record. +- **Parameters are fixed at creation.** BKT parameters sent to `POST /knowledge-tracing/update` only apply when the record is first created. An existing record keeps its stored parameters. +- **Failure isolation.** In the Socratic turn a failed knowledge write is logged, not thrown: the reply is already saved, and a job retry would store and bill it twice. ## API Reference @@ -407,7 +389,7 @@ When contributing to this module: 1. **Preserve mathematical correctness** - BKT is a proven algorithm 2. **Add tests for new features** - maintain >90% coverage 3. **Document parameter changes** - educational implications matter -4. **Benchmark performance** - cache is critical for scale +4. **Keep writes in `KnowledgeStateStore`** - never write `knowledge_states` directly ## License diff --git a/packages/mukti-api/src/modules/knowledge-tracing/services/__tests__/knowledge-state-store.service.spec.ts b/packages/mukti-api/src/modules/knowledge-tracing/services/__tests__/knowledge-state-store.service.spec.ts index 50f0b901..0ffafd8e 100644 --- a/packages/mukti-api/src/modules/knowledge-tracing/services/__tests__/knowledge-state-store.service.spec.ts +++ b/packages/mukti-api/src/modules/knowledge-tracing/services/__tests__/knowledge-state-store.service.spec.ts @@ -138,7 +138,8 @@ describe('KnowledgeStateStore', () => { }); it('gives up after repeated conflicts instead of looping forever', async () => { - for (let i = 0; i < 5; i++) { + const cap = 20; + for (let i = 0; i < cap; i++) { reads(stored(i, 0.4)); writes(0); } @@ -146,6 +147,23 @@ describe('KnowledgeStateStore', () => { await expect( store.applyObservation(userId, 'recursion', true), ).rejects.toThrow('kept changing'); - expect(model.updateOne).toHaveBeenCalledTimes(5); + expect(model.updateOne).toHaveBeenCalledTimes(cap); + }); + + it('does not count creating the record against the retry budget', async () => { + // A missing record, then 19 conflicts, then success: still within the cap. + reads(null); + for (let i = 0; i < 20; i++) { + reads(stored(i, 0.4)); + } + for (let i = 0; i < 19; i++) { + writes(0); + } + writes(1); + + await expect( + store.applyObservation(userId, 'recursion', true), + ).resolves.toBeDefined(); + expect(model.create).toHaveBeenCalledTimes(1); }); }); diff --git a/packages/mukti-api/src/modules/knowledge-tracing/services/knowledge-state-store.service.ts b/packages/mukti-api/src/modules/knowledge-tracing/services/knowledge-state-store.service.ts index b2cced1d..9aa925eb 100644 --- a/packages/mukti-api/src/modules/knowledge-tracing/services/knowledge-state-store.service.ts +++ b/packages/mukti-api/src/modules/knowledge-tracing/services/knowledge-state-store.service.ts @@ -14,12 +14,22 @@ import { } from '../interfaces/bkt.interface'; import { BKTAlgorithmService } from './bkt-algorithm.service'; -/** How many times to re-read and retry when another writer got in first. */ -const MAX_ATTEMPTS = 5; +/** + * How many times to re-read and retry when another writer got in first. + * Each conflict means some other observation succeeded, so the loop always + * makes progress; the cap only bounds pathological contention. + */ +const MAX_ATTEMPTS = 20; + +/** Upper bound of the random pause between conflicting attempts, in ms. */ +const MAX_BACKOFF_MS = 25; /** MongoDB duplicate-key error (the unique `{ conceptId, userId }` index). */ const DUPLICATE_KEY = 11000; +const sleep = (ms: number) => + new Promise((resolve) => setTimeout(resolve, ms)); + export interface AppliedObservation { previousProbability: number; result: BKTUpdateResult; @@ -63,16 +73,19 @@ export class KnowledgeStateStore { correct: boolean, initialParameters: BKTParameters = DEFAULT_BKT_PARAMS, ): Promise { - for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) { + let attempt = 0; + while (attempt < MAX_ATTEMPTS) { const doc = await this.knowledgeStateModel .findOne({ conceptId, userId }) .lean() .exec(); if (!doc) { + // Creating is not a conflict; whoever wins, the next read finds it. await this.createIfAbsent(userId, conceptId, initialParameters); continue; } + attempt++; const current: KnowledgeState = { attempts: doc.attempts, @@ -106,6 +119,8 @@ export class KnowledgeStateStore { this.logger.debug( `Knowledge state ${userId.toString()}:${conceptId} changed concurrently, retrying (${attempt}/${MAX_ATTEMPTS})`, ); + // Jitter so writers that collided do not collide again in lockstep. + await sleep(Math.random() * MAX_BACKOFF_MS); } throw new Error( diff --git a/packages/mukti-api/src/modules/thought-map/__tests__/thought-map.controller.spec.ts b/packages/mukti-api/src/modules/thought-map/__tests__/thought-map.controller.spec.ts index 9f5ba57a..633b0588 100644 --- a/packages/mukti-api/src/modules/thought-map/__tests__/thought-map.controller.spec.ts +++ b/packages/mukti-api/src/modules/thought-map/__tests__/thought-map.controller.spec.ts @@ -25,6 +25,9 @@ import { ThoughtMapShareService } from '../services/thought-map-share.service'; import { ThoughtMapService } from '../services/thought-map.service'; import { ThoughtMapController } from '../thought-map.controller'; +/** Lets chained promise callbacks (several hops) settle. */ +const flushPromises = () => new Promise((resolve) => setImmediate(resolve)); + describe('ThoughtMapController', () => { let controller: ThoughtMapController; @@ -665,7 +668,7 @@ describe('ThoughtMapController', () => { mockMapExtractionService.addConnection.mockImplementation( (_jobId: string, _connectionId: string, cb: (event: unknown) => void) => { emitFn = cb; - return cleanup; + return { close: cleanup, ready: Promise.resolve() }; }, ); @@ -674,6 +677,7 @@ describe('ThoughtMapController', () => { const subscription = stream.subscribe((event) => events.push(event)); emitFn({ data: { status: 'started' }, type: 'processing' }); + await flushPromises(); subscription.unsubscribe(); expect(events).toEqual([ @@ -685,6 +689,144 @@ describe('ThoughtMapController', () => { expect(cleanup).toHaveBeenCalled(); }); + describe('extraction stream subscribe race', () => { + const draftMap = { + map: { _id: 'map-1', status: 'draft' }, + nodes: [{ nodeId: 'topic-0' }], + }; + const result = { conversationId: 'c-1', mapId: 'map-1', nodeCount: 1 }; + const job = (state: string, extra: Record = {}) => ({ + data: { userId: mockUser._id.toString() }, + getState: jest.fn().mockResolvedValue(state), + ...extra, + }); + + /** Subscribes, holding `ready` until `subscribed()` is called. */ + const open = async () => { + let emitFn!: (event: unknown) => void; + let subscribed!: () => void; + mockMapExtractionService.addConnection.mockImplementation( + (_j: string, _c: string, cb: (event: unknown) => void) => { + emitFn = cb; + return { + close: jest.fn(), + ready: new Promise((resolve) => (subscribed = resolve)), + }; + }, + ); + const stream = await controller.streamExtraction( + 'job-1', + mockUser as any, + ); + const types: string[] = []; + stream.subscribe((event) => types.push((event.data as any).type)); + return { emit: (e: unknown) => emitFn(e), subscribed, types }; + }; + + beforeEach(() => { + mockMapExtractionService.getExtractionJob.mockReset(); + mockMapExtractionService.addConnection.mockReset(); + mockThoughtMapService.getMap.mockResolvedValue(draftMap); + }); + + it('replays the result when the job completed while subscribing', async () => { + mockMapExtractionService.getExtractionJob + .mockResolvedValueOnce(job('active')) + .mockResolvedValueOnce(job('completed', { returnvalue: result })); + + const s = await open(); + s.subscribed(); + await flushPromises(); + + expect(s.types).toEqual(['preview', 'complete']); + }); + + it('replays the failure when the job failed while subscribing', async () => { + mockMapExtractionService.getExtractionJob + .mockResolvedValueOnce(job('active')) + .mockResolvedValueOnce(job('failed', { failedReason: 'model down' })); + + const s = await open(); + s.subscribed(); + await flushPromises(); + + expect(s.types).toEqual(['error']); + }); + + it('does nothing extra when the job is still running after subscribing', async () => { + mockMapExtractionService.getExtractionJob.mockResolvedValue( + job('active'), + ); + + const s = await open(); + s.subscribed(); + await flushPromises(); + s.emit({ data: draftMap, type: 'preview' }); + s.emit({ data: { jobId: 'job-1' }, type: 'complete' }); + + expect(s.types).toEqual(['preview', 'complete']); + expect(mockThoughtMapService.getMap).not.toHaveBeenCalled(); + }); + + it('skips the re-check once the live stream already completed', async () => { + mockMapExtractionService.getExtractionJob.mockResolvedValueOnce( + job('active'), + ); + + const s = await open(); + s.emit({ data: draftMap, type: 'preview' }); + s.emit({ data: { jobId: 'job-1' }, type: 'complete' }); + s.subscribed(); + await flushPromises(); + + expect(s.types).toEqual(['preview', 'complete']); + expect(mockMapExtractionService.getExtractionJob).toHaveBeenCalledTimes( + 1, + ); + }); + + it('does not repeat a live preview when the replay arrives after it', async () => { + mockMapExtractionService.getExtractionJob + .mockResolvedValueOnce(job('active')) + .mockResolvedValueOnce(job('completed', { returnvalue: result })); + + const s = await open(); + s.emit({ data: draftMap, type: 'preview' }); + s.subscribed(); + await flushPromises(); + // A late live `complete` after the replay is dropped too. + s.emit({ data: { jobId: 'job-1' }, type: 'complete' }); + + expect(s.types).toEqual(['preview', 'complete']); + }); + + it('reloads a job that finished after it was loaded instead of reporting a missing result', async () => { + // BullMQ: getState() reads Redis live, but returnvalue is the snapshot + // taken by getJob(). A job that completes in between reports + // `completed` with no returnvalue yet. + mockMapExtractionService.getExtractionJob + .mockResolvedValueOnce(job('completed')) + .mockResolvedValueOnce(job('completed', { returnvalue: result })); + + const stream = await controller.streamExtraction( + 'job-1', + mockUser as any, + ); + const types: string[] = []; + await new Promise((resolve) => + stream.subscribe({ + complete: () => resolve(), + next: (event) => types.push((event.data as any).type), + }), + ); + + expect(types).toEqual(['preview', 'complete']); + expect(mockMapExtractionService.getExtractionJob).toHaveBeenCalledTimes( + 2, + ); + }); + }); + it('replays completed extraction jobs for late subscribers', async () => { mockMapExtractionService.getExtractionJob.mockResolvedValue({ data: { userId: mockUser._id.toString() }, diff --git a/packages/mukti-api/src/modules/thought-map/services/__tests__/map-extraction.service.spec.ts b/packages/mukti-api/src/modules/thought-map/services/__tests__/map-extraction.service.spec.ts index 0c4f3b8b..71de04ad 100644 --- a/packages/mukti-api/src/modules/thought-map/services/__tests__/map-extraction.service.spec.ts +++ b/packages/mukti-api/src/modules/thought-map/services/__tests__/map-extraction.service.spec.ts @@ -83,13 +83,13 @@ describe('MapExtractionService', () => { describe('connection management', () => { it('adds extraction connections and removes them via cleanup', () => { const emitFn = jest.fn(); - const cleanup = service.addConnection('job-1', 'conn-1', emitFn); + const { close } = service.addConnection('job-1', 'conn-1', emitFn); (service as any).emit('job-1', { data: { jobId: 'job-1', status: 'started' }, type: 'processing', }); - cleanup(); + close(); (service as any).emit('job-1', { data: { jobId: 'job-1', status: 'started' }, type: 'processing', diff --git a/packages/mukti-api/src/modules/thought-map/services/__tests__/thought-map-dialogue-queue.service.spec.ts b/packages/mukti-api/src/modules/thought-map/services/__tests__/thought-map-dialogue-queue.service.spec.ts index ff001374..4f05ead1 100644 --- a/packages/mukti-api/src/modules/thought-map/services/__tests__/thought-map-dialogue-queue.service.spec.ts +++ b/packages/mukti-api/src/modules/thought-map/services/__tests__/thought-map-dialogue-queue.service.spec.ts @@ -512,6 +512,12 @@ describe('ThoughtMapDialogueQueueService', () => { rootGap: 'gap-root', }), true, + // Guarded on the counters this turn read (failures 1, successes 2) + { + consecutiveFailures: 1, + consecutiveSuccesses: 2, + currentScaffoldLevel: { $in: [0, null] }, + }, ); }); @@ -942,13 +948,42 @@ describe('ThoughtMapDialogueQueueService', () => { }); describe('private helpers', () => { - it('validates curated models when BYOK is not used', () => { - expect(() => - (service as any).validateEffectiveModel('blocked-model', false), - ).toThrow('MODEL_NOT_ALLOWED'); + it('rejects a non-curated model for a user without their own key before calling the AI', async () => { + jest.spyOn(service as any, 'getOrCreateMapDialogue').mockResolvedValue({ + _id: new Types.ObjectId(), + }); + mockDialogueService.addMessage.mockResolvedValue({ + _id: new Types.ObjectId(), + createdAt: new Date(), + sequence: 1, + }); + mockDialogueService.getMessages.mockResolvedValue({ + messages: [], + pagination: { total: 0 }, + }); + + await expect( + service.process({ + data: { + depth: 0, + fromSuggestion: false, + mapId: validMapId(), + message: 'Hi', + model: 'blocked-model', + nodeId: 'node-1', + nodeLabel: 'Node', + nodeType: 'thought', + siblings: 0, + subscriptionTier: 'free', + usedByok: false, + userId: new Types.ObjectId().toString(), + }, + id: 'job-1', + } as any), + ).rejects.toThrow('MODEL_NOT_ALLOWED'); expect( - (service as any).validateEffectiveModel('allowed-model', false), - ).toBe('allowed-model'); + mockDialogueAIService.generateScaffoldedResponseWithPrompt, + ).not.toHaveBeenCalled(); }); it('falls back to "Unknown topic" when the root node cannot be found', async () => { diff --git a/packages/mukti-api/src/modules/thought-map/services/branch-suggestion.service.ts b/packages/mukti-api/src/modules/thought-map/services/branch-suggestion.service.ts index 446533bc..2c5d8d99 100644 --- a/packages/mukti-api/src/modules/thought-map/services/branch-suggestion.service.ts +++ b/packages/mukti-api/src/modules/thought-map/services/branch-suggestion.service.ts @@ -1,3 +1,8 @@ +import type { + BranchSuggestionPayload, + SuggestionStreamEventContract, +} from '@muktiai/contracts'; + import { InjectQueue, Processor, WorkerHost } from '@nestjs/bullmq'; import { Inject, Injectable, Logger } from '@nestjs/common'; import { InjectModel } from '@nestjs/mongoose'; @@ -39,16 +44,9 @@ const SUGGESTION_COUNT = 3; // ============================================================================ /** - * A single AI-generated branch suggestion. + * A single AI-generated branch suggestion. Wire shape from the shared contract. */ -export interface BranchSuggestion { - /** Question text — always phrased as a question */ - label: string; - /** nodeId of the parent this should branch from */ - parentId: string; - /** Suggested type for the new node */ - suggestedType: 'question' | 'thought'; -} +export type BranchSuggestion = BranchSuggestionPayload; /** * Job data for a branch suggestion request. @@ -85,25 +83,9 @@ export interface BranchSuggestionJobResult { // ============================================================================ /** - * Union of all suggestion SSE event payloads. + * Union of all suggestion SSE event payloads (shared wire contract). */ -export type SuggestionStreamEvent = - | { - data: BranchSuggestion; - type: 'suggestion'; - } - | { - data: { code: string; message: string; retriable: boolean }; - type: 'error'; - } - | { - data: { jobId: string; status: 'started' }; - type: 'processing'; - } - | { - data: { jobId: string; suggestionCount: number }; - type: 'complete'; - }; +export type SuggestionStreamEvent = SuggestionStreamEventContract; /** * An active SSE listener registered for a suggestion job. diff --git a/packages/mukti-api/src/modules/thought-map/services/map-extraction.service.ts b/packages/mukti-api/src/modules/thought-map/services/map-extraction.service.ts index 711252c3..baab1629 100644 --- a/packages/mukti-api/src/modules/thought-map/services/map-extraction.service.ts +++ b/packages/mukti-api/src/modules/thought-map/services/map-extraction.service.ts @@ -1,3 +1,5 @@ +import type { ExtractionStreamEventContract } from '@muktiai/contracts'; + import { InjectQueue, Processor, WorkerHost } from '@nestjs/bullmq'; import { Inject, Injectable, Logger } from '@nestjs/common'; import { InjectModel } from '@nestjs/mongoose'; @@ -42,27 +44,10 @@ import { ConversationService } from '../../conversations/services/conversation.s * * SSE sequence: processing → preview → complete | error */ -export type ExtractionStreamEvent = - | { - /** The full draft ThoughtMap + nodes ready for client review */ - data: { - map: ThoughtMap; - nodes: ThoughtNode[]; - }; - type: 'preview'; - } - | { - data: { code: string; message: string; retriable: boolean }; - type: 'error'; - } - | { - data: { jobId: string; mapId: string; nodeCount: number }; - type: 'complete'; - } - | { - data: { jobId: string; status: 'started' }; - type: 'processing'; - }; +export type ExtractionStreamEvent = ExtractionStreamEventContract<{ + map: ThoughtMap; + nodes: ThoughtNode[]; +}>; /** * Job data for a map extraction request. @@ -199,13 +184,15 @@ export class MapExtractionService extends WorkerHost { * @param jobId - BullMQ job ID * @param connectionId - Unique connection identifier * @param emitFn - Callback to push events to the client - * @returns Cleanup function to call on disconnect + * @returns `close` to call on disconnect, and `ready`, which resolves once + * events published from now on will reach this connection. The job may have + * started before the client connected, so callers re-check it after `ready`. */ addConnection( jobId: string, connectionId: string, emitFn: (event: ExtractionStreamEvent) => void, - ): () => void { + ): { close: () => void; ready: Promise } { const connection: ExtractionStreamConnection = { connectionId, emitFn, @@ -213,6 +200,9 @@ export class MapExtractionService extends WorkerHost { }; const existing = this.connections.get(jobId) ?? []; + // Each connection holds one bus reference; the bus subscribes once per + // stream and every caller can wait for that subscription. + const ready = this.streamBus.track(EXTRACTION_BUS_NAMESPACE, jobId); existing.push(connection); this.connections.set(jobId, existing); @@ -220,8 +210,11 @@ export class MapExtractionService extends WorkerHost { `Extraction SSE connection added: jobId=${jobId} (connectionId=${connectionId})`, ); - return () => { - this.removeConnection(jobId, connectionId); + return { + close: () => { + this.removeConnection(jobId, connectionId); + }, + ready, }; } @@ -410,8 +403,15 @@ export class MapExtractionService extends WorkerHost { * Removes a specific SSE connection. */ removeConnection(jobId: string, connectionId: string): void { - const existing = this.connections.get(jobId) ?? []; + const existing = this.connections.get(jobId); + if (!existing) { + return; + } const filtered = existing.filter((c) => c.connectionId !== connectionId); + if (filtered.length === existing.length) { + return; + } + this.streamBus.untrack(EXTRACTION_BUS_NAMESPACE, jobId); if (filtered.length === 0) { this.connections.delete(jobId); @@ -494,7 +494,7 @@ export class MapExtractionService extends WorkerHost { } } - /** Publishes an extraction event to every instance via the stream bus. */ + /** Publishes an extraction event to the instances holding the job's stream. */ private emit(jobId: string, event: ExtractionStreamEvent): void { this.streamBus.publish(EXTRACTION_BUS_NAMESPACE, jobId, event); } diff --git a/packages/mukti-api/src/modules/thought-map/thought-map.controller.ts b/packages/mukti-api/src/modules/thought-map/thought-map.controller.ts index 544d35ea..4c2848f2 100644 --- a/packages/mukti-api/src/modules/thought-map/thought-map.controller.ts +++ b/packages/mukti-api/src/modules/thought-map/thought-map.controller.ts @@ -1,3 +1,5 @@ +import type { Job } from 'bullmq'; + import { Body, Controller, @@ -6,6 +8,7 @@ import { Get, HttpCode, HttpStatus, + Logger, MessageEvent, NotFoundException, Param, @@ -56,7 +59,12 @@ import { import { UpdateThoughtMapSettingsDto } from './dto/update-thought-map-settings.dto'; import { UpdateThoughtNodeDto } from './dto/update-thought-node.dto'; import { BranchSuggestionService } from './services/branch-suggestion.service'; -import { MapExtractionService } from './services/map-extraction.service'; +import { + type ExtractionStreamEvent, + type MapExtractionJobData, + type MapExtractionJobResult, + MapExtractionService, +} from './services/map-extraction.service'; import { ThoughtMapShareService } from './services/thought-map-share.service'; import { ThoughtMapService } from './services/thought-map.service'; @@ -71,6 +79,8 @@ import { ThoughtMapService } from './services/thought-map.service'; @Controller('thought-maps') @UseGuards(JwtAuthGuard) export class ThoughtMapController { + private readonly logger = new Logger(ThoughtMapController.name); + constructor( private readonly aiRequestGate: AiRequestGate, private readonly thoughtMapService: ThoughtMapService, @@ -473,108 +483,130 @@ export class ThoughtMapController { ); } - const jobState = await job.getState(); - - if (jobState === 'completed') { + const replay = await this.finishedExtractionEvents(job, jobId, user); + if (replay) { return new Observable((observer) => { - void (async () => { - try { - const result = job.returnvalue; - if (!result?.mapId) { - observer.next({ - data: { - data: { - code: 'EXTRACTION_ERROR', - message: 'Extraction job result missing mapId', - retriable: false, - }, - type: 'error', - }, - type: 'message', - } as MessageEvent); - observer.complete(); - return; - } - - const draftMap = await this.thoughtMapService.getMap( - result.mapId, - user._id, - ); - - observer.next({ - data: { - data: draftMap, - type: 'preview', - }, - type: 'message', - } as MessageEvent); - observer.next({ - data: { - data: { - jobId, - mapId: result.mapId, - nodeCount: result.nodeCount, - }, - type: 'complete', - }, - type: 'message', - } as MessageEvent); - } catch (error: unknown) { - observer.next({ - data: { - data: { - code: 'EXTRACTION_ERROR', - message: - error instanceof Error - ? error.message - : 'Extraction stream failed', - retriable: false, - }, - type: 'error', - }, - type: 'message', - } as MessageEvent); - } finally { - observer.complete(); - } - })(); - }); - } - - if (jobState === 'failed') { - return new Observable((observer) => { - observer.next({ - data: { - data: { - code: 'EXTRACTION_ERROR', - message: job.failedReason || 'Extraction job failed', - retriable: false, - }, - type: 'error', - }, - type: 'message', - } as MessageEvent); + for (const event of replay) { + observer.next({ data: event, type: 'message' } as MessageEvent); + } observer.complete(); }); } return new Observable((observer) => { const connectionId = `conn-${Date.now()}-${Math.random().toString(36).substring(2, 9)}`; + const seen = new Set(); + // Set by a live `complete` or by a replay, which is the final state. + let ended = false; + const deliver = (event: ExtractionStreamEvent) => { + seen.add(event.type); + observer.next({ data: event, type: 'message' } as MessageEvent); + }; - const cleanup = this.mapExtractionService.addConnection( + const connection = this.mapExtractionService.addConnection( jobId, connectionId, (event) => { - observer.next({ data: event, type: 'message' } as MessageEvent); + if (ended) { + return; + } + deliver(event); + ended = event.type === 'complete'; }, ); + // The job was queued before this stream opened. If it finished while + // this instance was subscribing, its live events are gone, so replay + // the stored result once the subscription is in place, skipping + // anything the live stream already delivered. + void connection.ready + .then(() => + ended ? null : this.mapExtractionService.getExtractionJob(jobId), + ) + .then((fresh) => + fresh ? this.finishedExtractionEvents(fresh, jobId, user) : null, + ) + .then((events) => { + if (!events || ended) { + return; + } + ended = true; + for (const event of events) { + if (!seen.has(event.type)) { + deliver(event); + } + } + }) + .catch((error: unknown) => + this.logger.warn( + `Could not re-check extraction job ${jobId} after subscribing: ${String(error)}`, + ), + ); + return () => { - cleanup(); + connection.close(); }; }); } + /** + * Events that replay a finished extraction job (`preview` then `complete`, + * or one `error`), or `null` while it is still queued or running. + */ + private async finishedExtractionEvents( + job: Job, + jobId: string, + user: User, + ): Promise { + const error = (message: string): ExtractionStreamEvent => ({ + data: { code: 'EXTRACTION_ERROR', message, retriable: false }, + type: 'error', + }); + + const state = await job.getState(); + if (state !== 'completed' && state !== 'failed') { + return null; + } + + // `getState()` asks Redis live, but `returnvalue` and `failedReason` are + // whatever was loaded with `job`. If the job finished after that load, + // the snapshot is stale (no result yet), so reload it once. + const settled = + (state === 'completed' && !job.returnvalue) || + (state === 'failed' && !job.failedReason) + ? ((await this.mapExtractionService.getExtractionJob(jobId)) ?? job) + : job; + + if (state === 'failed') { + return [error(settled.failedReason || 'Extraction job failed')]; + } + + const result = settled.returnvalue; + if (!result?.mapId) { + return [error('Extraction job result missing mapId')]; + } + + try { + const draftMap = await this.thoughtMapService.getMap( + result.mapId, + user._id, + ); + return [ + { data: draftMap, type: 'preview' }, + { + data: { jobId, mapId: result.mapId, nodeCount: result.nodeCount }, + type: 'complete', + }, + ]; + } catch (caught: unknown) { + return [ + error( + caught instanceof Error ? caught.message : 'Extraction stream failed', + ), + ]; + } + } + /** * Confirms a draft Thought Map, transitioning it to active status. * Only maps with status "draft" that are owned by the user can be confirmed. diff --git a/packages/mukti-api/test/app.e2e-spec.ts b/packages/mukti-api/test/app.e2e-spec.ts index b8b0096d..c4175cf7 100644 --- a/packages/mukti-api/test/app.e2e-spec.ts +++ b/packages/mukti-api/test/app.e2e-spec.ts @@ -1,26 +1,158 @@ -import { type INestApplication } from '@nestjs/common'; -import { Test, type TestingModule } from '@nestjs/testing'; +import type { INestApplication } from '@nestjs/common'; + +import { getConnectionToken } from '@nestjs/mongoose'; +import { mkdtempSync, rmSync } from 'fs'; +import { type Connection } from 'mongoose'; +import { tmpdir } from 'os'; +import { join } from 'path'; import * as request from 'supertest'; -import { type App } from 'supertest/types'; -import { AppModule } from './../src/app.module'; +/** + * Boots the real API the way `npx muktiai` does (`startApi`, local mode, an + * embedded MongoDB, no Redis) and drives it over HTTP. These checks cover the + * whole wiring: module graph, global guard and interceptor, bootstrap + * middleware, and the architecture fixes that cross module boundaries. + */ +describe('API (e2e, local mode)', () => { + jest.setTimeout(120_000); + + const env = { ...process.env }; + let app: INestApplication; + let dbPath: string; + let http: ReturnType; + + beforeAll(async () => { + dbPath = mkdtempSync(join(tmpdir(), 'mukti-e2e-')); + process.env.MUKTI_LOCAL = '1'; + process.env.MUKTI_LOCAL_DB_PATH = dbPath; + delete process.env.MUKTI_LOCAL_MONGO_URI; + process.env.NODE_ENV = 'test'; + process.env.FRONTEND_URL = 'http://localhost:3001'; + delete process.env.CORS_ORIGINS; + // Import after the env is set: module decorators (e.g. AuthModule's + // isLocalMode() provider list) are evaluated at import time. + const { startApi } = await import('../src/start'); + + app = await startApi({ host: '127.0.0.1', port: 0 }); + http = request(app.getHttpServer()); + }); + + afterAll(async () => { + await app?.close(); + rmSync(dbPath, { force: true, recursive: true }); + process.env = env; + }); + + describe('bootstrap', () => { + it('serves liveness under the versioned prefix', async () => { + // Health routes skip the envelope so probes get a plain body. + const res = await http.get('/api/v1/health/live').expect(200); + expect(res.body).toEqual({ status: 'ok' }); + }); + + it('reports MongoDB as up on readiness', async () => { + const res = await http.get('/api/v1/health/ready').expect(200); + expect(JSON.stringify(res.body)).toContain('"up"'); + }); + + it('sends helmet security headers', async () => { + const res = await http.get('/api/v1/health/live'); + expect(res.headers['x-content-type-options']).toBe('nosniff'); + expect(res.headers['x-frame-options']).toBeDefined(); + expect(res.headers['x-dns-prefetch-control']).toBeDefined(); + }); + + it('allows the configured web origin with credentials and exposes the CSRF header', async () => { + const res = await http + .options('/api/v1/health/live') + .set('Origin', 'http://localhost:3001') + .set('Access-Control-Request-Method', 'POST') + .set('Access-Control-Request-Headers', 'Content-Type,Authorization'); + + expect(res.headers['access-control-allow-origin']).toBe( + 'http://localhost:3001', + ); + expect(res.headers['access-control-allow-credentials']).toBe('true'); + expect(res.headers['access-control-allow-headers']).toContain( + 'Authorization', + ); + }); + + it('does not echo an unknown origin', async () => { + const res = await http + .get('/api/v1/health/live') + .set('Origin', 'https://evil.example'); + expect(res.headers['access-control-allow-origin']).toBeUndefined(); + }); + + it('wraps errors in the standard envelope', async () => { + const res = await http.get('/api/v1/does-not-exist').expect(404); + expect(res.body).toMatchObject({ success: false }); + expect(res.body.error).toBeDefined(); + }); + }); + + describe('knowledge tracing (KnowledgeStateStore)', () => { + const conceptId = 'e2e-recursion'; + + it('creates on first observation and keeps the stored parameters after', async () => { + const first = await http + .post('/api/v1/knowledge-tracing/update') + .send({ conceptId, correct: true, pInit: 0.5 }) + .expect(201); + expect(first.body.data.state.attempts).toBe(1); + expect(first.body.data.state.parameters.pInit).toBe(0.5); + + // Parameters only apply on creation; this pInit must be ignored. + const second = await http + .post('/api/v1/knowledge-tracing/update') + .send({ conceptId, correct: true, pInit: 0.1 }) + .expect(201); + expect(second.body.data.state.attempts).toBe(2); + expect(second.body.data.state.parameters.pInit).toBe(0.5); + }); + + it('records every one of many concurrent observations on one concept', async () => { + const concept = 'e2e-concurrent'; + const n = 12; + + const responses = await Promise.all( + Array.from({ length: n }, (_, i) => + http + .post('/api/v1/knowledge-tracing/update') + .send({ conceptId: concept, correct: i % 2 === 0 }), + ), + ); + // Every writer must succeed, not just the ones that won a race. + expect(responses.map((r) => r.status)).toEqual(Array(n).fill(201)); -describe('AppController (e2e)', () => { - let app: INestApplication; + // Read the record directly: no observation may be lost to a race, + // including the race to create it. + const connection = app.get(getConnectionToken()); + const docs = await connection + .collection('knowledge_states') + .find({ conceptId: concept }) + .toArray(); - beforeEach(async () => { - const moduleFixture: TestingModule = await Test.createTestingModule({ - imports: [AppModule], - }).compile(); + expect(docs).toHaveLength(1); + expect(docs[0].attempts).toBe(n); + expect(docs[0].correctAttempts).toBe(n / 2); + }); - app = moduleFixture.createNestApplication(); - await app.init(); + it('no longer exposes the removed cache-stats endpoint', async () => { + await http.get('/api/v1/knowledge-tracing/admin/cache-stats').expect(404); + }); }); - it('/ (GET)', () => { - return request(app.getHttpServer()) - .get('/') - .expect(200) - .expect('Hello World!'); + describe('thought maps read other domains through their services', () => { + it('reports a missing canvas session through the canvas domain', async () => { + const res = await http + .post( + '/api/v1/thought-maps/convert-from-canvas/507f1f77bcf86cd799439011', + ) + .send({}) + .expect(404); + expect(JSON.stringify(res.body)).toContain('Canvas session'); + }); }); }); diff --git a/packages/mukti-api/test/jest-e2e.json b/packages/mukti-api/test/jest-e2e.json index e6b2d9a2..4f5ae427 100644 --- a/packages/mukti-api/test/jest-e2e.json +++ b/packages/mukti-api/test/jest-e2e.json @@ -6,6 +6,6 @@ "transform": { "^.+\\.(t|j)s$": "ts-jest" }, - "transformIgnorePatterns": ["node_modules/(?!(@openrouter)/)"], + "transformIgnorePatterns": ["node_modules/(?!(@openrouter|@scalar|uuid)/)"], "watchman": false } diff --git a/packages/mukti-api/test/security.e2e-spec.ts b/packages/mukti-api/test/security.e2e-spec.ts deleted file mode 100644 index 87eb2618..00000000 --- a/packages/mukti-api/test/security.e2e-spec.ts +++ /dev/null @@ -1,168 +0,0 @@ -import { type INestApplication, ValidationPipe } from '@nestjs/common'; -import { ConfigService } from '@nestjs/config'; -import { Test, type TestingModule } from '@nestjs/testing'; -import * as cookieParser from 'cookie-parser'; -import helmet from 'helmet'; -import * as request from 'supertest'; -import { type App } from 'supertest/types'; - -import { AppModule } from '../src/app.module'; -import { HttpExceptionFilter } from '../src/common/filters/http-exception.filter'; - -describe('Security Configuration (e2e)', () => { - let app: INestApplication; - let configService: ConfigService; - - beforeEach(async () => { - const moduleFixture: TestingModule = await Test.createTestingModule({ - imports: [AppModule], - }).compile(); - - app = moduleFixture.createNestApplication(); - configService = app.get(ConfigService); - - // Apply the same middleware as in main.ts - const isProduction = configService.get('NODE_ENV') === 'production'; - app.use( - helmet({ - contentSecurityPolicy: isProduction - ? undefined - : { - directives: { - defaultSrc: ["'self'"], - imgSrc: ["'self'", 'data:', 'https:'], - scriptSrc: ["'self'", "'unsafe-inline'", "'unsafe-eval'"], - styleSrc: ["'self'", "'unsafe-inline'"], - }, - }, - crossOriginEmbedderPolicy: !isProduction, - crossOriginResourcePolicy: { policy: 'cross-origin' }, - }), - ); - - const cookieSecret = configService.get('SESSION_SECRET'); - app.use(cookieParser(cookieSecret)); - - app.useGlobalFilters(new HttpExceptionFilter()); - - app.useGlobalPipes( - new ValidationPipe({ - forbidNonWhitelisted: true, - forbidUnknownValues: true, - transform: true, - whitelist: true, - }), - ); - - const apiPrefix = configService.get('API_PREFIX') ?? 'api'; - app.setGlobalPrefix(apiPrefix); - - const corsOrigins = configService.get('CORS_ORIGINS'); - const frontendUrl = configService.get('FRONTEND_URL'); - - app.enableCors({ - allowedHeaders: ['Content-Type', 'Authorization', 'X-CSRF-Token'], - credentials: true, - exposedHeaders: ['X-CSRF-Token'], - maxAge: 86400, - methods: ['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS'], - origin: corsOrigins - ? corsOrigins.split(',').map((origin) => origin.trim()) - : (frontendUrl ?? 'http://localhost:3001'), - }); - - await app.init(); - }); - - afterEach(async () => { - await app.close(); - }); - - describe('Security Headers', () => { - it('should include helmet security headers', async () => { - const response = await request(app.getHttpServer()).get('/'); - - // Check for common security headers set by helmet - expect(response.headers['x-dns-prefetch-control']).toBeDefined(); - expect(response.headers['x-frame-options']).toBeDefined(); - expect(response.headers['x-content-type-options']).toBeDefined(); - expect(response.headers['x-download-options']).toBeDefined(); - }); - - it('should set X-Content-Type-Options to nosniff', async () => { - const response = await request(app.getHttpServer()).get('/'); - - expect(response.headers['x-content-type-options']).toBe('nosniff'); - }); - - it('should set X-Frame-Options', async () => { - const response = await request(app.getHttpServer()).get('/'); - - expect(response.headers['x-frame-options']).toBeDefined(); - }); - }); - - describe('CORS Configuration', () => { - it('should allow requests from configured origin', async () => { - const frontendUrl = - configService.get('FRONTEND_URL') ?? 'http://localhost:3001'; - - const response = await request(app.getHttpServer()) - .get('/') - .set('Origin', frontendUrl); - - expect(response.headers['access-control-allow-origin']).toBe(frontendUrl); - expect(response.headers['access-control-allow-credentials']).toBe('true'); - }); - - it('should expose X-CSRF-Token header', async () => { - const frontendUrl = - configService.get('FRONTEND_URL') ?? 'http://localhost:3001'; - - const response = await request(app.getHttpServer()) - .options('/') - .set('Origin', frontendUrl); - - expect(response.headers['access-control-expose-headers']).toContain( - 'X-CSRF-Token', - ); - }); - - it('should allow required headers', async () => { - const frontendUrl = - configService.get('FRONTEND_URL') ?? 'http://localhost:3001'; - - const response = await request(app.getHttpServer()) - .options('/') - .set('Origin', frontendUrl) - .set('Access-Control-Request-Method', 'POST') - .set('Access-Control-Request-Headers', 'Content-Type,Authorization'); - - expect(response.headers['access-control-allow-headers']).toBeDefined(); - }); - }); - - describe('Cookie Configuration', () => { - it('should parse cookies with cookie-parser', async () => { - // This test verifies that cookie-parser is configured - // by checking if the app can handle cookie-based authentication - const response = await request(app.getHttpServer()) - .get('/') - .set('Cookie', ['test=value']); - - // If cookie-parser is working, the request should be processed - expect(response.status).toBeDefined(); - }); - }); - - describe('CSRF Token Endpoint', () => { - it('should provide CSRF token endpoint', async () => { - const response = await request(app.getHttpServer()).get( - '/api/v1/auth/csrf-token', - ); - - expect(response.status).toBe(200); - expect(response.body).toHaveProperty('csrfToken'); - }); - }); -}); diff --git a/packages/mukti-contracts/README.md b/packages/mukti-contracts/README.md new file mode 100644 index 00000000..0fe36edc --- /dev/null +++ b/packages/mukti-contracts/README.md @@ -0,0 +1,22 @@ +# @muktiai/contracts + +Type-only wire contracts shared by `@muktiai/api` and `@muktiai/web`. Private to the +workspace and never published. + +| File | What it pins | +| ---------------------- | ----------------------------------------------------------------------------------------------------- | +| `src/stream-events.ts` | SSE events for chat, canvas/map node dialogue, branch suggestions and conversation-to-map extraction. | + +## How it is enforced + +- The API's stream services alias these types, so the server cannot emit another shape. +- The web compiles `src/types/stream-contract.check.ts`, a two-way equality check + between the client's event types and these, so drift fails the web `type-check` and + `next build`. + +## Rules + +- Types only: no runtime values, no imports from other packages. Consumers use + `import type { ... } from '@muktiai/contracts'`. +- The API compiles to CommonJS and must not need this package at runtime. Because the + imports are type-only they are erased, so the published API does not depend on it. diff --git a/packages/mukti-contracts/package.json b/packages/mukti-contracts/package.json new file mode 100644 index 00000000..3a0efffa --- /dev/null +++ b/packages/mukti-contracts/package.json @@ -0,0 +1,14 @@ +{ + "name": "@muktiai/contracts", + "version": "0.0.0", + "private": true, + "description": "Type-only wire contracts shared by the Mukti API and web app. No runtime code.", + "license": "MIT", + "types": "./src/index.ts", + "exports": { + ".": { + "types": "./src/index.ts" + } + }, + "sideEffects": false +} diff --git a/packages/mukti-contracts/project.json b/packages/mukti-contracts/project.json new file mode 100644 index 00000000..eb81fd51 --- /dev/null +++ b/packages/mukti-contracts/project.json @@ -0,0 +1,23 @@ +{ + "name": "@muktiai/contracts", + "$schema": "../../node_modules/nx/schemas/project-schema.json", + "sourceRoot": "packages/mukti-contracts/src", + "projectType": "library", + "tags": ["scope:shared", "type:contracts"], + "targets": { + "type-check": { + "executor": "nx:run-commands", + "options": { + "command": "tsc --noEmit -p tsconfig.json", + "cwd": "packages/mukti-contracts" + } + }, + "format:check": { + "executor": "nx:run-commands", + "options": { + "command": "prettier --check \"src/**/*.ts\" \"*.{json,md}\"", + "cwd": "packages/mukti-contracts" + } + } + } +} diff --git a/packages/mukti-contracts/src/index.ts b/packages/mukti-contracts/src/index.ts new file mode 100644 index 00000000..ced34ddc --- /dev/null +++ b/packages/mukti-contracts/src/index.ts @@ -0,0 +1,7 @@ +/** + * Wire contracts between the Mukti API and web app. + * + * Type-only: import with `import type`. There is no runtime code here, so the + * package needs no build step and adds nothing to either bundle. + */ +export type * from './stream-events'; diff --git a/packages/mukti-api/src/contracts/stream-events.ts b/packages/mukti-contracts/src/stream-events.ts similarity index 57% rename from packages/mukti-api/src/contracts/stream-events.ts rename to packages/mukti-contracts/src/stream-events.ts index 742aab4e..7b613a11 100644 --- a/packages/mukti-api/src/contracts/stream-events.ts +++ b/packages/mukti-contracts/src/stream-events.ts @@ -38,6 +38,44 @@ export type NodeDialogueStreamEventContract = { [T in StreamEventType]: StreamEventOf; }[StreamEventType]; +// --------------------------------------------------------------------------- +// Thought-map job streams. These carry no envelope fields and have their own +// event sets, so they are listed out rather than built from the map above. +// --------------------------------------------------------------------------- + +/** One AI-suggested branch (`/thought-maps/:id/suggest/stream`). */ +export interface BranchSuggestionPayload { + /** Always phrased as a question. */ + label: string; + /** nodeId of the parent the branch would grow from. */ + parentId: string; + suggestedType: 'question' | 'thought'; +} + +/** Branch-suggestion stream: `processing → suggestion* → complete | error`. */ +export type SuggestionStreamEventContract = + | { data: BranchSuggestionPayload; type: 'suggestion' } + | { data: StreamEventPayloads['error']; type: 'error' } + | { data: StreamEventPayloads['processing']; type: 'processing' } + | { data: { jobId: string; suggestionCount: number }; type: 'complete' }; + +/** + * Conversation-to-map extraction stream: + * `processing → preview → complete | error`. + * + * @typeParam Preview - The draft map payload. The API sends stored documents + * (`_id`, `parentId`) and the web reshapes them, so each side supplies its own + * type here; the contract pins the event set and every other payload. + */ +export type ExtractionStreamEventContract = + | { data: Preview; type: 'preview' } + | { data: StreamEventPayloads['error']; type: 'error' } + | { data: StreamEventPayloads['processing']; type: 'processing' } + | { + data: { jobId: string; mapId: string; nodeCount: number }; + type: 'complete'; + }; + /** One event of type `T`, with the envelope fields a stream adds. */ export type StreamEventOf = Envelope & { data: StreamEventPayloads[T]; @@ -79,9 +117,4 @@ export interface StreamEventPayloads { } /** Event types every live stream may carry, in emission order. */ -export type StreamEventType = - | 'complete' - | 'error' - | 'message' - | 'processing' - | 'progress'; +export type StreamEventType = 'complete' | 'error' | 'message' | 'processing' | 'progress'; diff --git a/packages/mukti-contracts/tsconfig.json b/packages/mukti-contracts/tsconfig.json new file mode 100644 index 00000000..732b6819 --- /dev/null +++ b/packages/mukti-contracts/tsconfig.json @@ -0,0 +1,13 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "ESNext", + "moduleResolution": "bundler", + "strict": true, + "noEmit": true, + "skipLibCheck": true, + "isolatedModules": true, + "verbatimModuleSyntax": true + }, + "include": ["src/**/*.ts"] +} diff --git a/packages/mukti-mcp/CONTRIBUTING.md b/packages/mukti-mcp/CONTRIBUTING.md index c41d6bc1..af4cb354 100644 --- a/packages/mukti-mcp/CONTRIBUTING.md +++ b/packages/mukti-mcp/CONTRIBUTING.md @@ -29,7 +29,7 @@ Mukti is a Model Context Protocol (MCP) server that implements Socratic question ```bash git clone https://github.com/shettydev/mukti.git - cd mukti/mukti-mcp-server + cd mukti/packages/mukti-mcp ``` 2. **Install Dependencies** @@ -58,7 +58,7 @@ Mukti is a Model Context Protocol (MCP) server that implements Socratic question ### Project Structure ``` -mukti-mcp-server/ +packages/mukti-mcp/ ├── src/ │ ├── index.ts # Main MCP server implementation │ ├── types.ts # Type definitions diff --git a/packages/mukti-mcp/README.md b/packages/mukti-mcp/README.md index 670cec57..dd58ab8c 100644 --- a/packages/mukti-mcp/README.md +++ b/packages/mukti-mcp/README.md @@ -38,7 +38,7 @@ The Mukti MCP Server is designed around a core belief: **the process of inquiry 2. **Install dependencies**: ```bash - cd mukti-mcp-server + cd packages/mukti-mcp npm install ``` diff --git a/packages/mukti-web/Dockerfile b/packages/mukti-web/Dockerfile index 7389f29f..e610ced2 100644 --- a/packages/mukti-web/Dockerfile +++ b/packages/mukti-web/Dockerfile @@ -6,6 +6,7 @@ FROM base AS deps COPY package.json bun.lock ./ COPY packages/mukti-api/package.json packages/mukti-api/ COPY packages/mukti-web/package.json packages/mukti-web/ +COPY packages/mukti-contracts/package.json packages/mukti-contracts/ RUN bun install # Stage 2: Build diff --git a/packages/mukti-web/src/lib/api/thought-map.ts b/packages/mukti-web/src/lib/api/thought-map.ts index 42cce117..da942810 100644 --- a/packages/mukti-web/src/lib/api/thought-map.ts +++ b/packages/mukti-web/src/lib/api/thought-map.ts @@ -11,6 +11,8 @@ * - Response transformation from backend format to frontend types (_id → id) */ +import type { BranchSuggestionPayload, SuggestionStreamEventContract } from '@muktiai/contracts'; + import { config } from '@/lib/config'; import { type ConvertCanvasRequest, @@ -35,16 +37,10 @@ import { apiClient } from './client'; // ============================================================================ /** - * A single AI-generated branch suggestion delivered via SSE. + * A branch-suggestion SSE event. The wire shape is the API's shared contract, + * so `event.type` narrows `event.data` without casts. */ -export interface BranchSuggestionEvent { - data: - | { code: string; message: string; retriable: boolean } - | { jobId: string; status: string } - | { jobId: string; suggestionCount: number } - | { label: string; parentId: string; suggestedType: 'question' | 'thought' }; - type: 'complete' | 'error' | 'processing' | 'suggestion'; -} +export type BranchSuggestionEvent = SuggestionStreamEventContract; /** * Request body for POST /thought-maps/:id/suggest @@ -67,7 +63,7 @@ export interface SuggestionJobStatusResponse { count: number; mapId: string; parentNodeId: string; - suggestions: { label: string; parentId: string; suggestedType: 'question' | 'thought' }[]; + suggestions: BranchSuggestionPayload[]; }; state: string; } diff --git a/packages/mukti-web/src/lib/hooks/use-thought-map-extraction.ts b/packages/mukti-web/src/lib/hooks/use-thought-map-extraction.ts index 95c0bc70..b79c8725 100644 --- a/packages/mukti-web/src/lib/hooks/use-thought-map-extraction.ts +++ b/packages/mukti-web/src/lib/hooks/use-thought-map-extraction.ts @@ -111,8 +111,7 @@ export function useThoughtMapExtraction(conversationId: string) { break; case 'error': { - const errData = event.data as { message: string }; - setState({ error: new Error(errData.message), isStreaming: false }); + setState({ error: new Error(event.data.message), isStreaming: false }); resetExtraction(); break; } diff --git a/packages/mukti-web/src/lib/hooks/use-thought-map-suggestions.ts b/packages/mukti-web/src/lib/hooks/use-thought-map-suggestions.ts index 3d807d41..5cd45839 100644 --- a/packages/mukti-web/src/lib/hooks/use-thought-map-suggestions.ts +++ b/packages/mukti-web/src/lib/hooks/use-thought-map-suggestions.ts @@ -115,16 +115,7 @@ export function useColdStartSuggestions(mapId: string, rootNodeId: string, enabl cleanupPolling = startSuggestionPolling(mapId, jobId, (event) => { if (event.type === 'suggestion') { - const s = event.data as { - label: string; - parentId: string; - suggestedType: 'question' | 'thought'; - }; - addGhostNode({ - label: s.label, - parentId: s.parentId, - suggestedType: s.suggestedType, - }); + addGhostNode(event.data); } }); } catch { @@ -267,25 +258,13 @@ export function useThoughtMapSuggestions(mapId: string, parentNodeId: string) { setState({ error: null, isStreaming: false }); break; - case 'error': { - const errData = event.data as { message: string }; - setState({ error: new Error(errData.message), isStreaming: false }); + case 'error': + setState({ error: new Error(event.data.message), isStreaming: false }); break; - } - case 'suggestion': { - const s = event.data as { - label: string; - parentId: string; - suggestedType: 'question' | 'thought'; - }; - addGhostNode({ - label: s.label, - parentId: s.parentId, - suggestedType: s.suggestedType, - }); + case 'suggestion': + addGhostNode(event.data); break; - } default: break; diff --git a/packages/mukti-web/src/lib/stores/thought-map-store.ts b/packages/mukti-web/src/lib/stores/thought-map-store.ts index 0c48f5aa..faec9813 100644 --- a/packages/mukti-web/src/lib/stores/thought-map-store.ts +++ b/packages/mukti-web/src/lib/stores/thought-map-store.ts @@ -11,6 +11,8 @@ * - ghostNodes stores AI-suggested ghost nodes keyed by a client-side ghost ID */ +import type { BranchSuggestionPayload } from '@muktiai/contracts'; + import { create } from 'zustand'; import { useShallow } from 'zustand/react/shallow'; @@ -56,8 +58,8 @@ export interface GhostNode { label: string; /** nodeId of the parent node this branches from */ parentId: string; - /** Suggested node type */ - suggestedType: 'question' | 'thought'; + /** Suggested node type (from the shared suggestion contract) */ + suggestedType: BranchSuggestionPayload['suggestedType']; } /** diff --git a/packages/mukti-web/src/types/stream-contract.check.ts b/packages/mukti-web/src/types/stream-contract.check.ts index 4dbfebb3..aeae4934 100644 --- a/packages/mukti-web/src/types/stream-contract.check.ts +++ b/packages/mukti-web/src/types/stream-contract.check.ts @@ -3,16 +3,21 @@ * wire contract. This file has no runtime code; it only fails `tsc` (the web * `type-check` target) if the two sides drift apart. * - * The contract lives in the API package - * (`mukti-api/src/contracts/stream-events.ts`) and is type-only, so importing - * it here pulls nothing into the bundle. + * The contract lives in the `@muktiai/contracts` workspace package and is + * type-only, so importing it here pulls nothing into the bundle. */ import type { + BranchSuggestionPayload, ConversationStreamEventContract, + ExtractionStreamEventContract, NodeDialogueStreamEventContract, -} from '../../../mukti-api/src/contracts/stream-events'; + SuggestionStreamEventContract, +} from '@muktiai/contracts'; + +import type { BranchSuggestionEvent } from '../lib/api/thought-map'; import type { StreamEvent } from '../lib/hooks/use-conversation-stream'; import type { DialogueStreamEvent } from './dialogue.types'; +import type { ExtractionStreamEvent, ThoughtMapWithNodes } from './thought-map'; export type ConversationStreamHasNoExtraTypes = Assert< Same @@ -24,6 +29,12 @@ export type ConversationStreamMatchesApi = Assert< AllTrue> >; +// The web reshapes the extraction preview (`_id` → `id`), so the contract is +// instantiated with the client's own preview type; every other event must match. +export type ExtractionStreamMatchesApi = Assert< + Same> +>; + export type NodeDialogueStreamHasNoExtraTypes = Assert< Same >; @@ -32,6 +43,17 @@ export type NodeDialogueStreamMatchesApi = Assert< AllTrue> >; +// The web derives its suggestion types from the contract, so the check above +// cannot see the API widening or narrowing `suggestedType`. Pin it to the node +// types the client knows how to render and create. +export type SuggestedTypeIsRenderable = Assert< + Same +>; + +export type SuggestionStreamMatchesApi = Assert< + Same +>; + type AllTrue = R[keyof R] extends true ? true : false; /** Fails to compile unless its argument is `true`. */ type Assert = T;