Add AsyncAPI support with SignalR overlay and server composition refactor

AGENTS.md

@@ -48,30 +48,31 @@
 <!-- gitnexus:start -->
 # GitNexus — Code Intelligence
 
-This project is indexed by GitNexus as **oasmock** (2049 symbols, 4322 relationships, 79 execution flows). Use the GitNexus MCP tools to understand code, assess impact, and navigate safely.
+This project is indexed by GitNexus as **oasmock** (2518 symbols, 5666 relationships, 133 execution flows).
 
-> Index stale? Run `node .gitnexus/run.cjs analyze` from the project root — it auto-selects an available runner. No `.gitnexus/run.cjs` yet? `npx gitnexus analyze` (npm 11 crash → `npm i -g gitnexus`; #1939).
+> Index stale? Run `node .gitnexus/run.cjs analyze --index-only` from the project root — it auto-selects an available runner. No `.gitnexus/run.cjs` yet? Bootstrap with `npx`, `bunx`, or `pnpm dlx` — e.g. `bunx gitnexus@latest analyze` (npm 11 npx crash; #1939).
 
 ## Always Do
 
-- **MUST run impact analysis before editing any symbol.** Before modifying a function, class, or method, run `impact({target: "symbolName", direction: "upstream"})` and report the blast radius (direct callers, affected processes, risk level) to the user.
-- **MUST run `detect_changes()` before committing** to verify your changes only affect expected symbols and execution flows. For regression review, compare against the default branch: `detect_changes({scope: "compare", base_ref: "main"})`.
+- **MUST run impact analysis before editing.** Use `impact({target: "symbolName", direction: "upstream"})` (MCP) or `node .gitnexus/run.cjs impact "symbolName" --direction upstream --repo .` (CLI fallback); report callers, processes, and risk. Never substitute grep for graph analysis.
+- **MUST analyze graph changes before committing.** Use `detect_changes({scope: "all"})` (MCP) or `node .gitnexus/run.cjs detect-changes --scope all --repo .` (CLI fallback). `partial: true` or `truncated: true` is not a clean check — a zero means unseen, not unaffected; re-run it. For regression review: `detect_changes({scope: "compare", base_ref: "main"})` or `node .gitnexus/run.cjs detect-changes --scope compare --base-ref "main" --repo .`.
 - **MUST warn the user** if impact analysis returns HIGH or CRITICAL risk before proceeding with edits.
+- **MUST treat `risk: UNKNOWN` as unresolved, not as low.** An empty caller set is not evidence the symbol is unused — it can also mean the callers are not resolvable by the index (plain-object property access, dynamic dispatch, cross-language calls). `impact` pairs `UNKNOWN` with a `riskNote` saying so. Confirm with a text search before treating the symbol as safe to change or delete; do not proceed on the strength of a zero.
 - When exploring unfamiliar code, use `query({search_query: "concept"})` to find execution flows instead of grepping. It returns process-grouped results ranked by relevance.
 - When you need full context on a specific symbol — callers, callees, which execution flows it participates in — use `context({name: "symbolName"})`.
 - For security review, `explain({target: "fileOrSymbol"})` lists taint findings (source→sink flows; needs `analyze --pdg`).
 
 ## Never Do
 
-- NEVER edit a function, class, or method without first running `impact` on it.
-- NEVER ignore HIGH or CRITICAL risk warnings from impact analysis.
+- NEVER edit a function, class, or method before MCP/CLI impact analysis.
+- NEVER ignore HIGH or CRITICAL risk warnings from impact analysis, and never read `UNKNOWN` as an all-clear — it means the walk could not answer, which is the one verdict that requires confirming by other means.
 - NEVER rename symbols with find-and-replace — use `rename` which understands the call graph.
-- NEVER commit changes without running `detect_changes()` to check affected scope.
+- NEVER commit before MCP/CLI graph change analysis.
 
 ## Resources
 
 | Resource | Use for |
-|----------|---------|
+| --- | --- |
 | `gitnexus://repo/oasmock/context` | Codebase overview, check index freshness |
 | `gitnexus://repo/oasmock/clusters` | All functional areas |
 | `gitnexus://repo/oasmock/processes` | All execution flows |
@@ -80,12 +81,12 @@ This project is indexed by GitNexus as **oasmock** (2049 symbols, 4322 relations
 ## CLI
 
 | Task | Read this skill file |
-|------|---------------------|
-| Understand architecture / "How does X work?" | `.claude/skills/gitnexus/gitnexus-exploring/SKILL.md` |
-| Blast radius / "What breaks if I change X?" | `.claude/skills/gitnexus/gitnexus-impact-analysis/SKILL.md` |
-| Trace bugs / "Why is X failing?" | `.claude/skills/gitnexus/gitnexus-debugging/SKILL.md` |
-| Rename / extract / split / refactor | `.claude/skills/gitnexus/gitnexus-refactoring/SKILL.md` |
-| Tools, resources, schema reference | `.claude/skills/gitnexus/gitnexus-guide/SKILL.md` |
-| Index, status, clean, wiki CLI commands | `.claude/skills/gitnexus/gitnexus-cli/SKILL.md` |
+| --- | --- |
+| Understand architecture / "How does X work?" | `.claude/skills/gitnexus-exploring/SKILL.md` |
+| Blast radius / "What breaks if I change X?" | `.claude/skills/gitnexus-impact-analysis/SKILL.md` |
+| Trace bugs / "Why is X failing?" | `.claude/skills/gitnexus-debugging/SKILL.md` |
+| Rename / extract / split / refactor | `.claude/skills/gitnexus-refactoring/SKILL.md` |
+| Tools, resources, schema reference | `.claude/skills/gitnexus-guide/SKILL.md` |
+| Index, status, clean, wiki CLI commands | `.claude/skills/gitnexus-cli/SKILL.md` |
 
 <!-- gitnexus:end -->

api/openapi.yaml

@@ -81,17 +81,149 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/RequestHistoryResponse'
+  /events/fire:
+    post:
+      operationId: fireEvent
+      summary: Fire a named event on the event bus
+      description: |
+        Fires a named event ad-hoc, delivering it (immediately or after a delay)
+        to x-send-events consumers across the loaded schemas.
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/FireEventRequest'
+      responses:
+        '200':
+          description: Event accepted
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/AsyncActionResponse'
+        '400':
+          description: Invalid request
+  /ws/push:
+    post:
+      operationId: pushToChannel
+      summary: Push a message to channel consumers
+      description: |
+        Pushes a message to the connected consumers of an AsyncAPI channel.
+        Supports immediate or delayed delivery, targeted (by connectionId) or
+        broadcast push, and runtime-expression templating of the payload.
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/PushRequest'
+      responses:
+        '200':
+          description: Push accepted
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/AsyncActionResponse'
+        '400':
+          description: Invalid request
+        '404':
+          description: Unknown connectionId
+  /ws/consumers:
+    get:
+      operationId: listConsumers
+      summary: List connected consumers per channel
+      description: |
+        Returns the currently connected consumers for an AsyncAPI channel,
+        including open SignalR streams when applicable.
+      parameters:
+        - name: channel
+          in: query
+          required: true
+          description: Channel address
+          schema:
+            type: string
+      responses:
+        '200':
+          description: Consumers listed
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ConsumersResponse'
+  /ws/schedule:
+    post:
+      operationId: scheduleRecurringPush
+      summary: Schedule a recurring push
+      description: |
+        Schedules a message to be pushed to a channel at a fixed interval until
+        cancelled by its push ID.
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/ScheduleRequest'
+      responses:
+        '200':
+          description: Schedule created
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/AsyncActionResponse'
+        '400':
+          description: Invalid request
+  /ws/schedule/{pushId}:
+    delete:
+      operationId: stopRecurringPush
+      summary: Stop a recurring push
+      parameters:
+        - name: pushId
+          in: path
+          required: true
+          description: Push ID returned by the schedule endpoint
+          schema:
+            type: string
+      responses:
+        '200':
+          description: Schedule stopped
+        '404':
+          description: Unknown pushId
+  /ws/disconnect:
+    post:
+      operationId: disconnectConsumer
+      summary: Force-disconnect a consumer
+      description: |
+        Terminates a connected consumer's WebSocket connection, with an optional
+        close reason/code, or simulates an abrupt drop.
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/DisconnectRequest'
+      responses:
+        '200':
+          description: Consumer disconnected
+        '400':
+          description: Invalid request
+        '404':
+          description: Unknown connectionId
 components:
   schemas:
     AddExampleRequest:
       type: object
       required:
-        - path
         - response
       properties:
         path:
           type: string
           description: The request path (including path parameters) to match
+        protocol:
+          type: string
+          enum: [http, ws]
+          description: AsyncAPI protocol when targeting an AsyncAPI channel
+        channel:
+          type: string
+          description: AsyncAPI channel address (prefixed) when targeting an AsyncAPI channel
         method:
           type: string
           enum: [GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS]
@@ -184,3 +316,110 @@ components:
           type: array
           items:
             $ref: '#/components/schemas/RequestHistoryItem'
+    FireEventRequest:
+      type: object
+      required:
+        - event
+      properties:
+        event:
+          type: string
+          description: The named event to fire
+        payload:
+          type: object
+          description: Event payload exposed to consuming templates via {$event.*}
+        delay:
+          type: integer
+          minimum: 0
+          default: 0
+          description: Delivery delay in milliseconds
+        global:
+          type: boolean
+          default: false
+          description: When true, the event is broadcast over all loaded schemas
+    PushRequest:
+      type: object
+      required:
+        - channel
+      properties:
+        channel:
+          type: string
+          description: AsyncAPI channel address to push to
+        connectionId:
+          type: string
+          description: Target a specific consumer; omitted broadcasts to all
+        payload:
+          type: object
+          description: Message payload, templated with {$state.*}/{$env.*}
+        delay:
+          type: integer
+          minimum: 0
+          default: 0
+          description: Delivery delay in milliseconds
+    ScheduleRequest:
+      type: object
+      required:
+        - channel
+        - interval
+      properties:
+        channel:
+          type: string
+          description: AsyncAPI channel address to push to
+        interval:
+          type: integer
+          minimum: 1
+          description: Delivery interval in milliseconds
+        payload:
+          type: object
+          description: Message payload pushed at each interval
+    DisconnectRequest:
+      type: object
+      required:
+        - connectionId
+      properties:
+        connectionId:
+          type: string
+          description: Active consumer connection id
+        reason:
+          type: string
+          description: Optional close reason
+        code:
+          type: integer
+          description: Optional WebSocket close code
+        abrupt:
+          type: boolean
+          default: false
+          description: Simulate an abrupt network drop (no close frame)
+    AsyncActionResponse:
+      type: object
+      properties:
+        success:
+          type: boolean
+        event:
+          type: string
+          description: Fired event name (fire-event endpoint only)
+        pushId:
+          type: string
+          description: Scheduled push id (schedule endpoint only)
+    ConsumersResponse:
+      type: object
+      properties:
+        consumers:
+          type: array
+          items:
+            type: object
+            properties:
+              connectionId:
+                type: string
+              channel:
+                type: string
+              streams:
+                type: array
+                items:
+                  type: object
+                  properties:
+                    connectionId:
+                      type: string
+                    invocationId:
+                      type: string
+                    streamId:
+                      type: string

cmd/oasmock/mock.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"net/http"
@@ -55,6 +56,15 @@ func portError(format string, args ...any) error {
 	}
 }
 
+// validatePort validates the --port flag value. Port 0 selects an OS-assigned
+// (ephemeral) port and is always valid (RS.CLI.32).
+func validatePort(port int) error {
+	if port != 0 && (port < minPort || port > maxPort) {
+		return validationError("port must be between 1 and 65535")
+	}
+	return nil
+}
+
 func parseSchemaConfig(cmd *cobra.Command) error {
 	// If --from flag was provided, ignore YAML schema configuration
 	if cmd != nil && cmd.Flags().Changed("from") {
@@ -182,8 +192,9 @@ func runMock(cmd *cobra.Command, args []string) error {
 	if len(config.sources) != len(config.prefixes) && len(config.prefixes) != 0 {
 		return validationError("number of --prefix flags must match number of --from flags, or no --prefix flags provided")
 	}
-	if port <= 0 || port > maxPort {
-		return validationError("port must be between 1 and 65535")
+	// port 0 selects an OS-assigned (ephemeral) port (RS.CLI.32).
+	if err := validatePort(port); err != nil {
+		return err
 	}
 	if delay < 0 {
 		return validationError("delay cannot be negative")
@@ -214,17 +225,29 @@ func runMock(cmd *cobra.Command, args []string) error {
 		return schemaError("failed to create server: %v", err)
 	}
 
-	// Start server in a goroutine so we can handle signals
+	// Bind the port up front so the "started" log is only emitted after a
+	// successful bind and carries the actual bound port (--port 0 selects an
+	// OS-assigned port, RS.CLI.11/RS.CLI.32). A collision is reported
+	// synchronously with exit code 4 (RS.CLI.17).
+	ln, boundPort, err := srv.Listen()
+	if err != nil {
+		if errors.Is(err, syscall.EADDRINUSE) {
+			return portError("port %d is already in use", port)
+		}
+		return portError("failed to listen on port %d: %v", port, err)
+	}
+
+	// Serve in a goroutine so we can handle signals
 	serverErrChan := make(chan error, 1)
 	go func() {
-		if err := srv.Start(); err != nil && err != http.ErrServerClosed {
+		if err := srv.Serve(ln); err != nil && err != http.ErrServerClosed {
 			slog.Error("Server error", "err", err)
 			serverErrChan <- err
 		}
 	}()
 
 	// Wait for interrupt signal
-	slog.Info("Mock server started", "port", port)
+	slog.Info("Mock server started", "port", boundPort)
 	slog.Info("Press Ctrl+C to stop")
 
 	// Set up signal handling for graceful shutdown