diff --git a/.env.example b/.env.example index 7ee3cddb..b412464c 100644 --- a/.env.example +++ b/.env.example @@ -1,5 +1,11 @@ # Server Configuration NODE_ENV=development +# Optional base port for local multi-instance runs: +# MCP_HTTP_PORT=PCP_PORT_BASE +# WEB_PORT=PCP_PORT_BASE+1 +# MYRA_HTTP_PORT=PCP_PORT_BASE+2 +# PORT defaults to PCP_PORT_BASE-1 (if needed) +PCP_PORT_BASE=3001 PORT=3000 # Database - Supabase (use newer naming convention) @@ -24,6 +30,9 @@ MCP_AUTH_TOKEN=your-secret-token-for-http-transport # Myra (persistent messaging process) MYRA_HTTP_PORT=3003 +# Web dashboard (used by PM2 ecosystem config) +WEB_PORT=3002 + # Authentication JWT_SECRET=your-jwt-secret-key-min-32-chars-long JWT_EXPIRES_IN=7d diff --git a/.gitignore b/.gitignore index 5ea547ee..d620f44a 100644 --- a/.gitignore +++ b/.gitignore @@ -147,3 +147,6 @@ vite.config.ts.timestamp-* # Backend-specific config (generated by sb mcp sync) .codex/ .gemini/ + +# Local PCP identity should be machine/user-specific +.pcp/identity.json diff --git a/.pcp/identity.json b/.pcp/identity.json deleted file mode 100644 index 7af8aeef..00000000 --- a/.pcp/identity.json +++ /dev/null @@ -1,5 +0,0 @@ -{ - "agentId": "wren", - "context": "pcp-development", - "description": "Primary development context for PCP. Wren is the Claude Code collaborator." -} diff --git a/README.md b/README.md index 6b8dafe8..fa339229 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,66 @@ sb See [packages/cli/README.md](./packages/cli/README.md) for full CLI documentation. +## Database Setup (Supabase) + +PCP supports both: + +- **Remote Supabase** (hosted Supabase project) +- **Local Supabase** (Docker + Supabase CLI) + +### Option A: Remote Supabase (quickest to start) + +1. Create/select a Supabase project. +2. Copy your project URL + API keys. +3. Fill `.env.local` from `.env.example`: + - `SUPABASE_URL` + - `SUPABASE_PUBLISHABLE_KEY` + - `SUPABASE_SECRET_KEY` +4. Start PCP: + +```bash +yarn dev +``` + +### Option B: Local Supabase (best for offline/dev parity) + +1. Install Supabase CLI and Docker: + - Supabase CLI install docs: https://supabase.com/docs/guides/cli/getting-started +2. Start local Supabase from this repo root: + +```bash +supabase start +``` + +3. Reset/apply migrations + seed data: + +```bash +supabase db reset +``` + +4. Print local env values: + +```bash +supabase status -o env +``` + +5. Map local values into `.env.local`: + - `API_URL` → `SUPABASE_URL` + - `ANON_KEY` → `SUPABASE_PUBLISHABLE_KEY` + - `SERVICE_ROLE_KEY` → `SUPABASE_SECRET_KEY` + +6. Start PCP: + +```bash +yarn dev +``` + +Useful Supabase docs: + +- Local development workflow: https://supabase.com/docs/guides/cli/local-development +- CLI reference (`start`, `status`, `db reset`, etc.): https://supabase.com/docs/reference/cli/start +- API key types and guidance: https://supabase.com/docs/guides/api/api-keys + ## Project Structure ``` diff --git a/ecosystem.config.cjs b/ecosystem.config.cjs index 47a3b03a..91d73d02 100644 --- a/ecosystem.config.cjs +++ b/ecosystem.config.cjs @@ -26,6 +26,11 @@ const path = require('path'); const rootDir = __dirname; const apiDir = path.join(rootDir, 'packages/api'); const webDir = path.join(rootDir, 'packages/web'); +const basePort = Number(process.env.PCP_PORT_BASE || 3001); // MCP-first base +const apiPort = Number(process.env.PORT || basePort - 1); +const mcpPort = Number(process.env.MCP_HTTP_PORT || basePort); +const webPort = Number(process.env.WEB_PORT || basePort + 1); +const myraPort = Number(process.env.MYRA_HTTP_PORT || basePort + 2); // Yarn workspaces hoists dependencies to root node_modules const tsxBin = path.join(rootDir, 'node_modules/.bin/tsx'); @@ -44,6 +49,9 @@ module.exports = { env: { NODE_ENV: 'development', MCP_TRANSPORT: 'http', + PORT: String(apiPort), + MCP_HTTP_PORT: String(mcpPort), + MYRA_HTTP_PORT: String(myraPort), ENABLE_WHATSAPP: 'true', AGENT_ID: 'myra', // Identity for the Claude Code backend }, @@ -64,7 +72,7 @@ module.exports = { name: 'web', cwd: webDir, script: nextBin, - args: 'dev -p 3002', + args: `dev -p ${webPort}`, watch: false, env: { NODE_ENV: 'development', diff --git a/packages/api/package.json b/packages/api/package.json index d4413e47..4a03a995 100644 --- a/packages/api/package.json +++ b/packages/api/package.json @@ -15,6 +15,7 @@ "test": "vitest run", "test:watch": "vitest", "test:integration": "vitest run --config vitest.integration.config.ts", + "test:codex-e2e": "RUN_CODEX_E2E=1 vitest run --config vitest.integration.config.ts src/services/sessions/codex-runner.integration.test.ts", "test:coverage": "vitest run --coverage", "test:connection": "tsx src/test-connection.ts", "test:channels": "tsx src/test-channels.ts", diff --git a/packages/api/src/config/env.ts b/packages/api/src/config/env.ts index 4f1ee653..1437fd95 100644 --- a/packages/api/src/config/env.ts +++ b/packages/api/src/config/env.ts @@ -25,7 +25,8 @@ const optionalUrl = z.string().url().optional().or(z.literal('')).transform(val const envSchema = z.object({ // Server NODE_ENV: z.enum(['development', 'production', 'test']).default('development'), - PORT: z.string().transform(Number).default('3000'), + PCP_PORT_BASE: z.string().transform(Number).optional(), + PORT: z.string().transform(Number).optional(), // Database - Supabase (supports both old and new naming conventions) SUPABASE_URL: z.string().url(), @@ -43,12 +44,12 @@ const envSchema = z.object({ // MCP Server MCP_TRANSPORT: z.enum(['stdio', 'http']).default('stdio'), - MCP_HTTP_PORT: z.string().transform(Number).default('3001'), + MCP_HTTP_PORT: z.string().transform(Number).optional(), MCP_BASE_URL: optionalUrl, // Public base URL (e.g., https://pcp.example.com). Defaults to http://localhost:{MCP_HTTP_PORT} MCP_AUTH_TOKEN: optionalString, // Myra (persistent messaging process) - MYRA_HTTP_PORT: z.string().transform(Number).default('3003'), + MYRA_HTTP_PORT: z.string().transform(Number).optional(), // Authentication JWT_SECRET: z.string().min(32), @@ -85,11 +86,38 @@ const parseEnv = () => { } // Create normalized keys (prefer new naming) + const hasBaseOverride = parsed.PCP_PORT_BASE !== undefined; + // Base is MCP-first: MCP=base, WEB=base+1, MYRA=base+2 + const portBase = parsed.PCP_PORT_BASE ?? 3001; + + // If PCP_PORT_BASE is provided and legacy defaults are still present, + // treat those defaults as unset so the base can drive derived ports. + const port = + parsed.PORT === undefined || (hasBaseOverride && parsed.PORT === 3000) + ? portBase - 1 + : parsed.PORT; + const mcpHttpPort = + parsed.MCP_HTTP_PORT === undefined || (hasBaseOverride && parsed.MCP_HTTP_PORT === 3001) + ? portBase + : parsed.MCP_HTTP_PORT; + const myraHttpPort = + parsed.MYRA_HTTP_PORT === undefined || (hasBaseOverride && parsed.MYRA_HTTP_PORT === 3003) + ? portBase + 2 + : parsed.MYRA_HTTP_PORT; + return { ...parsed, + PCP_PORT_BASE: portBase, + PORT: port, + MCP_HTTP_PORT: mcpHttpPort, + MYRA_HTTP_PORT: myraHttpPort, SUPABASE_PUBLISHABLE_KEY: parsed.SUPABASE_PUBLISHABLE_KEY || parsed.SUPABASE_ANON_KEY, SUPABASE_SECRET_KEY: parsed.SUPABASE_SECRET_KEY || parsed.SUPABASE_SERVICE_KEY, } as typeof parsed & { + PCP_PORT_BASE: number; + PORT: number; + MCP_HTTP_PORT: number; + MYRA_HTTP_PORT: number; SUPABASE_PUBLISHABLE_KEY: string; SUPABASE_SECRET_KEY: string; }; diff --git a/packages/api/src/data/supabase/types.ts b/packages/api/src/data/supabase/types.ts index ab7a72a3..eb8edf5c 100644 --- a/packages/api/src/data/supabase/types.ts +++ b/packages/api/src/data/supabase/types.ts @@ -121,6 +121,7 @@ export type Database = { agent_identities: { Row: { agent_id: string + backend: string | null capabilities: Json | null created_at: string | null description: string | null @@ -138,6 +139,7 @@ export type Database = { } Insert: { agent_id: string + backend?: string | null capabilities?: Json | null created_at?: string | null description?: string | null @@ -155,6 +157,7 @@ export type Database = { } Update: { agent_id?: string + backend?: string | null capabilities?: Json | null created_at?: string | null description?: string | null @@ -184,6 +187,7 @@ export type Database = { Row: { agent_id: string archived_at: string | null + backend: string | null capabilities: Json | null change_type: string created_at: string @@ -203,6 +207,7 @@ export type Database = { Insert: { agent_id: string archived_at?: string | null + backend?: string | null capabilities?: Json | null change_type?: string created_at: string @@ -222,6 +227,7 @@ export type Database = { Update: { agent_id?: string archived_at?: string | null + backend?: string | null capabilities?: Json | null change_type?: string created_at?: string diff --git a/packages/api/src/services/sessions/codex-runner.integration.test.ts b/packages/api/src/services/sessions/codex-runner.integration.test.ts new file mode 100644 index 00000000..991b85bb --- /dev/null +++ b/packages/api/src/services/sessions/codex-runner.integration.test.ts @@ -0,0 +1,77 @@ +/** + * CodexRunner E2E Integration Test + * + * This test invokes the real Codex CLI and therefore requires: + * - codex installed + * - codex login (or API-key login) configured + * + * It is gated behind RUN_CODEX_E2E=1 to avoid accidental usage/cost. + */ + +import { describe, it, expect, beforeAll } from 'vitest'; +import { execSync } from 'child_process'; +import { CodexRunner } from './codex-runner.js'; + +let shouldRun = false; +let skipReason = ''; + +describe('CodexRunner E2E (real codex cli)', () => { + beforeAll(() => { + if (process.env.RUN_CODEX_E2E !== '1') { + skipReason = 'Set RUN_CODEX_E2E=1 to run real Codex E2E tests'; + // eslint-disable-next-line no-console + console.warn(`[codex-e2e] Skipping: ${skipReason}`); + return; + } + + try { + execSync('codex --version', { stdio: 'pipe' }); + } catch { + skipReason = 'codex binary is not installed'; + // eslint-disable-next-line no-console + console.warn(`[codex-e2e] Skipping: ${skipReason}`); + return; + } + + try { + // codex currently prints login status to stderr, so capture both streams + const status = execSync('codex login status 2>&1', { stdio: 'pipe', encoding: 'utf-8' }); + const loggedIn = /logged in/i.test(status); + if (!loggedIn) { + skipReason = 'codex login is required (run `codex login` or `codex login --with-api-key`)'; + // eslint-disable-next-line no-console + console.warn(`[codex-e2e] Skipping: ${skipReason}`); + return; + } + } catch { + skipReason = 'codex login status check failed'; + // eslint-disable-next-line no-console + console.warn(`[codex-e2e] Skipping: ${skipReason}`); + return; + } + + shouldRun = true; + }); + + it('can run a simple prompt via real codex exec --json', async () => { + if (!shouldRun) { + expect(skipReason.length).toBeGreaterThan(0); + return; + } + + const runner = new CodexRunner(); + const result = await runner.run('Reply with exactly: CODEX_E2E_OK', { + config: { + workingDirectory: process.cwd(), + mcpConfigPath: '', + model: process.env.CODEX_E2E_MODEL || 'gpt-5-codex', + appendSystemPrompt: 'You are running an integration test. Keep responses brief.', + }, + }); + + expect(result.success).toBe(true); + expect(typeof result.claudeSessionId).toBe('string'); + expect(result.claudeSessionId.length).toBeGreaterThan(0); + expect(result.finalTextResponse).toBeTruthy(); + }); +}); diff --git a/packages/api/src/services/sessions/codex-runner.test.ts b/packages/api/src/services/sessions/codex-runner.test.ts new file mode 100644 index 00000000..5ea4af43 --- /dev/null +++ b/packages/api/src/services/sessions/codex-runner.test.ts @@ -0,0 +1,120 @@ +import { describe, it, expect, vi, beforeEach, type Mock } from 'vitest'; +import { EventEmitter } from 'events'; + +vi.mock('child_process', () => ({ + spawn: vi.fn(), +})); + +vi.mock('../../utils/logger.js', () => ({ + logger: { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + }, +})); + +import { spawn } from 'child_process'; +import { CodexRunner } from './codex-runner.js'; + +function createMockProcess() { + return Object.assign(new EventEmitter(), { + stdout: new EventEmitter(), + stderr: new EventEmitter(), + kill: vi.fn(), + }); +} + +describe('CodexRunner', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('should parse send_response tool calls and usage from codex json stream', async () => { + const mockProc = createMockProcess(); + (spawn as Mock).mockReturnValue(mockProc); + + const runner = new CodexRunner(); + const runPromise = runner.run('hello', { + config: { + workingDirectory: process.cwd(), + mcpConfigPath: '', + model: 'gpt-5-codex', + appendSystemPrompt: 'identity override', + }, + }); + + setTimeout(() => { + mockProc.stdout.emit('data', Buffer.from( + `${JSON.stringify({ + type: 'tool_use', + id: 'tu-1', + name: 'mcp__pcp__send_response', + input: { channel: 'telegram', conversationId: 'chat-1', content: 'hi from codex' }, + })}\n` + )); + mockProc.stdout.emit('data', Buffer.from( + `${JSON.stringify({ + session_id: 'codex-session-123', + input_tokens: 12, + output_tokens: 5, + context_tokens: 42, + result: 'done', + })}\n` + )); + mockProc.emit('close', 0); + }, 5); + + const result = await runPromise; + expect(result.success).toBe(true); + expect(result.claudeSessionId).toBe('codex-session-123'); + expect(result.responses).toEqual([ + { + channel: 'telegram', + conversationId: 'chat-1', + content: 'hi from codex', + format: undefined, + replyToMessageId: undefined, + }, + ]); + expect(result.usage).toEqual({ + contextTokens: 42, + inputTokens: 12, + outputTokens: 5, + }); + expect(result.finalTextResponse).toBe('done'); + expect(result.toolCalls?.length).toBe(1); + }); + + it('should run resume mode when session id exists', async () => { + const mockProc = createMockProcess(); + (spawn as Mock).mockReturnValue(mockProc); + + const runner = new CodexRunner(); + const runPromise = runner.run('resume msg', { + claudeSessionId: 'existing-session-abc', + config: { + workingDirectory: process.cwd(), + mcpConfigPath: '', + model: 'gpt-5-codex', + appendSystemPrompt: 'identity override', + }, + }); + + setTimeout(() => { + mockProc.stdout.emit('data', Buffer.from(`${JSON.stringify({ result: 'ok' })}\n`)); + mockProc.emit('close', 0); + }, 5); + + await runPromise; + + expect(spawn).toHaveBeenCalledTimes(1); + const [, args] = (spawn as Mock).mock.calls[0] as [string, string[]]; + expect(args[0]).toBe('exec'); + expect(args[1]).toBe('resume'); + expect(args).toContain('--json'); + expect(args).toContain('existing-session-abc'); + expect(args).toContain('resume msg'); + }); +}); + diff --git a/packages/api/src/services/sessions/codex-runner.ts b/packages/api/src/services/sessions/codex-runner.ts new file mode 100644 index 00000000..21191da9 --- /dev/null +++ b/packages/api/src/services/sessions/codex-runner.ts @@ -0,0 +1,391 @@ +/** + * Codex Runner + * + * Spawns Codex CLI in non-interactive JSON mode. + * Supports fresh runs and resume runs for session continuity. + */ + +import { spawn, type ChildProcess } from 'child_process'; +import { randomUUID } from 'crypto'; +import { mkdtempSync, rmSync, writeFileSync } from 'fs'; +import { join } from 'path'; +import { tmpdir } from 'os'; +import type { + InjectedContext, + ClaudeRunnerConfig, + ClaudeRunnerResult, + ChannelResponse, + ChannelType, + IClaudeRunner, + ToolCall, +} from './types.js'; +import { formatInjectedContext } from './context-builder.js'; +import { logger } from '../../utils/logger.js'; + +const PROCESS_TIMEOUT_MS = 10 * 60 * 1000; // 10 minutes + +interface CodexUsageStats { + contextTokens: number; + inputTokens: number; + outputTokens: number; +} + +export class CodexRunner implements IClaudeRunner { + async run( + message: string, + options: { + claudeSessionId?: string; + injectedContext?: InjectedContext; + config: ClaudeRunnerConfig; + } + ): Promise { + const { claudeSessionId, injectedContext, config } = options; + const isResume = !!claudeSessionId; + + let sessionId = claudeSessionId || randomUUID(); + + let fullMessage = message; + if (injectedContext && !isResume) { + const contextBlock = formatInjectedContext(injectedContext); + fullMessage = `${contextBlock}\n\n---\n\n${message}`; + } + + const { promptPath, cleanup } = this.createIdentityPromptTempFile( + config.appendSystemPrompt || config.systemPrompt || '' + ); + + try { + const args = this.buildArgs(sessionId, isResume, fullMessage, config, promptPath); + logger.info('Spawning Codex CLI', { + sessionId, + isResume, + workingDirectory: config.workingDirectory, + messageLength: fullMessage.length, + }); + + const result = await this.spawnProcess(args, config); + if (result.sessionId) { + sessionId = result.sessionId; + } + + return { + success: true, + claudeSessionId: sessionId, + responses: result.responses, + usage: result.usage, + finalTextResponse: result.finalTextResponse, + toolCalls: result.toolCalls, + }; + } catch (error) { + logger.error('Codex process failed', { + sessionId, + error: error instanceof Error ? error.message : String(error), + }); + return { + success: false, + claudeSessionId: sessionId, + responses: [], + error: error instanceof Error ? error.message : 'Unknown error', + }; + } finally { + cleanup(); + } + } + + private buildArgs( + sessionId: string, + isResume: boolean, + message: string, + config: ClaudeRunnerConfig, + promptPath: string + ): string[] { + const args: string[] = ['exec']; + if (isResume) { + args.push('resume'); + } + + args.push('--json'); + args.push('-c', `model_instructions_file=${promptPath}`); + + if (config.model) { + args.push('-m', config.model); + } + + if (isResume) { + args.push(sessionId); + args.push(message); + } else { + args.push(message); + } + + return args; + } + + private async spawnProcess( + args: string[], + config: ClaudeRunnerConfig + ): Promise<{ + responses: ChannelResponse[]; + usage?: CodexUsageStats; + finalTextResponse?: string; + toolCalls: ToolCall[]; + sessionId?: string; + }> { + return new Promise((resolve, reject) => { + const proc = spawn('codex', args, { + cwd: config.workingDirectory, + env: { ...process.env, HOME: process.env.HOME, PATH: process.env.PATH }, + stdio: ['ignore', 'pipe', 'pipe'], + }); + + let stderr = ''; + const responses: ChannelResponse[] = []; + const toolCalls: ToolCall[] = []; + let usage: CodexUsageStats | undefined; + let finalTextResponse: string | undefined; + let resolvedSessionId: string | undefined; + + let settled = false; + const timeout = setTimeout(() => { + if (!settled) { + settled = true; + this.killProcess(proc); + resolve({ + responses, + usage, + finalTextResponse: finalTextResponse || '[Codex process timed out]', + toolCalls, + sessionId: resolvedSessionId, + }); + } + }, PROCESS_TIMEOUT_MS); + + proc.stdout.on('data', (data) => { + const chunk = data.toString(); + const lines = chunk.split('\n').filter((line: string) => line.trim()); + + for (const line of lines) { + try { + const parsed = JSON.parse(line) as Record; + + const maybeSessionId = this.extractSessionId(parsed); + if (maybeSessionId) resolvedSessionId = maybeSessionId; + + const maybeUsage = this.extractUsage(parsed); + if (maybeUsage) usage = maybeUsage; + + const maybeText = this.extractFinalText(parsed); + if (maybeText) finalTextResponse = maybeText; + + const extracted = this.extractToolData(parsed); + responses.push(...extracted.responses); + toolCalls.push(...extracted.toolCalls); + } catch { + // ignore non-JSON lines + } + } + }); + + proc.stderr.on('data', (data) => { + stderr += data.toString(); + }); + + proc.on('error', (error) => { + clearTimeout(timeout); + if (!settled) { + settled = true; + reject(new Error(`Failed to spawn Codex: ${error.message}`)); + } + }); + + proc.on('close', (code) => { + clearTimeout(timeout); + if (settled) return; + settled = true; + + if (code !== 0 && !finalTextResponse && responses.length === 0) { + reject(new Error(`Codex exited with code ${code}: ${stderr}`)); + return; + } + + resolve({ + responses, + usage, + finalTextResponse, + toolCalls, + sessionId: resolvedSessionId, + }); + }); + }); + } + + private killProcess(proc: ChildProcess): void { + try { + proc.kill('SIGTERM'); + setTimeout(() => { + try { + if (!proc.killed) proc.kill('SIGKILL'); + } catch { + // already dead + } + }, 5000); + } catch { + // already dead + } + } + + private createIdentityPromptTempFile(content: string): { promptPath: string; cleanup: () => void } { + const dir = mkdtempSync(join(tmpdir(), 'pcp-codex-')); + const promptPath = join(dir, 'identity.md'); + writeFileSync(promptPath, content || 'Follow system identity instructions.'); + return { + promptPath, + cleanup: () => { + try { + rmSync(dir, { recursive: true, force: true }); + } catch { + // ignore cleanup errors + } + }, + }; + } + + private extractSessionId(event: Record): string | undefined { + const queue: unknown[] = [event]; + const sessionKeys = new Set(['session_id', 'sessionId', 'conversation_id', 'conversationId', 'thread_id', 'threadId']); + + while (queue.length > 0) { + const current = queue.shift(); + if (!current || typeof current !== 'object') continue; + const obj = current as Record; + + for (const [key, value] of Object.entries(obj)) { + if (typeof value === 'string' && sessionKeys.has(key)) { + return value; + } + if (value && typeof value === 'object') queue.push(value); + } + } + return undefined; + } + + private extractUsage(event: Record): CodexUsageStats | undefined { + const queue: unknown[] = [event]; + while (queue.length > 0) { + const current = queue.shift(); + if (!current || typeof current !== 'object') continue; + const obj = current as Record; + + const maybeInput = obj.input_tokens; + const maybeOutput = obj.output_tokens; + const maybeContext = obj.context_tokens; + if (typeof maybeInput === 'number' && typeof maybeOutput === 'number') { + const cachedInput = typeof obj.cached_input_tokens === 'number' ? obj.cached_input_tokens : 0; + const cacheRead = typeof obj.cache_read_input_tokens === 'number' ? obj.cache_read_input_tokens : 0; + const cacheCreate = typeof obj.cache_creation_input_tokens === 'number' ? obj.cache_creation_input_tokens : 0; + const totalInput = maybeInput + cachedInput + cacheRead + cacheCreate; + return { + contextTokens: typeof maybeContext === 'number' ? maybeContext : totalInput, + inputTokens: totalInput, + outputTokens: maybeOutput, + }; + } + + for (const value of Object.values(obj)) { + if (value && typeof value === 'object') queue.push(value); + } + } + + return undefined; + } + + private extractFinalText(event: Record): string | undefined { + const candidates: unknown[] = [ + event.result, + event.output_text, + event.text, + (event.message as Record | undefined)?.content, + ((event.item as Record | undefined)?.text), + ]; + + for (const candidate of candidates) { + if (typeof candidate === 'string' && candidate.trim()) { + return candidate; + } + } + + // Fallback: recursively find first "text" field with non-empty string + const queue: unknown[] = [event]; + while (queue.length > 0) { + const current = queue.shift(); + if (!current || typeof current !== 'object') continue; + const obj = current as Record; + if (typeof obj.text === 'string' && obj.text.trim()) { + return obj.text; + } + for (const value of Object.values(obj)) { + if (value && typeof value === 'object') queue.push(value); + } + } + + return undefined; + } + + private extractToolData(event: Record): { responses: ChannelResponse[]; toolCalls: ToolCall[] } { + const responses: ChannelResponse[] = []; + const toolCalls: ToolCall[] = []; + const queue: unknown[] = [event]; + + while (queue.length > 0) { + const current = queue.shift(); + if (!current || typeof current !== 'object') continue; + const obj = current as Record; + + const name = obj.name; + if (typeof name === 'string') { + const rawInput = obj.input ?? obj.args ?? obj.arguments ?? {}; + const input = this.normalizeInput(rawInput); + + if (input && typeof input === 'object') { + toolCalls.push({ + toolUseId: typeof obj.id === 'string' ? obj.id : randomUUID(), + toolName: name, + input: input as Record, + }); + + if (name === 'mcp__pcp__send_response') { + const channel = (input as Record).channel as ChannelType | undefined; + const conversationId = (input as Record).conversationId as string | undefined; + const content = (input as Record).content as string | undefined; + if (channel && conversationId && content) { + responses.push({ + channel, + conversationId, + content, + format: (input as Record).format as 'text' | 'markdown' | 'code' | 'json' | undefined, + replyToMessageId: (input as Record).replyToMessageId as string | undefined, + }); + } + } + } + } + + for (const value of Object.values(obj)) { + if (value && typeof value === 'object') queue.push(value); + } + } + + return { responses, toolCalls }; + } + + private normalizeInput(raw: unknown): unknown { + if (typeof raw === 'string') { + try { + return JSON.parse(raw); + } catch { + return {}; + } + } + return raw; + } +} diff --git a/packages/api/src/services/sessions/context-builder.ts b/packages/api/src/services/sessions/context-builder.ts index fbaee2ad..9b2f502c 100644 --- a/packages/api/src/services/sessions/context-builder.ts +++ b/packages/api/src/services/sessions/context-builder.ts @@ -32,6 +32,7 @@ function mapAgentIdentity(row: DbAgentIdentity): AgentIdentity { name: row.name, role: row.role, description: row.description || undefined, + backend: row.backend || undefined, values: Array.isArray(row.values) ? (row.values as string[]) : [], capabilities: Array.isArray(row.capabilities) ? (row.capabilities as string[]) : [], soul: row.soul || undefined, @@ -199,6 +200,26 @@ export class ContextBuilder implements IContextBuilder { }; } + async getAgentBackend( + userId: string, + agentId: string + ): Promise { + const { data, error } = await this.supabase + .from('agent_identities') + .select('backend') + .eq('user_id', userId) + .eq('agent_id', agentId) + .single(); + + if (error) { + if (error.code === 'PGRST116') return null; + logger.error('Error fetching agent backend', { userId, agentId, error }); + return null; + } + + return data?.backend || null; + } + private async getAgentIdentity( userId: string, agentId: string diff --git a/packages/api/src/services/sessions/index.ts b/packages/api/src/services/sessions/index.ts index 4a234fc1..3c7b6e1f 100644 --- a/packages/api/src/services/sessions/index.ts +++ b/packages/api/src/services/sessions/index.ts @@ -20,6 +20,7 @@ export { ContextBuilder, formatInjectedContext } from './context-builder.js'; // Claude runner export { ClaudeRunner, buildIdentityPrompt } from './claude-runner.js'; +export { CodexRunner } from './codex-runner.js'; // Types export type { diff --git a/packages/api/src/services/sessions/session-codex.integration.test.ts b/packages/api/src/services/sessions/session-codex.integration.test.ts new file mode 100644 index 00000000..27ca9405 --- /dev/null +++ b/packages/api/src/services/sessions/session-codex.integration.test.ts @@ -0,0 +1,176 @@ +/** + * SessionService Codex Integration Tests + * + * Verifies database-driven backend selection using agent_identities.backend. + * Creates a test identity "echo_codex" and ensures SessionService routes it + * to Codex runner and persists backend as codex-cli. + */ + +import { describe, it, expect, beforeAll, afterAll, beforeEach, vi } from 'vitest'; +import { getDataComposer, type DataComposer } from '../../data/composer.js'; +import { SessionService } from './session-service.js'; +import { SessionRepository } from './session-repository.js'; +import { ContextBuilder } from './context-builder.js'; +import type { IActivityStream } from './session-service.js'; +import type { IClaudeRunner } from './types.js'; + +describe('SessionService Codex backend integration', () => { + let dataComposer: DataComposer | null = null; + let testUserId: string; + let sessionService: SessionService; + + const claudeRunner: IClaudeRunner = { + run: vi.fn(async () => ({ + success: true, + claudeSessionId: 'claude-test-session', + responses: [], + usage: { contextTokens: 10, inputTokens: 5, outputTokens: 5 }, + finalTextResponse: 'claude response', + toolCalls: [], + })), + }; + + const codexRunner: IClaudeRunner = { + run: vi.fn(async () => ({ + success: true, + claudeSessionId: 'codex-test-session', + responses: [], + usage: { contextTokens: 10, inputTokens: 5, outputTokens: 5 }, + finalTextResponse: 'codex response', + toolCalls: [], + })), + }; + + const activityStream: IActivityStream = { + logMessage: vi.fn(async () => ({ id: 'msg-id' })), + logActivity: vi.fn(async () => ({ id: 'activity-id' })), + }; + + beforeAll(async () => { + dataComposer = await getDataComposer(); + + // Resolve known test user via existing echo identity (seeded by migration 008) + const { data: echoIdentity, error } = await dataComposer.getClient() + .from('agent_identities') + .select('user_id') + .eq('agent_id', 'echo') + .single(); + + if (error || !echoIdentity) { + throw new Error( + 'Echo test agent identity not found. Apply migration 008_add_echo_test_agent.sql first.' + ); + } + + testUserId = echoIdentity.user_id; + + // Ensure echo_codex identity exists and is configured for codex backend + const { data: existing } = await dataComposer.getClient() + .from('agent_identities') + .select('id') + .eq('user_id', testUserId) + .eq('agent_id', 'echo_codex') + .single(); + + if (existing?.id) { + const { error: updateError } = await dataComposer.getClient() + .from('agent_identities') + .update({ + backend: 'codex', + name: 'Echo Codex', + role: 'Test agent identity for Codex backend integration', + }) + .eq('id', existing.id); + + if (updateError) { + throw new Error(`Failed to update echo_codex identity: ${updateError.message}`); + } + } else { + const { error: insertError } = await dataComposer.getClient() + .from('agent_identities') + .insert({ + user_id: testUserId, + agent_id: 'echo_codex', + name: 'Echo Codex', + role: 'Test agent identity for Codex backend integration', + description: 'Created by integration tests', + backend: 'codex', + values: [], + capabilities: [], + relationships: {}, + metadata: { test: true }, + }); + + if (insertError) { + throw new Error(`Failed to create echo_codex identity: ${insertError.message}`); + } + } + + sessionService = new SessionService( + new SessionRepository(dataComposer.getClient()), + new ContextBuilder(dataComposer.getClient()), + claudeRunner, + activityStream, + { + defaultWorkingDirectory: process.cwd(), + mcpConfigPath: '', + defaultModel: 'sonnet', + }, + codexRunner + ); + }); + + beforeEach(async () => { + vi.clearAllMocks(); + if (!dataComposer) return; + await dataComposer.getClient() + .from('sessions') + .delete() + .eq('user_id', testUserId) + .eq('agent_id', 'echo_codex'); + }); + + afterAll(async () => { + if (!dataComposer) return; + await dataComposer.getClient() + .from('sessions') + .delete() + .eq('user_id', testUserId) + .eq('agent_id', 'echo_codex'); + + await dataComposer.getClient() + .from('agent_identities') + .delete() + .eq('user_id', testUserId) + .eq('agent_id', 'echo_codex'); + }); + + it('creates session with codex-cli backend from agent identity backend=codex', async () => { + const session = await sessionService.getOrCreateSession(testUserId, 'echo_codex'); + expect(session.backend).toBe('codex-cli'); + }); + + it('routes message handling through codex runner for echo_codex', async () => { + const result = await sessionService.handleMessage({ + userId: testUserId, + agentId: 'echo_codex', + channel: 'agent', + conversationId: 'integration:echo-codex', + sender: { id: 'integration-test', name: 'Integration Test' }, + content: 'Say hello from integration test', + metadata: { triggerType: 'agent', chatType: 'direct' }, + }); + + expect(result.success).toBe(true); + expect(codexRunner.run).toHaveBeenCalledTimes(1); + expect(claudeRunner.run).not.toHaveBeenCalled(); + + const { data: persisted } = await dataComposer.getClient() + .from('sessions') + .select('backend') + .eq('id', result.sessionId) + .single(); + + expect(persisted?.backend).toBe('codex-cli'); + }); +}); diff --git a/packages/api/src/services/sessions/session-service.test.ts b/packages/api/src/services/sessions/session-service.test.ts index 8a607362..b356a68c 100644 --- a/packages/api/src/services/sessions/session-service.test.ts +++ b/packages/api/src/services/sessions/session-service.test.ts @@ -46,6 +46,7 @@ describe('SessionService', () => { let mockRepository: ISessionRepository; let mockContextBuilder: IContextBuilder; let mockClaudeRunner: IClaudeRunner; + let mockCodexRunner: IClaudeRunner; let mockActivityStream: IActivityStream; const createMockSession = (overrides: Partial = {}): Session => ({ @@ -141,12 +142,17 @@ describe('SessionService', () => { temporal: createMockInjectedContext().temporal, agent: createMockInjectedContext().agent, }), + getAgentBackend: vi.fn().mockResolvedValue('claude'), }; mockClaudeRunner = { run: vi.fn().mockResolvedValue(createMockClaudeResult()), }; + mockCodexRunner = { + run: vi.fn().mockResolvedValue(createMockClaudeResult({ claudeSessionId: 'codex-session-1' })), + }; + mockActivityStream = { logMessage: vi.fn().mockResolvedValue({ id: 'msg-123' }), logActivity: vi.fn().mockResolvedValue({ id: 'activity-123' }), @@ -163,7 +169,8 @@ describe('SessionService', () => { mcpConfigPath: '/test/.mcp.json', defaultModel: 'sonnet', compactionThreshold: 150000, - } + }, + mockCodexRunner ); }); @@ -456,6 +463,41 @@ describe('SessionService', () => { ); }); + it('should resolve codex backend from agent identity when creating a new session', async () => { + vi.mocked(mockRepository.findByUserAndAgent).mockResolvedValue(null); + vi.mocked(mockContextBuilder.getAgentBackend).mockResolvedValue('codex'); + vi.mocked(mockRepository.create).mockResolvedValue(createMockSession({ id: 'new-session' })); + + const request = createMockRequest(); + await sessionService.handleMessage(request); + + expect(mockRepository.create).toHaveBeenCalledWith( + expect.objectContaining({ + backend: 'codex-cli', + }) + ); + }); + + it('should use codex runner when session backend is codex-cli', async () => { + const codexSession = createMockSession({ + id: 'codex-session', + backend: 'codex-cli', + claudeSessionId: null, + }); + vi.mocked(mockRepository.findByUserAndAgent).mockResolvedValue(codexSession); + + const request = createMockRequest({ content: 'Use codex backend please' }); + const result = await sessionService.handleMessage(request); + + expect(result.success).toBe(true); + expect(mockCodexRunner.run).toHaveBeenCalledTimes(1); + expect(mockClaudeRunner.run).not.toHaveBeenCalled(); + expect(mockRepository.update).toHaveBeenCalledWith( + 'codex-session', + expect.objectContaining({ backend: 'codex-cli' }) + ); + }); + it('should increment messageCount after each processed message', async () => { const session = createMockSession({ messageCount: 5 }); vi.mocked(mockRepository.findByUserAndAgent).mockResolvedValue(session); @@ -523,10 +565,13 @@ describe('SessionService', () => { const request = createMockRequest(); await sessionService.handleMessage(request); - expect(mockRepository.update).toHaveBeenCalledWith('session-123', { - claudeSessionId: 'new-claude-id', - messageCount: 1, - }); + expect(mockRepository.update).toHaveBeenCalledWith('session-123', + expect.objectContaining({ + claudeSessionId: 'new-claude-id', + messageCount: 1, + backend: 'claude-code', + }) + ); }); }); diff --git a/packages/api/src/services/sessions/session-service.ts b/packages/api/src/services/sessions/session-service.ts index 29292893..6d3dfbf5 100644 --- a/packages/api/src/services/sessions/session-service.ts +++ b/packages/api/src/services/sessions/session-service.ts @@ -29,6 +29,7 @@ import type { Json } from '../../data/supabase/types.js'; import { SessionRepository } from './session-repository.js'; import { ContextBuilder } from './context-builder.js'; import { ClaudeRunner, buildIdentityPrompt } from './claude-runner.js'; +import { CodexRunner } from './codex-runner.js'; import { ActivityStreamRepository } from '../../data/repositories/activity-stream.repository.js'; import { logger } from '../../utils/logger.js'; @@ -97,6 +98,7 @@ export class SessionService implements ISessionService { private repository: ISessionRepository; private contextBuilder: IContextBuilder; private claudeRunner: IClaudeRunner; + private codexRunner: IClaudeRunner; private activityStream: IActivityStream; private config: SessionServiceConfig; @@ -125,11 +127,13 @@ export class SessionService implements ISessionService { contextBuilder: IContextBuilder, claudeRunner: IClaudeRunner, activityStream: IActivityStream, - config: Partial = {} + config: Partial = {}, + codexRunner?: IClaudeRunner ) { this.repository = repository; this.contextBuilder = contextBuilder; this.claudeRunner = claudeRunner; + this.codexRunner = codexRunner || claudeRunner; this.activityStream = activityStream; this.config = { ...DEFAULT_CONFIG, ...config }; } @@ -311,8 +315,16 @@ export class SessionService implements ISessionService { ), }; - // 4. Run Claude Code - const result = await this.claudeRunner.run(formattedMessage, { + // 4. Select runtime backend and run + const resolvedBackend = this.resolveRuntimeBackend( + session.backend, + injectedContext.agent.backend + ); + const runner = resolvedBackend === 'codex-cli' + ? this.codexRunner + : this.claudeRunner; + + const result = await runner.run(formattedMessage, { claudeSessionId: session.claudeSessionId || undefined, injectedContext: session.claudeSessionId ? undefined : injectedContext, config: runnerConfig, @@ -330,10 +342,12 @@ export class SessionService implements ISessionService { await this.repository.update(session.id, { claudeSessionId: result.claudeSessionId, messageCount: session.messageCount + 1, + backend: resolvedBackend, }); } else { await this.repository.update(session.id, { messageCount: session.messageCount + 1, + backend: resolvedBackend, }); } @@ -382,6 +396,8 @@ export class SessionService implements ISessionService { ): Promise { const type = options?.type || 'primary'; + const backend = await this.resolveAgentBackend(userId, agentId); + // For primary sessions, try to find existing active session if (type === 'primary') { const existing = await this.repository.findByUserAndAgent( @@ -413,7 +429,7 @@ export class SessionService implements ISessionService { totalOutputTokens: 0, messageCount: 0, tokenCount: 0, - backend: 'claude-code', + backend, model: this.config.defaultModel, lastCompactionAt: null, compactionCount: 0, @@ -511,8 +527,13 @@ This session will continue with a fresh context after compaction. Your identity, ), }; + const runtimeBackend = this.resolveRuntimeBackend(session.backend, context.agent.backend); + const runner = runtimeBackend === 'codex-cli' + ? this.codexRunner + : this.claudeRunner; + // Phase 1: Send compaction prompt — agent saves context, notifies users, ends session - const result = await this.claudeRunner.run(compactionPrompt, { + const result = await runner.run(compactionPrompt, { claudeSessionId: session.claudeSessionId, config: runnerConfig, }); @@ -543,6 +564,49 @@ This session will continue with a fresh context after compaction. Your identity, } } + /** + * Normalize backend value to runtime backend IDs used by sessions. + */ + private normalizeBackend(raw: string | null | undefined): 'claude-code' | 'codex-cli' { + const value = (raw || '').toLowerCase().trim(); + if (value === 'codex' || value === 'codex-cli') return 'codex-cli'; + if (value === 'claude' || value === 'claude-code' || value === '') return 'claude-code'; + if (value === 'gemini') { + logger.warn('Gemini backend configured but not yet supported in SessionService, falling back to claude-code'); + return 'claude-code'; + } + logger.warn('Unknown backend configured, falling back to claude-code', { raw }); + return 'claude-code'; + } + + /** + * Resolve backend for a new session from agent identity. + */ + private async resolveAgentBackend(userId: string, agentId: string): Promise<'claude-code' | 'codex-cli'> { + try { + const identityBackend = await this.contextBuilder.getAgentBackend(userId, agentId); + return this.normalizeBackend(identityBackend); + } catch (error) { + logger.warn('Failed to resolve agent backend, falling back to claude-code', { + userId, + agentId, + error: error instanceof Error ? error.message : String(error), + }); + return 'claude-code'; + } + } + + /** + * Resolve backend for this execution, prioritizing persisted session backend. + */ + private resolveRuntimeBackend( + sessionBackend: string | null | undefined, + identityBackend: string | null | undefined + ): 'claude-code' | 'codex-cli' { + if (sessionBackend) return this.normalizeBackend(sessionBackend); + return this.normalizeBackend(identityBackend); + } + async endSession(sessionId: string, summary?: string): Promise { const session = await this.repository.findById(sessionId); if (!session) { @@ -704,6 +768,7 @@ export function createSessionService( new ContextBuilder(supabase), new ClaudeRunner(), new ActivityStreamRepository(supabase), - config + config, + new CodexRunner() ); } diff --git a/packages/api/src/services/sessions/types.ts b/packages/api/src/services/sessions/types.ts index 53e105c1..424d83d1 100644 --- a/packages/api/src/services/sessions/types.ts +++ b/packages/api/src/services/sessions/types.ts @@ -154,6 +154,7 @@ export interface AgentIdentity { name: string; role: string; description?: string; + backend?: string; values: string[]; capabilities: string[]; soul?: string; @@ -333,6 +334,15 @@ export interface IContextBuilder { userId: string, agentId: string ): Promise>; + + /** + * Resolve the preferred runtime backend for an agent identity. + * Returns raw backend string from DB (e.g. "claude", "codex", "gemini"). + */ + getAgentBackend( + userId: string, + agentId: string + ): Promise; } // ─── Claude Runner Interface ─── diff --git a/packages/api/src/skills/README.md b/packages/api/src/skills/README.md index 73bf029e..3d0fc795 100644 --- a/packages/api/src/skills/README.md +++ b/packages/api/src/skills/README.md @@ -44,6 +44,21 @@ git clone https://github.com/user/skill-name curl -o ~/.pcp/skills/my-skill.md https://raw.githubusercontent.com/user/repo/main/SKILL.md ``` +### From URL / Registry (curl) + +If a registry (PCP Hub, community index, or docs page) gives you a direct skill file URL: + +```bash +mkdir -p ~/.pcp/skills +curl -fsSL "https://example.com/skills/my-skill/SKILL.md" -o ~/.pcp/skills/my-skill.md +``` + +If the skill is a directory package, place it under: + +```text +~/.pcp/skills// +``` + ### Future: PCP CLI (coming soon) ```bash @@ -267,9 +282,11 @@ PCP supports cloud-based skill storage and distribution: ### Loading Order -1. **Local skills** (`~/.pcp/skills/`) - Always loaded first -2. **Cloud installations** - User's installed skills from registry -3. **Deduplication** - Local skills take precedence over cloud +Default source priority is deterministic: + +1. **Cloud installations** - User's installed skills from registry +2. **Local skills** (`~/.pcp/skills/`) - Loaded after cloud +3. **Deduplication** - Later sources override earlier ones, so local skills take precedence over cloud when names collide ### User Installation Flow diff --git a/packages/api/src/skills/cloud-service.test.ts b/packages/api/src/skills/cloud-service.test.ts new file mode 100644 index 00000000..809c51c3 --- /dev/null +++ b/packages/api/src/skills/cloud-service.test.ts @@ -0,0 +1,64 @@ +import { describe, it, expect } from 'vitest'; +import type { LoadedSkill } from './types'; +import { CloudSkillsService } from './cloud-service'; +import type { SkillSourceProvider } from './providers'; + +function makeLoadedSkill(name: string, sourcePath: string): LoadedSkill { + return { + manifest: { + name, + version: '1.0.0', + description: `${name} description`, + type: 'guide', + displayName: name, + }, + skillContent: `# ${name}`, + sourcePath, + eligibility: { eligible: true }, + }; +} + +describe('CloudSkillsService', () => { + it('applies configured deterministic source priority when loading user skills', async () => { + const providers: SkillSourceProvider[] = [ + { + id: 'cloud', + loadUserSkills: async () => [makeLoadedSkill('shared', 'cloud://shared')], + }, + { + id: 'local', + loadUserSkills: async () => [makeLoadedSkill('shared', '/tmp/shared.md')], + }, + ]; + + const service = new CloudSkillsService({} as never, { + providers, + sourcePriority: ['cloud', 'local'], + }); + + const loaded = await service.loadUserSkills('user-1'); + expect(loaded).toHaveLength(1); + expect(loaded[0].sourcePath).toBe('/tmp/shared.md'); + }); + + it('continues loading if one provider fails', async () => { + const providers: SkillSourceProvider[] = [ + { + id: 'cloud', + loadUserSkills: async () => { + throw new Error('cloud unavailable'); + }, + }, + { + id: 'local', + loadUserSkills: async () => [makeLoadedSkill('local-only', '/tmp/local-only.md')], + }, + ]; + + const service = new CloudSkillsService({} as never, { providers }); + const loaded = await service.loadUserSkills('user-1'); + + expect(loaded).toHaveLength(1); + expect(loaded[0].manifest.name).toBe('local-only'); + }); +}); diff --git a/packages/api/src/skills/cloud-service.ts b/packages/api/src/skills/cloud-service.ts index 61b2289a..73af33b0 100644 --- a/packages/api/src/skills/cloud-service.ts +++ b/packages/api/src/skills/cloud-service.ts @@ -14,6 +14,14 @@ import { SkillsRepository } from './repository'; import { loadAllSkills as loadLocalSkills } from './loader'; import { checkEligibility } from './eligibility'; import { logger } from '../utils/logger'; +import { + CloudSkillSourceProvider, + DEFAULT_SKILL_SOURCE_PRIORITY, + LocalSkillSourceProvider, + mergeSkillsByPriority, + type SkillSourceId, + type SkillSourceProvider, +} from './providers'; import type { LoadedSkill, SkillSummary, @@ -24,14 +32,26 @@ import type { ListRegistrySkillsOptions, InstallSkillOptions, PublishSkillOptions, - SkillManifest, } from './types'; +interface CloudSkillsServiceOptions { + userSkillsPath?: string; + sourcePriority?: SkillSourceId[]; + providers?: SkillSourceProvider[]; +} + export class CloudSkillsService { private repository: SkillsRepository; + private providers: SkillSourceProvider[]; + private sourcePriority: SkillSourceId[]; - constructor(supabase: SupabaseClient) { + constructor(supabase: SupabaseClient, options: CloudSkillsServiceOptions = {}) { this.repository = new SkillsRepository(supabase); + this.sourcePriority = options.sourcePriority || DEFAULT_SKILL_SOURCE_PRIORITY; + this.providers = options.providers || [ + new CloudSkillSourceProvider(this.repository), + new LocalSkillSourceProvider(options.userSkillsPath), + ]; } // =========================================================================== @@ -139,33 +159,19 @@ export class CloudSkillsService { * Used during bootstrap to get the user's complete skill set. */ async loadUserSkills(userId: string): Promise { - const skills: LoadedSkill[] = []; - - // 1. Load local skills (always available) - const localSkills = loadLocalSkills(); - skills.push(...localSkills); - - // 2. Load user's installed cloud skills - try { - const cloudSkills = await this.repository.getUserInstalledSkills(userId); - - for (const cloudSkill of cloudSkills) { - // Check if already loaded locally (local takes precedence) - const existsLocally = skills.some((s) => s.manifest.name === cloudSkill.name); - if (existsLocally) { - logger.debug(`Skill ${cloudSkill.name} exists locally, skipping cloud version`); - continue; + const loadResults = await Promise.all( + this.providers.map(async (provider) => { + try { + const skills = await provider.loadUserSkills(userId); + return { source: provider.id, skills }; + } catch (error) { + logger.error(`Failed to load ${provider.id} skills; continuing with other sources`, error); + return { source: provider.id, skills: [] }; } + }) + ); - // Convert to LoadedSkill format - const loadedSkill = this.cloudSkillToLoaded(cloudSkill); - skills.push(loadedSkill); - } - } catch (error) { - logger.error('Failed to load cloud skills, using local only:', error); - } - - return skills; + return mergeSkillsByPriority(loadResults, this.sourcePriority); } /** @@ -265,30 +271,6 @@ export class CloudSkillsService { // Helpers // =========================================================================== - /** - * Convert a cloud UserInstalledSkill to LoadedSkill format - */ - private cloudSkillToLoaded(cloud: UserInstalledSkill): LoadedSkill { - const manifest: SkillManifest = { - ...cloud.resolvedManifest, - name: cloud.name, - version: cloud.resolvedVersion, - displayName: cloud.displayName, - description: cloud.description, - type: cloud.type, - category: cloud.category || undefined, - tags: cloud.tags, - emoji: cloud.emoji || undefined, - }; - - return { - manifest, - skillContent: cloud.resolvedContent, - sourcePath: `cloud://${cloud.skillId}`, - eligibility: checkEligibility(manifest.requirements), - }; - } - /** * Convert LoadedSkill to SkillSummary */ diff --git a/packages/api/src/skills/index.ts b/packages/api/src/skills/index.ts index 2f4dae00..cfd1a474 100644 --- a/packages/api/src/skills/index.ts +++ b/packages/api/src/skills/index.ts @@ -17,3 +17,4 @@ export * from './loader'; export * from './service'; export * from './repository'; export * from './cloud-service'; +export * from './providers'; diff --git a/packages/api/src/skills/providers.test.ts b/packages/api/src/skills/providers.test.ts new file mode 100644 index 00000000..adda53f9 --- /dev/null +++ b/packages/api/src/skills/providers.test.ts @@ -0,0 +1,97 @@ +import { describe, it, expect } from 'vitest'; +import type { LoadedSkill } from './types'; +import { mergeSkillsByPriority, cloudInstalledSkillToLoaded } from './providers'; + +function makeLoadedSkill(name: string, sourcePath: string): LoadedSkill { + return { + manifest: { + name, + version: '1.0.0', + description: `${name} description`, + type: 'guide', + displayName: name, + }, + skillContent: `# ${name}`, + sourcePath, + eligibility: { eligible: true }, + }; +} + +describe('skill providers', () => { + describe('mergeSkillsByPriority', () => { + it('applies deterministic source precedence (later source overrides earlier)', () => { + const cloudSkill = makeLoadedSkill('shared-skill', 'cloud://abc'); + const localSkill = makeLoadedSkill('shared-skill', '/Users/test/.pcp/skills/shared-skill.md'); + + const merged = mergeSkillsByPriority( + [ + { source: 'cloud', skills: [cloudSkill] }, + { source: 'local', skills: [localSkill] }, + ], + ['cloud', 'local'] + ); + + expect(merged).toHaveLength(1); + expect(merged[0].sourcePath).toBe('/Users/test/.pcp/skills/shared-skill.md'); + }); + + it('returns deterministic alphabetical ordering of final skills', () => { + const merged = mergeSkillsByPriority( + [ + { source: 'cloud', skills: [makeLoadedSkill('zeta', 'cloud://z'), makeLoadedSkill('alpha', 'cloud://a')] }, + ], + ['cloud', 'local'] + ); + + expect(merged.map((s) => s.manifest.name)).toEqual(['alpha', 'zeta']); + }); + }); + + describe('cloudInstalledSkillToLoaded', () => { + it('converts cloud installed skill shape into runtime loaded skill', () => { + const loaded = cloudInstalledSkillToLoaded({ + installationId: 'inst-1', + userId: 'user-1', + enabled: true, + userConfig: {}, + versionPinned: null, + installedAt: '2026-01-01T00:00:00Z', + lastUsedAt: null, + usageCount: 0, + skillId: 'skill-1', + name: 'browser-check', + displayName: 'Browser Check', + description: 'Checks a site in browser', + type: 'guide', + category: 'qa', + tags: ['browser', 'qa'], + emoji: '🧪', + currentVersion: '1.0.0', + manifest: { + name: 'browser-check', + version: '1.0.0', + description: 'Checks a site in browser', + type: 'guide', + }, + content: '# Browser Check', + isOfficial: false, + isVerified: false, + author: null, + repositoryUrl: null, + resolvedContent: '# Browser Check', + resolvedManifest: { + name: 'browser-check', + version: '1.1.0', + description: 'Checks a site in browser', + type: 'guide', + }, + resolvedVersion: '1.1.0', + }); + + expect(loaded.manifest.name).toBe('browser-check'); + expect(loaded.manifest.version).toBe('1.1.0'); + expect(loaded.skillContent).toBe('# Browser Check'); + expect(loaded.sourcePath).toBe('cloud://skill-1'); + }); + }); +}); diff --git a/packages/api/src/skills/providers.ts b/packages/api/src/skills/providers.ts new file mode 100644 index 00000000..e30d0518 --- /dev/null +++ b/packages/api/src/skills/providers.ts @@ -0,0 +1,123 @@ +/** + * Skill Source Providers + * + * Phase 1 provider abstraction for loading user skills from deterministic sources. + * Current providers: + * - cloud: PCP skills registry installations + * - local: ~/.pcp/skills filesystem directory + */ + +import { SkillsRepository } from './repository'; +import { loadAllSkills as loadLocalSkills } from './loader'; +import { checkEligibility } from './eligibility'; +import type { LoadedSkill, SkillManifest, SkillType, UserInstalledSkill } from './types'; + +export type SkillSourceId = 'cloud' | 'local'; + +export const DEFAULT_SKILL_SOURCE_PRIORITY: SkillSourceId[] = ['cloud', 'local']; + +export interface SkillSourceProvider { + id: SkillSourceId; + loadUserSkills(userId: string): Promise; +} + +export interface SkillSourceLoadResult { + source: SkillSourceId; + skills: LoadedSkill[]; +} + +/** + * cloud:// installed skill -> runtime LoadedSkill + */ +export function cloudInstalledSkillToLoaded(cloud: UserInstalledSkill): LoadedSkill { + const manifest: SkillManifest = { + ...cloud.resolvedManifest, + name: cloud.name, + version: cloud.resolvedVersion, + displayName: cloud.displayName, + description: cloud.description, + type: cloud.type as SkillType, + category: cloud.category || undefined, + tags: cloud.tags, + emoji: cloud.emoji || undefined, + }; + + return { + manifest, + skillContent: cloud.resolvedContent, + sourcePath: `cloud://${cloud.skillId}`, + eligibility: checkEligibility(manifest.requirements), + }; +} + +/** + * Local filesystem provider (~/.pcp/skills + builtin) + */ +export class LocalSkillSourceProvider implements SkillSourceProvider { + readonly id: SkillSourceId = 'local'; + + constructor(private readonly userSkillsPath?: string) {} + + async loadUserSkills(_userId: string): Promise { + return loadLocalSkills(this.userSkillsPath); + } +} + +/** + * Cloud installations provider (Supabase-backed) + */ +export class CloudSkillSourceProvider implements SkillSourceProvider { + readonly id: SkillSourceId = 'cloud'; + + constructor(private readonly repository: SkillsRepository) {} + + async loadUserSkills(userId: string): Promise { + const cloudSkills = await this.repository.getUserInstalledSkills(userId); + return cloudSkills.map(cloudInstalledSkillToLoaded); + } +} + +/** + * Deterministic skill merge by source priority. + * + * Rules: + * - sources are applied in listed priority order + * - later sources override earlier sources on name collision + * - output sorted by skill name for deterministic ordering + */ +export function mergeSkillsByPriority( + results: SkillSourceLoadResult[], + priority: SkillSourceId[] +): LoadedSkill[] { + const resultBySource = new Map(results.map((result) => [result.source, result.skills])); + const orderedPriority: SkillSourceId[] = []; + const seen = new Set(); + + // Apply explicit priority first + for (const source of priority) { + if (!seen.has(source)) { + seen.add(source); + orderedPriority.push(source); + } + } + + // Apply any missing sources in deterministic source-id order + const missingSources = Array.from(resultBySource.keys()) + .filter((source) => !seen.has(source)) + .sort(); + for (const source of missingSources) { + orderedPriority.push(source); + } + + const merged = new Map(); + + for (const source of orderedPriority) { + const skills = resultBySource.get(source) ?? []; + const sortedSkills = [...skills].sort((a, b) => a.manifest.name.localeCompare(b.manifest.name)); + for (const skill of sortedSkills) { + merged.set(skill.manifest.name, skill); + } + } + + return Array.from(merged.values()).sort((a, b) => a.manifest.name.localeCompare(b.manifest.name)); +}