From 038679d4bb32de1459e165de80944a3a8ab0c022 Mon Sep 17 00:00:00 2001 From: Prathik Shetty Date: Mon, 19 Oct 2026 23:54:27 +0530 Subject: [PATCH 1/8] chore(deps): Sync bun.lock with ioredis and released package versions --- bun.lock | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/bun.lock b/bun.lock index 1e07fc4..a891550 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", }, @@ -154,7 +155,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", From 70b8fdf08f9fe57380ce164bfb499bdd9218e093 Mon Sep 17 00:00:00 2001 From: Prathik Shetty Date: Wed, 21 Oct 2026 20:03:11 +0530 Subject: [PATCH 2/8] docs: Bring architecture docs in line with the code - Rewrite docs/reference/architecture/overview.md from the code: packages, hosted vs local, API module map with the gate/turn/store/bus rules, and a reply sequence diagram. packages/ARCHITECTURE.md now points there and labels the old PNG as aspirational. - CLAUDE.md: real module tree and web routes, architecture rules, Redis roles, refresh tokens are not rotated, MCP at packages/mukti-mcp. - Knowledge-tracing README: replace the removed cache with the KnowledgeStateStore consistency model. - Drop the removed cache-stats endpoint from RFC-0001 and RFC-0007; add a code-drift note to the RFC-0007 design. - Fix broken mukti-mcp-server links in README, RFC-0001 and the MCP package docs. --- CLAUDE.md | 93 +++++----- README.md | 8 +- .../rfc-0007-learner-readings-judge-port.md | 22 ++- docs/reference/architecture/overview.md | 160 +++++++++++++----- .../rfc-0001-knowledge-gap-detection/index.md | 5 +- packages/ARCHITECTURE.md | 47 +---- .../src/modules/knowledge-tracing/README.md | 50 ++---- packages/mukti-mcp/CONTRIBUTING.md | 4 +- packages/mukti-mcp/README.md | 2 +- 9 files changed, 224 insertions(+), 167 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 6aa78bd..987ca45 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -15,20 +15,24 @@ 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-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 +58,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 +82,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. +- **Read another domain's data through its service**, not its model (e.g. thought maps use `ConversationService`, `CanvasService`). +- **SSE event shapes live in `src/contracts/stream-events.ts`.** 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 +134,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 +151,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 +194,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 561cb90..0177088 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/docs/designs/rfc-0007-learner-readings-judge-port.md b/docs/designs/rfc-0007-learner-readings-judge-port.md index 521c215..a78babd 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 b024bd0..68529d5 100644 --- a/docs/reference/architecture/overview.md +++ b/docs/reference/architecture/overview.md @@ -1,46 +1,118 @@ # 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) + 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). Every instance delivers to the connections it holds. + +## Contracts between web and API + +The SSE event shapes for chat and node chat are defined once in +`packages/mukti-api/src/contracts/stream-events.ts`. The API's stream services use it +directly, and the web compiles a two-way equality check against it +(`packages/mukti-web/src/types/stream-contract.check.ts`), so drift fails `tsc` and +`next build`. 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 f0d0636..60bfc05 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 3b98c30..6ae7496 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/src/modules/knowledge-tracing/README.md b/packages/mukti-api/src/modules/knowledge-tracing/README.md index 7b79187..da8f530 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 re-reads and re-applies, up to 5 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-mcp/CONTRIBUTING.md b/packages/mukti-mcp/CONTRIBUTING.md index c41d6bc..af4cb35 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 670cec5..dd58ab8 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 ``` From 48b8ba8d0adde8ffda771fd2cba5b43f96f52197 Mon Sep 17 00:00:00 2001 From: Prathik Shetty Date: Wed, 21 Oct 2026 20:24:20 +0530 Subject: [PATCH 3/8] test(api): Fix failing specs and replace dead e2e with a real-app suite - thought-map dialogue queue spec: expect the scaffold-state guard argument, and test model validation through process() now that it lives in SocraticTurnEngine. - e2e: the old suites never ran (ESM uuid/@scalar not transformed) and asserted a removed GET / controller. New app.e2e-spec boots startApi in local mode on an embedded MongoDB and checks the versioned prefix, health, helmet, CORS, the error envelope, the removed cache-stats route, thought-map->canvas via service, and that 12 concurrent knowledge updates on one concept all land. - That concurrency test found the CAS retry cap of 5 was too low under burst load (11/12 landed). KnowledgeStateStore now backs off with jitter, allows 20 conflicts, and does not count record creation against the budget. - CI runs the e2e target in the quality gate. --- .github/workflows/ci.yml | 5 + packages/mukti-api/project.json | 2 +- .../src/modules/knowledge-tracing/README.md | 2 +- .../knowledge-state-store.service.spec.ts | 22 ++- .../services/knowledge-state-store.service.ts | 21 ++- ...thought-map-dialogue-queue.service.spec.ts | 47 ++++- packages/mukti-api/test/app.e2e-spec.ts | 165 +++++++++++++++-- packages/mukti-api/test/jest-e2e.json | 2 +- packages/mukti-api/test/security.e2e-spec.ts | 168 ------------------ 9 files changed, 235 insertions(+), 199 deletions(-) delete mode 100644 packages/mukti-api/test/security.e2e-spec.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0c34fbf..2db7b23 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/packages/mukti-api/project.json b/packages/mukti-api/project.json index e3dcea7..6f83fa3 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/modules/knowledge-tracing/README.md b/packages/mukti-api/src/modules/knowledge-tracing/README.md index da8f530..205d48f 100644 --- a/packages/mukti-api/src/modules/knowledge-tracing/README.md +++ b/packages/mukti-api/src/modules/knowledge-tracing/README.md @@ -51,7 +51,7 @@ flowchart TD ### Consistency Strategy - **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 re-reads and re-applies, up to 5 times. +- **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. 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 50f0b90..0ffafd8 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 b2cced1..9aa925e 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/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 ff00137..4f05ead 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/test/app.e2e-spec.ts b/packages/mukti-api/test/app.e2e-spec.ts index b8b0096..f0a3675 100644 --- a/packages/mukti-api/test/app.e2e-spec.ts +++ b/packages/mukti-api/test/app.e2e-spec.ts @@ -1,26 +1,157 @@ -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'; +import { startApi } from '../src/start'; + +/** + * 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; + + 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 e6b2d9a..4f5ae42 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 87eb261..0000000 --- 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'); - }); - }); -}); From e0af985b3dcba49d1c3d00126e8537357f9bb80b Mon Sep 17 00:00:00 2001 From: Prathik Shetty Date: Wed, 21 Oct 2026 20:39:11 +0530 Subject: [PATCH 4/8] refactor(architecture): Share every SSE stream contract via private @muktiai/contracts New type-only workspace package packages/mukti-contracts (private, never published, Nx type-check target). API and web import it with import type, so it is erased from runtime JS. Contracts now also pin the branch-suggestion and extraction streams. API services alias them. The web drops its looser suggestion type and the casts that hid drift from tsc, and its compile-time check asserts both streams plus that suggestedType stays renderable. Mutation-tested: adding, removing or renaming a payload field, or widening or narrowing suggestedType, each fails web tsc. The published @muktiai/api ships only dist/start.d.ts, its public entry, so internal declarations never reference the private package (verified with npm pack). Dockerfiles copy the new manifest so bun links it. Docs updated. --- CLAUDE.md | 3 +- bun.lock | 6 +++ docs/reference/architecture/overview.md | 10 +++-- packages/mukti-api/Dockerfile | 2 + packages/mukti-api/package.json | 4 +- .../conversations/services/stream.service.ts | 4 +- .../services/dialogue-stream.service.ts | 4 +- .../services/branch-suggestion.service.ts | 36 ++++----------- .../services/map-extraction.service.ts | 27 +++-------- packages/mukti-contracts/README.md | 22 +++++++++ packages/mukti-contracts/package.json | 14 ++++++ packages/mukti-contracts/project.json | 23 ++++++++++ packages/mukti-contracts/src/index.ts | 7 +++ .../src}/stream-events.ts | 45 ++++++++++++++++--- packages/mukti-contracts/tsconfig.json | 13 ++++++ packages/mukti-web/Dockerfile | 1 + packages/mukti-web/src/lib/api/thought-map.ts | 16 +++---- .../lib/hooks/use-thought-map-extraction.ts | 3 +- .../lib/hooks/use-thought-map-suggestions.ts | 31 +++---------- .../src/lib/stores/thought-map-store.ts | 6 ++- .../src/types/stream-contract.check.ts | 30 +++++++++++-- 21 files changed, 199 insertions(+), 108 deletions(-) create mode 100644 packages/mukti-contracts/README.md create mode 100644 packages/mukti-contracts/package.json create mode 100644 packages/mukti-contracts/project.json create mode 100644 packages/mukti-contracts/src/index.ts rename packages/{mukti-api/src/contracts => mukti-contracts/src}/stream-events.ts (57%) create mode 100644 packages/mukti-contracts/tsconfig.json diff --git a/CLAUDE.md b/CLAUDE.md index 987ca45..cc78eac 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -25,6 +25,7 @@ Key features: ├── 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) @@ -105,7 +106,7 @@ SocraticTurnModule → ScaffoldingModule → KnowledgeTracingModule, DialogueQua - **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. - **Read another domain's data through its service**, not its model (e.g. thought maps use `ConversationService`, `CanvasService`). -- **SSE event shapes live in `src/contracts/stream-events.ts`.** The web type-checks against it; change both sides together. +- **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 diff --git a/bun.lock b/bun.lock index a891550..c6eb418 100644 --- a/bun.lock +++ b/bun.lock @@ -139,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", @@ -890,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/reference/architecture/overview.md b/docs/reference/architecture/overview.md index 68529d5..3c0f9f2 100644 --- a/docs/reference/architecture/overview.md +++ b/docs/reference/architecture/overview.md @@ -111,8 +111,10 @@ in-process when local). Every instance delivers to the connections it holds. ## Contracts between web and API -The SSE event shapes for chat and node chat are defined once in -`packages/mukti-api/src/contracts/stream-events.ts`. The API's stream services use it -directly, and the web compiles a two-way equality check against it +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`. +`next build`. The package is never published: imports are `import type`, so they +vanish from the API's runtime JavaScript. diff --git a/packages/mukti-api/Dockerfile b/packages/mukti-api/Dockerfile index 72110db..9fe7ff9 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 0e3695e..5cba968 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/src/modules/conversations/services/stream.service.ts b/packages/mukti-api/src/modules/conversations/services/stream.service.ts index afdc8db..deaaf65 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, 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 45193cd..97105f6 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, 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 446533b..2c5d8d9 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 711252c..5128054 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. diff --git a/packages/mukti-contracts/README.md b/packages/mukti-contracts/README.md new file mode 100644 index 0000000..0fe36ed --- /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 0000000..3a0efff --- /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 0000000..eb81fd5 --- /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 0000000..ced34dd --- /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 742aab4..7b613a1 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 0000000..732b681 --- /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-web/Dockerfile b/packages/mukti-web/Dockerfile index 7389f29..e610ced 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 42cce11..da94281 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 95c0bc7..b79c872 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 3d807d4..5cd4583 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 0c48f5a..faec981 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 4dbfebb..aeae493 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; From 9c6eb0fae3fabbf9da5d59919f820eb03eba3a6c Mon Sep 17 00:00:00 2001 From: Prathik Shetty Date: Wed, 21 Oct 2026 20:50:05 +0530 Subject: [PATCH 5/8] feat(redis): Route stream bus events over per-stream Redis channels Each stream now publishes on its own channel (mukti:stream:{namespace}:{key}) instead of the shared mukti:stream-events, so an event reaches only instances that hold a connection for that stream rather than every instance. StreamBus gains track/untrack. The chat, dialogue and extraction stream services call track on their first local connection for a key and untrack when the last one closes (remove or cleanup). RedisStreamBus reference-counts these and subscribes or unsubscribes the channel; the in-process bus ignores them. Order within a job is unchanged: one publisher per process and Redis keeps per-publisher, per-channel order. The fake Redis in the spec now honours channel subscriptions, and a new extraction cross-instance test closes a coverage gap. Removing any single track or untrack call fails the spec. Checked against a real redis-server: 201 events in order, no leakage to other instances, unsubscribe on last close, and resubscribe after the server kills pub/sub clients. --- CLAUDE.md | 2 +- docs/reference/architecture/overview.md | 8 +- .../stream-bus/__tests__/stream-bus.spec.ts | 181 ++++++++++++++++-- .../src/core/stream-bus/stream-bus.ts | 107 ++++++++--- .../conversations/services/stream.service.ts | 7 + .../services/dialogue-stream.service.ts | 6 + .../services/map-extraction.service.ts | 12 +- 7 files changed, 284 insertions(+), 39 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index cc78eac..c69a78a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -104,7 +104,7 @@ SocraticTurnModule → ScaffoldingModule → KnowledgeTracingModule, DialogueQua - **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. +- **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. diff --git a/docs/reference/architecture/overview.md b/docs/reference/architecture/overview.md index 3c0f9f2..ec16bde 100644 --- a/docs/reference/architecture/overview.md +++ b/docs/reference/architecture/overview.md @@ -94,6 +94,7 @@ sequenceDiagram 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 @@ -107,7 +108,12 @@ sequenceDiagram 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). Every instance delivers to the connections it holds. +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. ## Contracts between web and 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 c9c254f..eb3fc2c 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,142 @@ 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 close = api.addConnection('job-1', 'conn-1', emitFn); + await flush(); + + ( + 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' }), + ); + close(); + await flush(); + expect(server.subscriberCount(streamChannel('extraction', 'job-1'))).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 3cfdec1..79ad5c7 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,23 @@ 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. */ + abstract track(namespace: string, key: string): void; + + /** Releases one {@link track}. */ + abstract untrack(namespace: string, key: string): void; } /** @@ -53,17 +63,32 @@ export class InProcessStreamBus extends StreamBus { handler(key, event); } } + + /** One process sees every stream, so there is nothing to subscribe to. */ + track(): void { + return; + } + + /** 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 +98,22 @@ 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. Ioredis re-subscribes tracked channels on reconnect. */ @Injectable() export class RedisStreamBus extends StreamBus implements OnModuleDestroy { @@ -92,22 +121,18 @@ export class RedisStreamBus extends StreamBus implements OnModuleDestroy { private readonly logger = new Logger(RedisStreamBus.name); + /** 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 +152,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 +160,44 @@ export class RedisStreamBus extends StreamBus implements OnModuleDestroy { ); } + track(namespace: string, key: string): void { + const channel = streamChannel(namespace, key); + const count = this.tracked.get(channel) ?? 0; + this.tracked.set(channel, count + 1); + + if (count === 0) { + void this.redis.subscriber + .subscribe(channel) + .catch((error: unknown) => + this.logger.error( + `Stream bus could not subscribe to ${channel}; this client will miss live updates: ${String(error)}`, + ), + ); + } + } + + 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); + 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 deaaf65..fde0162 100644 --- a/packages/mukti-api/src/modules/conversations/services/stream.service.ts +++ b/packages/mukti-api/src/modules/conversations/services/stream.service.ts @@ -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) { + 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 97105f6..33429d4 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 @@ -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) { + 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/thought-map/services/map-extraction.service.ts b/packages/mukti-api/src/modules/thought-map/services/map-extraction.service.ts index 5128054..17546ad 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 @@ -198,6 +198,10 @@ export class MapExtractionService extends WorkerHost { }; const existing = this.connections.get(jobId) ?? []; + // First local connection for this stream: start receiving its events. + if (existing.length === 0) { + this.streamBus.track(EXTRACTION_BUS_NAMESPACE, jobId); + } existing.push(connection); this.connections.set(jobId, existing); @@ -395,11 +399,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 === 0) { this.connections.delete(jobId); + this.streamBus.untrack(EXTRACTION_BUS_NAMESPACE, jobId); } else { this.connections.set(jobId, filtered); } @@ -479,7 +487,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); } From 047f9dae6c7e53307fb43d02e5f9711001b491cd Mon Sep 17 00:00:00 2001 From: Prathik Shetty Date: Wed, 21 Oct 2026 21:16:20 +0530 Subject: [PATCH 6/8] fix(api): Replay extraction results that finish while the stream subscribes With per-stream channels an instance receives a stream's events only after its SUBSCRIBE lands. Map extraction jobs are enqueued before the client opens the SSE, so a fast job could publish preview and complete in that gap and the client would hang with no result. Reproduced against a real redis-server: the client received nothing. StreamBus.track now returns a promise that resolves once the subscription is in place (one shared subscribe per stream; resolves even if the subscribe fails, after logging). MapExtractionService.addConnection returns { close, ready } and each connection holds its own bus reference. The extraction SSE waits for ready, re-fetches the job, and if it completed or failed in the meantime replays the stored result, skipping event types already delivered live and dropping live events after the replay. The finished-job replay used by late subscribers is now one helper shared with this path. Chat and dialogue are unchanged: their jobs are enqueued after the stream is open, and suggestions poll BullMQ. Tests: controller race cases (completed, failed, still running, live complete first, live preview then replay) and bus readiness cases. Removing the re-check, the dedupe or the end guard each fails the suite. The real-Redis repro now delivers preview and complete. --- docs/reference/architecture/overview.md | 6 + .../stream-bus/__tests__/stream-bus.spec.ts | 72 ++++++- .../src/core/stream-bus/stream-bus.ts | 47 +++-- .../conversations/services/stream.service.ts | 2 +- .../services/dialogue-stream.service.ts | 2 +- .../__tests__/thought-map.controller.spec.ts | 118 ++++++++++- .../__tests__/map-extraction.service.spec.ts | 4 +- .../services/map-extraction.service.ts | 25 ++- .../thought-map/thought-map.controller.ts | 190 ++++++++++-------- 9 files changed, 351 insertions(+), 115 deletions(-) diff --git a/docs/reference/architecture/overview.md b/docs/reference/architecture/overview.md index ec16bde..49f15e1 100644 --- a/docs/reference/architecture/overview.md +++ b/docs/reference/architecture/overview.md @@ -115,6 +115,12 @@ unsubscribes when it loses its last, so events reach only the instances that hol 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, 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 eb3fc2c..ccefc97 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 @@ -269,8 +269,8 @@ describe('StreamBus', () => { workerInstance, ); const emitFn = jest.fn(); - const close = api.addConnection('job-1', 'conn-1', emitFn); - await flush(); + const connection = api.addConnection('job-1', 'conn-1', emitFn); + await connection.ready; ( worker as unknown as { @@ -282,13 +282,79 @@ describe('StreamBus', () => { expect(emitFn).toHaveBeenCalledWith( expect.objectContaining({ type: 'complete' }), ); - close(); + 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'); 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 79ad5c7..3565b7a 100644 --- a/packages/mukti-api/src/core/stream-bus/stream-bus.ts +++ b/packages/mukti-api/src/core/stream-bus/stream-bus.ts @@ -37,8 +37,14 @@ export abstract class StreamBus { /** 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. */ - abstract track(namespace: string, key: string): 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; @@ -65,8 +71,8 @@ export class InProcessStreamBus extends StreamBus { } /** One process sees every stream, so there is nothing to subscribe to. */ - track(): void { - return; + track(): Promise { + return Promise.resolve(); } /** See {@link track}. */ @@ -113,7 +119,10 @@ interface Envelope { * * 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. Ioredis re-subscribes tracked channels on reconnect. + * 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 { @@ -121,6 +130,9 @@ 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(); @@ -160,20 +172,26 @@ export class RedisStreamBus extends StreamBus implements OnModuleDestroy { ); } - track(namespace: string, key: string): void { + track(namespace: string, key: string): Promise { const channel = streamChannel(namespace, key); const count = this.tracked.get(channel) ?? 0; this.tracked.set(channel, count + 1); - if (count === 0) { - void this.redis.subscriber - .subscribe(channel) - .catch((error: unknown) => - this.logger.error( - `Stream bus could not subscribe to ${channel}; this client will miss live updates: ${String(error)}`, - ), - ); + 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 { @@ -189,6 +207,7 @@ export class RedisStreamBus extends StreamBus implements OnModuleDestroy { } this.tracked.delete(channel); + this.subscribed.delete(channel); void this.redis.subscriber .unsubscribe(channel) .catch((error: unknown) => 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 fde0162..6c86661 100644 --- a/packages/mukti-api/src/modules/conversations/services/stream.service.ts +++ b/packages/mukti-api/src/modules/conversations/services/stream.service.ts @@ -125,7 +125,7 @@ export class StreamService { // First local connection for this stream: start receiving its events. if (existingConnections.length === 0) { - this.bus.track(BUS_NAMESPACE, conversationId); + void this.bus.track(BUS_NAMESPACE, conversationId); } // Add new connection to the array 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 33429d4..5289c22 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 @@ -79,7 +79,7 @@ export class DialogueStreamService { const existingConnections = this.connections.get(key) ?? []; // First local connection for this stream: start receiving its events. if (existingConnections.length === 0) { - this.bus.track(BUS_NAMESPACE, key); + void this.bus.track(BUS_NAMESPACE, key); } existingConnections.push(connection); this.connections.set(key, existingConnections); 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 9f5ba57..202e0b5 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,118 @@ 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('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 0c4f3b8..71de04a 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/map-extraction.service.ts b/packages/mukti-api/src/modules/thought-map/services/map-extraction.service.ts index 17546ad..baab162 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 @@ -184,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, @@ -198,10 +200,9 @@ export class MapExtractionService extends WorkerHost { }; const existing = this.connections.get(jobId) ?? []; - // First local connection for this stream: start receiving its events. - if (existing.length === 0) { - this.streamBus.track(EXTRACTION_BUS_NAMESPACE, 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); @@ -209,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, }; } @@ -404,10 +408,13 @@ export class MapExtractionService extends WorkerHost { 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); - this.streamBus.untrack(EXTRACTION_BUS_NAMESPACE, jobId); } else { this.connections.set(jobId, filtered); } 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 544d35e..3e9de60 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,120 @@ export class ThoughtMapController { ); } - const jobState = await job.getState(); - - if (jobState === 'completed') { - 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') { + const replay = await this.finishedExtractionEvents(job, jobId, user); + if (replay) { 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 === 'failed') { + return [error(job.failedReason || 'Extraction job failed')]; + } + if (state !== 'completed') { + return null; + } + + const result = job.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. From fa6592e18d94063ecc03601a0e0dcefee99185b8 Mon Sep 17 00:00:00 2001 From: Prathik Shetty Date: Wed, 21 Oct 2026 21:41:38 +0530 Subject: [PATCH 7/8] fix(api): Reload extraction jobs that finish after they were loaded The extraction SSE re-check (047f9da) loads the BullMQ job, then asks getState(). getState() reads Redis live, but returnvalue and failedReason are the snapshot taken by getJob(). A job that completes between the two reports "completed" with no returnvalue, and the client got "Extraction job result missing mapId" instead of its map. Found by a hosted acceptance run: two real API processes (node dist/main) sharing MongoDB and Redis, instance A behind a 40 ms Redis link, jobs enqueued on B and streamed from A. 2 of 3 extractions failed that way before this change; 9 of 9 succeed after it across three runs. The replay helper now reloads the job once when the state is final but the snapshot has no result yet. Covered by a controller test that fails on the previous code. --- .../__tests__/thought-map.controller.spec.ts | 26 +++++++++++++++++++ .../thought-map/thought-map.controller.ts | 20 ++++++++++---- 2 files changed, 41 insertions(+), 5 deletions(-) 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 202e0b5..633b058 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 @@ -799,6 +799,32 @@ describe('ThoughtMapController', () => { 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 () => { 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 3e9de60..4c2848f 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 @@ -564,14 +564,24 @@ export class ThoughtMapController { }); const state = await job.getState(); - if (state === 'failed') { - return [error(job.failedReason || 'Extraction job failed')]; - } - if (state !== 'completed') { + if (state !== 'completed' && state !== 'failed') { return null; } - const result = job.returnvalue; + // `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')]; } From 609020ae8bb0c689a148653a1dcaa20242496e79 Mon Sep 17 00:00:00 2001 From: Prathik Shetty Date: Thu, 22 Oct 2026 22:23:09 +0530 Subject: [PATCH 8/8] test(e2e): Import startApi after setting local-mode env so CI boots without OAuth secrets AuthModule picks its providers from isLocalMode() at import time. The static import ran before beforeAll set MUKTI_LOCAL, so GoogleStrategy was registered and threw in CI, where there are no Google credentials. Locally, .env hid this. --- packages/mukti-api/test/app.e2e-spec.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/mukti-api/test/app.e2e-spec.ts b/packages/mukti-api/test/app.e2e-spec.ts index f0a3675..c4175cf 100644 --- a/packages/mukti-api/test/app.e2e-spec.ts +++ b/packages/mukti-api/test/app.e2e-spec.ts @@ -7,8 +7,6 @@ import { tmpdir } from 'os'; import { join } from 'path'; import * as request from 'supertest'; -import { startApi } from '../src/start'; - /** * 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 @@ -31,6 +29,9 @@ describe('API (e2e, local mode)', () => { 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());