runner.ts

/**
 * Agent Module Runner
 *
 * Combined module -- agent loop execution + agent CRUD + session management.
 *
 * Subscribes to: inbound:message, session:close, llm:stream:chunk, run:cancel
 * Publishes to:  outbound:{channelType}
 * Provides:      agents, sessions, skills
 */

import { resolve, join } from 'node:path';
import { homedir } from 'node:os';
import { ModuleRunner } from 'jsr:@pons/sdk@^0.2';
import type { ModuleManifest } from 'jsr:@pons/sdk@^0.2';

import type {
  AppConfig,
  InboundMessage,
  HttpRouteRegistration,
  HttpRequestContext,
  Logger,
} from './src/types.ts';
import { SimpleEventEmitter } from './src/events.ts';
import { AgentManager } from './src/manager.ts';
import { SessionStore } from './src/session.ts';
import { LongTermMemory } from './src/memory.ts';
import { PermissionRegistry } from './src/permissions.ts';
import { HookRegistry } from './src/hooks.ts';
import { InteractionManager } from './src/interactions.ts';
import { AgentHandlers } from './src/rpc-handlers/agent-handlers.ts';
import { SessionHandlers } from './src/rpc-handlers/session-handlers.ts';
import { SkillHandlers } from './src/rpc-handlers/skill-handlers.ts';
import { InboundDispatcher } from './src/inbound-dispatcher.ts';

// ─── Runner ─────────────────────────────────────────────────

class AgentModuleRunner extends ModuleRunner {
  readonly manifest: ModuleManifest = {
    id: 'agent',
    name: 'Agent',
    version: '0.1.0',
    description: 'Agent loop execution, CRUD, sessions, tools, and interactions',
    subscribes: ['inbound:message', 'session:close', 'llm:stream:chunk', 'run:cancel', 'interaction:resolved'],
    publishes: ['llm:stream:request', 'agent:turn:end', 'outbound:ws', 'outbound:reply'],
    provides: ['agents', 'sessions'],
    requires: ['llm'],
    optionalRequires: ['http-router', 'interactionQueue', 'model-router', 'skills'],
    priority: 15,
    configDependencies: ['models', 'agents', 'workspace'],
  };

  private agentManager!: AgentManager;
  private sessionStore!: SessionStore;
  private events!: SimpleEventEmitter;
  private interactionManager!: InteractionManager;
  private activeRuns = new Map<string, AbortController>();
  private _logger: Logger | null = null;

  // Handler delegates
  private agentHandlers!: AgentHandlers;
  private sessionHandlers!: SessionHandlers;
  private skillHandlers!: SkillHandlers;
  private inboundDispatcher!: InboundDispatcher;

  // ─── Init ──────────────────────────────────────────────────

  protected override async onInit(): Promise<void> {
    const config = this.config as AppConfig;
    this._logger = this.createLoggerAdapter();
    this.events = new SimpleEventEmitter();

    // Agent Manager
    const agentsDir = resolve(this.projectRoot, config.workspace?.agentsDir || 'agents');
    const personalSpaceRoot = config.workspace?.personalSpaceRoot || join(homedir(), '.pons');

    this.agentManager = new AgentManager(agentsDir, this.workspacePath, personalSpaceRoot, config, this._logger);
    await this.agentManager.loadOverrides();
    await this.agentManager.ensureAgentDirectories();

    // Session Store
    this.sessionStore = new SessionStore(this.workspacePath, this._logger);

    // Permissions
    const permissionRegistry = new PermissionRegistry();
    permissionRegistry.registerCoreScopes();

    // Hooks
    const hookRegistry = new HookRegistry(this._logger);

    // Long-term Memory
    const longTermMemory = new LongTermMemory(this.workspacePath, this._logger);

    // Interaction Manager
    this.interactionManager = new InteractionManager(this.events, this._logger);
    const interactionManager = this.interactionManager;

    // Forward interaction publish events to outbound
    this.events.on('interaction:publish', (_interaction: unknown) => {
      // Actual forwarding happens in InboundDispatcher's interactionHandler
    });

    // ── Create handler delegates ──

    this.agentHandlers = new AgentHandlers(this.agentManager);
    this.sessionHandlers = new SessionHandlers(this.sessionStore);
    this.skillHandlers = new SkillHandlers({
      rpcRequest: <T = unknown>(service: string, method: string, params?: unknown) => this.request<T>(service, method, params),
      agentManager: this.agentManager,
      projectRoot: this.projectRoot,
      config,
      logger: this._logger,
    });
    this.inboundDispatcher = new InboundDispatcher({
      agentManager: this.agentManager,
      sessionStore: this.sessionStore,
      permissionRegistry,
      hookRegistry,
      longTermMemory,
      interactionManager,
      events: this.events,
      logger: this._logger,
      projectRoot: this.projectRoot,
      workspacePath: this.workspacePath,
      config,
      activeRuns: this.activeRuns,
      publish: (topic, payload) => this.publish(topic, payload),
      request: <T = unknown>(service: string, method: string, params?: unknown) => this.request<T>(service, method, params),
      log: (level, msg, data) => this.log(level as 'info' | 'warn' | 'error', msg, data),
      logGroup: (level, msg, meta, items) => this.logGroup(level as 'info' | 'warn' | 'error', msg, meta, items),
      defaultAgentId: () => this.defaultAgentId(),
    });

    this.log('info', `Agent module ready (${this.agentManager.listAgents().length} agents, ${this.sessionStore.list().length} sessions)`);
  }

  // ─── Deps Ready (HTTP routes) ──────────────────────────────

  protected override async onDepsReady(): Promise<void> {
    try {
      // Agent routes
      await this.request('http-router', 'registerRoutes', {
        service: 'agents',
        prefix: '/api/agents',
        middleware: ['auth'],
        routes: [
          { method: 'GET', path: '/', handler: 'list' },
          { method: 'GET', path: '/:id', handler: 'get' },
          { method: 'POST', path: '/', handler: 'create' },
          { method: 'PUT', path: '/:id', handler: 'update' },
          { method: 'PUT', path: '/:id/persona', handler: 'updatePersona' },
          { method: 'DELETE', path: '/:id', handler: 'delete' },
        ],
      } satisfies HttpRouteRegistration);

      // Session routes
      await this.request('http-router', 'registerRoutes', {
        service: 'sessions',
        prefix: '/api/sessions',
        middleware: ['auth'],
        routes: [
          { method: 'GET', path: '/', handler: 'sessionList' },
          { method: 'GET', path: '/:id', handler: 'sessionGet' },
          { method: 'GET', path: '/:id/transcript', handler: 'getTranscript' },
          { method: 'DELETE', path: '/:id', handler: 'sessionDelete' },
        ],
      } satisfies HttpRouteRegistration);

      // Skill routes
      await this.request('http-router', 'registerRoutes', {
        service: 'agents',
        prefix: '/api/skills',
        middleware: ['auth'],
        routes: [
          { method: 'GET', path: '/', handler: 'skillList' },
          { method: 'GET', path: '/:id', handler: 'skillGet' },
          { method: 'POST', path: '/', handler: 'skillCreate' },
          { method: 'PUT', path: '/:id', handler: 'skillUpdate' },
        ],
      } satisfies HttpRouteRegistration);

      this.log('info', 'HTTP routes registered');
    } catch {
      this.log('warn', 'http-router not available — HTTP routes not registered');
    }
  }

  // ─── RPC ───────────────────────────────────────────────────

  protected override async onRequest(method: string, params: unknown): Promise<unknown> {
    // Gateway sends HTTP-originated calls as "http:<handler>" -- strip prefix
    const resolvedMethod = method.startsWith('http:') ? method.slice(5) : method;
    let resolvedParams = params;
    if (method.startsWith('http:') && params && typeof params === 'object') {
      const ctx = params as HttpRequestContext;
      resolvedParams = { ...ctx.params, ...ctx.query, ...(ctx.body && typeof ctx.body === 'object' ? ctx.body as Record<string, unknown> : {}) };
    }

    const p = resolvedParams as Record<string, unknown> | undefined;

    switch (resolvedMethod) {
      // ── agents ──
      case 'list':
        return this.agentHandlers.handleAgentList();
      case 'get':
        return this.agentHandlers.handleAgentGet(p);
      case 'create':
        return this.agentHandlers.handleAgentCreate(p);
      case 'update':
        return this.agentHandlers.handleAgentUpdate(p);
      case 'updatePersona':
        return this.agentHandlers.handleAgentUpdatePersona(p);
      case 'delete':
        return this.agentHandlers.handleAgentDelete(p);
      case 'getConfig':
        return this.agentHandlers.handleAgentGetConfig(p);
      case 'getResolvedConfig':
        return this.agentHandlers.handleAgentGetResolvedConfig(p);

      // ── skills ──
      case 'skillList':
        return this.skillHandlers.handleSkillList();
      case 'skillGet':
        return this.skillHandlers.handleSkillGet(p);
      case 'skillCreate':
        return this.skillHandlers.handleSkillCreate(p);
      case 'skillUpdate':
        return this.skillHandlers.handleSkillUpdate(p);

      // ── sessions ──
      case 'sessionList':
        return this.sessionHandlers.handleSessionList();
      case 'sessionGet':
        return this.sessionHandlers.handleSessionGet(p);
      case 'getOrCreate':
        return this.sessionHandlers.handleSessionGetOrCreate(p);
      case 'updateTitle':
        return this.sessionHandlers.handleSessionUpdateTitle(p);
      case 'updateMetadata':
        return this.sessionHandlers.handleSessionUpdateMetadata(p);
      case 'getTranscript':
        return this.sessionHandlers.handleSessionGetTranscript(p);
      case 'sessionDelete':
        return this.sessionHandlers.handleSessionDelete(p);

      default:
        throw new Error(`Unknown method: ${method}`);
    }
  }

  // ─── Bus Messages ──────────────────────────────────────────

  protected async onMessage(topic: string, payload: unknown): Promise<void> {
    switch (topic) {
      case 'inbound:message':
        await this.inboundDispatcher.handleInbound(payload as InboundMessage);
        break;

      case 'session:close': {
        const { channelType, channelId, userId } = payload as Record<string, string>;
        const key = `${channelType}:${channelId}:${userId}`;
        this.activeRuns.get(key)?.abort();
        this.activeRuns.delete(key);
        break;
      }

      case 'llm:stream:chunk':
        this.events.emit('llm:stream:chunk', payload);
        break;

      case 'run:cancel': {
        const msg = payload as { channelId?: string; sessionId?: string };
        for (const [key, ctrl] of this.activeRuns) {
          const keyChannelId = key.split(':')[1];
          if (msg.sessionId && this.sessionStore.get(msg.sessionId)?.key === key) {
            ctrl.abort();
            this.activeRuns.delete(key);
            this.publish(`outbound:ws`, {
              type: 'run:cancelled',
              channelId: msg.channelId,
              sessionId: msg.sessionId,
              reason: 'User cancelled',
            });
            break;
          }
          if (msg.channelId && keyChannelId === msg.channelId) {
            ctrl.abort();
            this.activeRuns.delete(key);
            this.publish(`outbound:ws`, {
              type: 'run:cancelled',
              channelId: msg.channelId,
              reason: 'User cancelled',
            });
            break;
          }
        }
        break;
      }

      case 'interaction:resolved':
        this.events.emit('interaction:resolved', payload);
        break;
    }
  }

  // ─── Helpers ──────────────────────────────────────────────

  private defaultAgentId(): string | undefined {
    const agents = this.agentManager.listAgents();
    return agents[0]?.id;
  }

  // ─── Shutdown ──────────────────────────────────────────────

  protected override async onShutdown(): Promise<void> {
    this.interactionManager.cancelAll();
    for (const ctrl of this.activeRuns.values()) ctrl.abort();
    this.activeRuns.clear();
    this.events.removeAllListeners();
    this.log('info', 'Agent module stopped');
  }
}

new AgentModuleRunner().start();