feat: core identity system and push notifications

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openacp/cli",
-  "version": "2026.411.1",
+  "version": "2026.413.1",
   "private": true,
   "license": "MIT",
   "type": "module",

packages/plugin-sdk/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openacp/plugin-sdk",
-  "version": "2026.411.1",
+  "version": "2026.413.1",
   "description": "SDK for building OpenACP plugins — types, base classes, and testing utilities",
   "type": "module",
   "exports": {

packages/plugin-sdk/src/testing/test-context.ts

@@ -154,6 +154,7 @@ export function createTestContext(opts: TestContextOpts): TestPluginContext {
     async sendMessage(sessionId: string, content: OutgoingMessage): Promise<void> {
       sentMessages.push({ sessionId, content })
     },
+    notify(_target: any, _message: any, _options?: any): void {},
     defineHook(_name: string): void {},
     async emitHook<T extends Record<string, unknown>>(_name: string, payload: T): Promise<T | null> {
       return payload

src/core/channel.ts

@@ -63,6 +63,19 @@ export interface IChannelAdapter {
 
   // Agent switch cleanup — optional, called when switching agents to clear adapter-side per-session state
   cleanupSessionState?(sessionId: string): Promise<void>
+
+  // --- User-targeted notifications (optional) ---
+  /** Send a notification directly to a user by platform ID. Best-effort delivery. */
+  sendUserNotification?(
+    platformId: string,
+    message: NotificationMessage,
+    options?: {
+      via?: 'dm' | 'thread' | 'topic'
+      topicId?: string
+      sessionId?: string
+      platformMention?: { platformUsername?: string; platformId: string }
+    }
+  ): Promise<void>
 }
 
 /**
@@ -100,4 +113,5 @@ export abstract class ChannelAdapter<TCore = unknown> implements IChannelAdapter
   async cleanupSkillCommands(_sessionId: string): Promise<void> {}
   async cleanupSessionState(_sessionId: string): Promise<void> {}
   async archiveSessionTopic(_sessionId: string): Promise<void> {}
+  async sendUserNotification(_platformId: string, _message: NotificationMessage, _options?: any): Promise<void> {}
 }

src/core/event-bus.ts

@@ -13,6 +13,7 @@ export interface EventBusEvents {
     sessionId: string;
     agent: string;
     status: SessionStatus;
+    userId?: string;
   }) => void;
   "session:updated": (data: {
     sessionId: string;
@@ -94,6 +95,15 @@ export interface EventBusEvents {
     resumed?: boolean;
     error?: string;
   }) => void;
+
+  // Identity lifecycle (emitted by @openacp/identity built-in plugin)
+  "identity:created": (data: { userId: string; identityId: string; source: string; displayName: string }) => void;
+  "identity:updated": (data: { userId: string; changes: string[] }) => void;
+  "identity:linked": (data: { userId: string; identityId: string; linkedFrom?: string }) => void;
+  "identity:unlinked": (data: { userId: string; identityId: string; newUserId: string }) => void;
+  "identity:userMerged": (data: { keptUserId: string; mergedUserId: string; movedIdentities: string[] }) => void;
+  "identity:roleChanged": (data: { userId: string; oldRole: string; newRole: string; changedBy?: string }) => void;
+  "identity:seen": (data: { userId: string; identityId: string; sessionId: string }) => void;
 }
 
 /**

src/core/events.ts

@@ -158,6 +158,22 @@ export const BusEvent = {
   // --- Usage ---
   /** Fired when a token usage record is captured (consumed by usage plugin). */
   USAGE_RECORDED: 'usage:recorded',
+
+  // --- Identity lifecycle ---
+  /** Fired when a new user+identity record is created. */
+  IDENTITY_CREATED: 'identity:created',
+  /** Fired when user profile fields change. */
+  IDENTITY_UPDATED: 'identity:updated',
+  /** Fired when two identities are linked (same person). */
+  IDENTITY_LINKED: 'identity:linked',
+  /** Fired when an identity is unlinked into a new user. */
+  IDENTITY_UNLINKED: 'identity:unlinked',
+  /** Fired when two user records are merged during a link operation. */
+  IDENTITY_USER_MERGED: 'identity:userMerged',
+  /** Fired when a user's role changes. */
+  IDENTITY_ROLE_CHANGED: 'identity:roleChanged',
+  /** Fired when a user is seen (throttled). */
+  IDENTITY_SEEN: 'identity:seen',
 } as const satisfies Record<string, keyof EventBusEvents>;
 
 export type BusEventName = typeof BusEvent[keyof typeof BusEvent];

src/core/plugin/plugin-context.ts

@@ -199,6 +199,19 @@ export function createPluginContext(opts: CreatePluginContextOpts): PluginContex
       }
     },
 
+    notify(
+      target: { identityId: string } | { userId: string } | { channelId: string; platformId: string },
+      message: { type: 'text'; text: string },
+      options?: any,
+    ): void {
+      requirePermission(permissions, 'notifications:send', 'notify()')
+      const svc = serviceRegistry.get<{ notifyUser(t: any, m: any, o: any): Promise<void> }>('notifications')
+      if (svc?.notifyUser) {
+        // Fire-and-forget — don't await, swallow errors
+        svc.notifyUser(target, message, options).catch(() => {})
+      }
+    },
+
     defineHook(_name: string): void {
       // MiddlewareChain creates hook entries lazily on first add(), so no pre-registration is needed.
       // This method exists to let plugins declare intent — documentation/discoverability only.

src/core/plugin/types.ts

@@ -44,6 +44,14 @@ export type PluginPermission =
   | 'kernel:access'
   /** Read-only session metadata without kernel:access */
   | 'sessions:read'
+  /** Read identity data (users, identities, search) */
+  | 'identity:read'
+  /** Write identity data (create, update, link, unlink, roles) */
+  | 'identity:write'
+  /** Register an identity source (adapters register their platform name) */
+  | 'identity:register-source'
+  /** Send push notifications to users */
+  | 'notifications:send'
 
 /**
  * The runtime plugin instance — the object a plugin module default-exports.
@@ -425,6 +433,23 @@ export interface PluginContext {
    */
   sendMessage(sessionId: string, content: OutgoingMessage): Promise<void>
 
+  /**
+   * Send a user-targeted notification. Fire-and-forget, best-effort.
+   * Resolves target via identity system — one call delivers across all linked platforms.
+   * Requires 'notifications:send' permission.
+   */
+  notify(
+    target: { identityId: string } | { userId: string } | { channelId: string; platformId: string },
+    message: { type: 'text'; text: string },
+    options?: {
+      via?: 'dm' | 'thread' | 'topic'
+      topicId?: string
+      sessionId?: string
+      onlyPlatforms?: string[]
+      excludePlatforms?: string[]
+    }
+  ): void
+
   /**
    * Define a custom hook that other plugins can register middleware on.
    * The hook name is automatically prefixed with `plugin:{pluginName}:`.
@@ -707,6 +732,18 @@ export interface FileServiceInterface {
 export interface NotificationService {
   notify(channelId: string, notification: NotificationMessage): Promise<void>
   notifyAll(notification: NotificationMessage): Promise<void>
+  /** Send a notification to a user across all their linked platforms. Fire-and-forget. */
+  notifyUser?(
+    target: { identityId: string } | { userId: string } | { channelId: string; platformId: string },
+    message: { type: 'text'; text: string },
+    options?: {
+      via?: 'dm' | 'thread' | 'topic'
+      topicId?: string
+      sessionId?: string
+      onlyPlatforms?: string[]
+      excludePlatforms?: string[]
+    }
+  ): Promise<void>
 }
 
 export interface UsageService {

src/core/sessions/session-factory.ts

@@ -27,6 +27,8 @@ export interface SessionCreateParams {
   existingSessionId?: string;
   initialName?: string;
   isAssistant?: boolean;
+  /** User ID from identity system — who is creating this session. */
+  userId?: string;
 }
 
 export interface SideEffectDeps {
@@ -89,7 +91,7 @@ export class SessionFactory {
       const payload = {
         agentName: params.agentName,
         workingDir: params.workingDirectory,
-        userId: '', // userId is not part of SessionCreateParams — resolved upstream
+        userId: params.userId ?? '',
         channelId: params.channelId,
         threadId: '', // threadId is assigned after session creation
       };
@@ -219,6 +221,7 @@ export class SessionFactory {
         sessionId: session.id,
         agent: session.agentName,
         status: session.status,
+        userId: createParams.userId,
       });
     }
 

src/core/types.ts

@@ -362,6 +362,10 @@ export interface SessionRecord<P = Record<string, unknown>> {
   firstAgent?: string;
   currentPromptCount?: number;
   agentSwitchHistory?: AgentSwitchEntry[];
+  /** userId of the user who created this session (from identity system). */
+  createdBy?: string;
+  /** userId[] of all users who have sent messages in this session. */
+  participants?: string[];
   // ACP state (cached — overridden by agent response on resume)
   acpState?: {
     // Primary fields (used on load)

src/plugins/api-server/auth/token-store.ts

@@ -182,6 +182,20 @@ export class TokenStore {
     }
   }
 
+  /** Associate a user ID with a token. Called by identity plugin after /identity/setup. */
+  setUserId(tokenId: string, userId: string): void {
+    const token = this.tokens.get(tokenId);
+    if (token) {
+      token.userId = userId;
+      this.scheduleSave();
+    }
+  }
+
+  /** Get the user ID associated with a token. */
+  getUserId(tokenId: string): string | undefined {
+    return this.tokens.get(tokenId)?.userId;
+  }
+
   /**
    * Generates a one-time authorization code that can be exchanged for a JWT.
    *

src/plugins/api-server/auth/types.ts

@@ -10,6 +10,8 @@ export interface StoredToken {
   refreshDeadline: string;
   lastUsedAt?: string;
   revoked: boolean;
+  /** User ID from identity system. Null until user completes /identity/setup. */
+  userId?: string;
 }
 
 /** Claims embedded in a signed JWT. `rfd` (refresh deadline) is a Unix timestamp (seconds). */

src/plugins/api-server/index.ts

@@ -222,6 +222,10 @@ function createApiServerPlugin(): OpenACPPlugin {
       await tokenStore.load()
       tokenStoreRef = tokenStore
 
+      // Expose tokenStore as a service so other plugins (e.g. identity) can associate
+      // their user IDs with tokens without a direct import dependency.
+      ctx.registerService('token-store', tokenStore)
+
       // Lazy import to avoid loading Fastify unless needed
       const { createApiServer } = await import('./server.js')
       const { SSEManager } = await import('./sse-manager.js')
@@ -289,7 +293,12 @@ function createApiServerPlugin(): OpenACPPlugin {
       server.registerPlugin('/api/v1/tunnel', async (app) => tunnelRoutes(app, deps))
       server.registerPlugin('/api/v1/notify', async (app) => notifyRoutes(app, deps))
       server.registerPlugin('/api/v1/commands', async (app) => commandRoutes(app, deps))
-      server.registerPlugin('/api/v1/auth', async (app) => authRoutes(app, { tokenStore, getJwtSecret: () => jwtSecret }))
+      server.registerPlugin('/api/v1/auth', async (app) => authRoutes(app, {
+        tokenStore,
+        getJwtSecret: () => jwtSecret,
+        // Lazy resolver: identity plugin may not be loaded, so we fetch it on demand
+        getIdentityService: () => ctx.getService<{ getUser(userId: string): Promise<{ displayName: string } | undefined> }>('identity') ?? undefined,
+      }))
       server.registerPlugin('/api/v1/plugins', async (app) => pluginRoutes(app, deps))
       const appConfig = core.configManager.get()
       const workspaceName = (appConfig as Record<string, unknown>).instanceName as string ?? 'Main'

src/plugins/api-server/routes/auth.ts

@@ -10,6 +10,11 @@ export interface AuthRouteDeps {
   tokenStore: TokenStore;
   /** Returns the current JWT signing secret. Fetched lazily to support future rotation. */
   getJwtSecret: () => string;
+  /**
+   * Optional resolver for the identity service. Provided lazily so the auth plugin
+   * does not hard-depend on identity — if identity is not loaded, this returns undefined.
+   */
+  getIdentityService?: () => { getUser(userId: string): Promise<{ displayName: string } | undefined> } | undefined;
 }
 
 /**
@@ -142,14 +147,29 @@ export async function authRoutes(
     };
   });
 
-  // GET /me — current auth info
+  // GET /me — current auth info, enriched with identity data if available.
+  // Identity fields are optional — callers must not assume they are always present.
   app.get('/me', async (request) => {
     const { auth } = request;
+    const userId = auth.tokenId ? deps.tokenStore.getUserId(auth.tokenId) : undefined;
+
+    let displayName: string | null = null;
+    if (userId && deps.getIdentityService) {
+      const identityService = deps.getIdentityService();
+      if (identityService) {
+        const user = await identityService.getUser(userId);
+        displayName = user?.displayName ?? null;
+      }
+    }
+
     return {
       type: auth.type,
       tokenId: auth.tokenId,
       role: auth.role,
       scopes: auth.scopes,
+      userId: userId ?? null,
+      displayName,
+      claimed: !!userId,
     };
   });
 

src/plugins/core-plugins.ts

@@ -4,6 +4,7 @@
  * Adapter plugins depend on service plugins, so they boot last.
  */
 import securityPlugin from './security/index.js'
+import identityPlugin from './identity/index.js'
 import fileServicePlugin from './file-service/index.js'
 import contextPlugin from './context/index.js'
 import speechPlugin from './speech/index.js'
@@ -30,6 +31,7 @@ import telegramPlugin from './telegram/index.js'
 export const corePlugins = [
   // Service plugins (no adapter dependencies)
   securityPlugin,
+  identityPlugin,   // Must boot after security (blocked users rejected before identity records are created)
   fileServicePlugin,
   contextPlugin,
   speechPlugin,