diff --git a/.env.example b/.env.example index 333699c..ff0c7d7 100644 --- a/.env.example +++ b/.env.example @@ -1,4 +1,19 @@ -# ── Ollama (local / self-hosted) ──────────────────────────────────────────────── +# ── Run mode ──────────────────────────────────────────────────────────────── +# gateway (default) starts HTTP API + streaming endpoint + simple web UI +# single runs one-shot mode (PROMPT/stdin) and exits +RUN_MODE=gateway + +# ── Gateway server ─────────────────────────────────────────────────────────── +GATEWAY_HOST=0.0.0.0 +GATEWAY_PORT=8787 +# Set to * or a specific origin for browser apps hosted elsewhere +# GATEWAY_CORS_ORIGIN=* +# Optional bearer token required on every API request +# GATEWAY_AUTH_TOKEN=replace-me +# Disable built-in web UI at / +GATEWAY_ENABLE_WEB_UI=true + +# ── Ollama (local / self-hosted) ──────────────────────────────────────────── # To use a local Ollama instance, override PROVIDER and MODEL: # PROVIDER=ollama # MODEL=llama3.2 # or qwen2.5-coder, mistral, deepseek-r1:14b … @@ -12,7 +27,7 @@ # Locally (no Docker): change to http://localhost:11434/v1 # # OLLAMA_BASE_URL=http://host.docker.internal:11434/v1 -# OLLAMA_CONTEXT_WINDOW=32768 # tokens passed as num_ctx +# OLLAMA_CONTEXT_WINDOW=32768 # OLLAMA_MAX_TOKENS=8192 # ── Model ──────────────────────────────────────────────────────────────────── @@ -22,7 +37,7 @@ PROVIDER=anthropic MODEL=claude-sonnet-4-20250514 -# ── Authentication ──────────────────────────────────────────────────────────── +# ── Authentication ─────────────────────────────────────────────────────────── # Generic API key for the selected PROVIDER (injected at runtime, not stored). API_KEY=sk-ant-... @@ -36,36 +51,37 @@ API_KEY=sk-ant-... # XAI_API_KEY=... # OPENROUTER_API_KEY=... -# ── Prompt ──────────────────────────────────────────────────────────────────── -# The user message to send. Alternatively pipe via stdin. -PROMPT=List all .ts files in the current directory. +# ── Prompt (single mode only) ─────────────────────────────────────────────── +# Used only when RUN_MODE=single. +# PROMPT=List all .ts files in the current directory. -# ── System prompt ───────────────────────────────────────────────────────────── +# ── System prompt ──────────────────────────────────────────────────────────── # Completely replaces the default system prompt (optional). # SYSTEM_PROMPT=You are a helpful assistant. Work inside /app only. # Appended to the (possibly overridden) system prompt (optional). # APPEND_SYSTEM_PROMPT=Always answer in English. -# ── Thinking ────────────────────────────────────────────────────────────────── -# off | minimal | low | medium | high | xhigh (default: off) +# ── Thinking ───────────────────────────────────────────────────────────────── +# off | minimal | low | medium | high | xhigh THINKING_LEVEL=off -# ── Tools ───────────────────────────────────────────────────────────────────── -# all → read, bash, edit, write (default) -# readonly → read only +# ── Tools ──────────────────────────────────────────────────────────────────── +# all → read, bash, edit, write (default) +# readonly → read, grep, find, ls # none → no built-in tools -# Or a comma-separated list: read,bash +# Or a comma-separated subset of coding tools: read,bash,edit,write TOOLS=all -# ── Working directory ───────────────────────────────────────────────────────── +# ── Working directory ──────────────────────────────────────────────────────── # Directory the agent reads/writes. Maps to the Docker volume mount point. CWD=/app -# ── Session persistence ─────────────────────────────────────────────────────── -# true → persist session files under CWD; false/unset → in-memory (default) -SESSION_PERSIST=false +# ── Session persistence ────────────────────────────────────────────────────── +# true → persist agent sessions and gateway conversation index under CWD/.gateway +# false → in-memory sessions only +SESSION_PERSIST=true -# ── Verbose tool logging ────────────────────────────────────────────────────── +# ── Verbose tool logging ───────────────────────────────────────────────────── # true → log tool start/end events to stderr VERBOSE_TOOLS=false diff --git a/.gitignore b/.gitignore index 1e56f98..acee971 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,6 @@ node_modules/ +dist/ data/** +.pi/ +.DS_Store .env \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index b4c8e88..250fa56 100644 --- a/Dockerfile +++ b/Dockerfile @@ -34,5 +34,11 @@ VOLUME ["/app"] ENV NODE_ENV=production # Default working directory for the agent (overridable via CWD env var) ENV CWD=/app +# Gateway defaults +ENV RUN_MODE=gateway +ENV GATEWAY_HOST=0.0.0.0 +ENV GATEWAY_PORT=8787 + +EXPOSE 8787 CMD ["node", "dist/index.js"] \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..ee216d2 --- /dev/null +++ b/README.md @@ -0,0 +1,130 @@ +# MoA Agent Gateway + +This project now runs as a **chat gateway** in front of `@mariozechner/pi-coding-agent`. + +It supports: +- long-lived agent conversations +- HTTP JSON API +- streaming responses via Server-Sent Events (SSE) +- a built-in browser chat UI at `/` + +You can also keep one-shot mode (`RUN_MODE=single`) for script usage. + +--- + +## Run + +```bash +cp .env.example .env +# set provider/model/key in .env + +docker compose up --build +``` + +Gateway default URL: +- `http://localhost:8787` + +Health check: + +```bash +curl http://localhost:8787/health +``` + +One-shot mode (legacy behavior): + +```bash +RUN_MODE=single PROMPT="List files" npm start +``` + +--- + +## API + +### Create/list conversations + +```bash +curl -X POST http://localhost:8787/v1/conversations \ + -H 'content-type: application/json' \ + -d '{"conversationId":"web:user-42:chat-a"}' + +curl http://localhost:8787/v1/conversations +``` + +### Chat (non-streaming) + +```bash +curl -X POST http://localhost:8787/v1/chat \ + -H 'content-type: application/json' \ + -d '{"conversationId":"web:user-42:chat-a","message":"Summarize this repo"}' +``` + +### Chat (streaming SSE) + +```bash +curl -N -X POST http://localhost:8787/v1/chat/stream \ + -H 'content-type: application/json' \ + -d '{"conversationId":"web:user-42:chat-a","message":"List TypeScript files"}' +``` + +### Adapter-friendly endpoint (Slack/Matrix/etc) + +Instead of constructing `conversationId` in your adapter, you can call: + +- `POST /v1/adapters/chat` +- `POST /v1/adapters/chat/stream` + +Body fields: + +```json +{ + "source": "slack", + "workspaceId": "T123", + "channelId": "C456", + "threadId": "1712233.991", + "userId": "U789", + "message": "Summarize the thread" +} +``` + +The gateway derives a stable conversation id: +`source:workspaceId:channelId:threadId` + +SSE events include: +- `assistant_text_delta` +- `assistant_thinking_delta` +- `tool_start`, `tool_update`, `tool_end` +- `agent_start`, `agent_end` +- `done` (final response payload) +- `error` + +--- + +## Built-in Web UI + +Open: + +- `http://localhost:8787/` + +The UI stores `conversationId` in local storage and reuses it to keep context. + +--- + +## Auth and CORS + +Optional env vars: +- `GATEWAY_AUTH_TOKEN`: require `Authorization: Bearer ` +- `GATEWAY_CORS_ORIGIN`: e.g. `*` or `https://my-ui.example` + +--- + +## Multi-client strategy (Slack / Matrix / custom UI) + +The gateway is transport-agnostic. For each upstream client, map its thread identity to a stable `conversationId`, for example: + +- Web: `web::` +- Slack: `slack:::` +- Matrix: `matrix::` + +Then post messages to `/v1/chat` or `/v1/chat/stream` with that `conversationId`. + +This keeps one agent session per thread across transports. diff --git a/compose.yaml b/compose.yaml index 4edcddc..f43540f 100644 --- a/compose.yaml +++ b/compose.yaml @@ -2,7 +2,21 @@ services: agent: # image: agent:latest build: . + ports: + - "${GATEWAY_PORT:-8787}:${GATEWAY_PORT:-8787}" environment: + # ── Run mode ─────────────────────────────────────────────────────────── + # gateway (default) starts the HTTP gateway + web UI + # single runs one-shot prompt mode (PROMPT/stdin) and exits + - RUN_MODE=${RUN_MODE:-gateway} + + # ── Gateway server ───────────────────────────────────────────────────── + - GATEWAY_HOST=0.0.0.0 + - GATEWAY_PORT=${GATEWAY_PORT:-8787} + - GATEWAY_CORS_ORIGIN=${GATEWAY_CORS_ORIGIN:-} + - GATEWAY_AUTH_TOKEN=${GATEWAY_AUTH_TOKEN:-} + - GATEWAY_ENABLE_WEB_UI=${GATEWAY_ENABLE_WEB_UI:-true} + # ── Model ────────────────────────────────────────────────────────────── # Both PROVIDER and MODEL must be set together to select a specific model. # If omitted, the agent uses the first available / settings model. @@ -32,7 +46,8 @@ services: - OLLAMA_CONTEXT_WINDOW=${OLLAMA_CONTEXT_WINDOW:-32768} - OLLAMA_MAX_TOKENS=${OLLAMA_MAX_TOKENS:-8192} - # ── Prompt ───────────────────────────────────────────────────────────── + # ── Prompt (single mode only) ───────────────────────────────────────── + # Used only when RUN_MODE=single. - PROMPT=${PROMPT:-} # ── System prompt ────────────────────────────────────────────────────── @@ -51,7 +66,7 @@ services: - CWD=/app # ── Session persistence ──────────────────────────────────────────────── - - SESSION_PERSIST=${SESSION_PERSIST:-false} + - SESSION_PERSIST=${SESSION_PERSIST:-true} # ── Verbose tool logging ─────────────────────────────────────────────── - VERBOSE_TOOLS=${VERBOSE_TOOLS:-false} diff --git a/package.json b/package.json index 0e9ff33..bc4bf2c 100644 --- a/package.json +++ b/package.json @@ -7,6 +7,8 @@ "scripts": { "build": "tsc", "start": "node dist/index.js", + "start:single": "RUN_MODE=single node dist/index.js", + "start:gateway": "RUN_MODE=gateway node dist/index.js", "dev": "tsx src/index.ts" }, "dependencies": { diff --git a/src/agent-session-factory.ts b/src/agent-session-factory.ts new file mode 100644 index 0000000..0623c1d --- /dev/null +++ b/src/agent-session-factory.ts @@ -0,0 +1,158 @@ +import { getModel, type Model } from "@mariozechner/pi-ai"; +import type { AgentTool, ThinkingLevel } from "@mariozechner/pi-agent-core"; +import { + AuthStorage, + createAgentSession, + createBashTool, + createCodingTools, + createEditTool, + createReadOnlyTools, + createReadTool, + createWriteTool, + DefaultResourceLoader, + ModelRegistry, + type AgentSession, + type SessionManager, +} from "@mariozechner/pi-coding-agent"; +import type { AgentConfig } from "./config.js"; + +function resolveModel(config: AgentConfig): Model | undefined { + const providerName = config.providerName; + const modelId = config.modelId; + + if (providerName?.toLowerCase() === "ollama") { + if (!modelId) { + throw new Error("PROVIDER=ollama requires MODEL to be set."); + } + + return { + id: modelId, + name: modelId, + api: "openai-completions", + provider: "ollama", + baseUrl: config.ollamaBaseUrl, + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: config.ollamaContextWindow, + maxTokens: config.ollamaMaxTokens, + compat: { + supportsStore: false, + supportsDeveloperRole: false, + supportsReasoningEffort: false, + supportsStrictMode: false, + maxTokensField: "max_tokens", + }, + } satisfies Model<"openai-completions">; + } + + if (providerName && modelId) { + const model = (getModel as (provider: string, model: string) => Model | undefined)( + providerName, + modelId, + ); + if (!model) { + throw new Error( + `Model "${providerName}/${modelId}" not found. Run \"pi --list-models\" to inspect available models.`, + ); + } + return model; + } + + if (providerName || modelId) { + console.warn( + "[agent] PROVIDER and MODEL must be set together. Falling back to default model resolution.", + ); + } + + return undefined; +} + +function resolveTools(cwd: string, toolsEnv: string): AgentTool[] { + const normalized = toolsEnv.toLowerCase().trim(); + + if (normalized === "all") { + return createCodingTools(cwd); + } + + if (normalized === "readonly" || normalized === "read-only") { + return createReadOnlyTools(cwd); + } + + if (normalized === "none") { + return []; + } + + const customFactories: Record AgentTool> = { + read: createReadTool, + bash: createBashTool, + edit: createEditTool, + write: createWriteTool, + }; + + const tools = normalized + .split(",") + .map((name) => name.trim()) + .filter(Boolean) + .map((name) => { + const factory = customFactories[name]; + if (!factory) { + console.warn(`[agent] Unknown tool \"${name}\", ignoring.`); + return undefined; + } + return factory(cwd); + }) + .filter((tool): tool is AgentTool => tool !== undefined); + + return tools; +} + +export class AgentSessionFactory { + constructor(private readonly config: AgentConfig) {} + + async createSession(sessionManager: SessionManager): Promise { + const authStorage = AuthStorage.create(); + + if (this.config.providerName && this.config.apiKey) { + authStorage.setRuntimeApiKey(this.config.providerName, this.config.apiKey); + } + + if (this.config.providerName?.toLowerCase() === "ollama" && !this.config.apiKey) { + authStorage.setRuntimeApiKey("ollama", "ollama"); + } + + const modelRegistry = new ModelRegistry(authStorage); + const model = resolveModel(this.config); + const tools = resolveTools(this.config.cwd, this.config.toolsEnv); + + let resourceLoader: DefaultResourceLoader | undefined; + if (this.config.systemPrompt !== undefined || this.config.appendSystemPrompt !== undefined) { + resourceLoader = new DefaultResourceLoader({ + cwd: this.config.cwd, + systemPromptOverride: (base: string | undefined) => { + let result = this.config.systemPrompt !== undefined ? this.config.systemPrompt : base; + if (this.config.appendSystemPrompt) { + result = result + ? `${result}\n\n${this.config.appendSystemPrompt}` + : this.config.appendSystemPrompt; + } + return result; + }, + }); + await resourceLoader.reload(); + } + + const { session } = await createAgentSession({ + ...(model ? { model } : {}), + ...(resourceLoader ? { resourceLoader } : {}), + thinkingLevel: this.config.thinkingLevel as ThinkingLevel, + tools, + sessionManager, + authStorage, + modelRegistry, + cwd: this.config.cwd, + }); + + return session; + } +} diff --git a/src/config.ts b/src/config.ts new file mode 100644 index 0000000..a674c0f --- /dev/null +++ b/src/config.ts @@ -0,0 +1,129 @@ +import path from "node:path"; + +export type ThinkingLevel = "off" | "minimal" | "low" | "medium" | "high" | "xhigh"; + +export interface AgentConfig { + providerName?: string; + modelId?: string; + apiKey?: string; + ollamaBaseUrl: string; + ollamaContextWindow: number; + ollamaMaxTokens: number; + thinkingLevel: ThinkingLevel; + toolsEnv: string; + systemPrompt?: string; + appendSystemPrompt?: string; + promptText?: string; + cwd: string; + sessionPersist: boolean; + verboseTools: boolean; +} + +export interface GatewayConfig { + runMode: "gateway" | "single"; + host: string; + port: number; + corsOrigin?: string; + authToken?: string; + enableWebUi: boolean; +} + +export interface AppConfig { + agent: AgentConfig; + gateway: GatewayConfig; +} + +function env(name: string): string | undefined { + const value = process.env[name]; + if (value === undefined || value === "") { + return undefined; + } + return value; +} + +function parseIntEnv(name: string, fallback: number): number { + const value = env(name); + if (!value) { + return fallback; + } + const parsed = Number.parseInt(value, 10); + if (Number.isNaN(parsed)) { + throw new Error(`Environment variable ${name} must be a valid integer.`); + } + return parsed; +} + +function parseBoolEnv(name: string, fallback: boolean): boolean { + const value = env(name); + if (!value) { + return fallback; + } + switch (value.toLowerCase()) { + case "1": + case "true": + case "yes": + case "on": + return true; + case "0": + case "false": + case "no": + case "off": + return false; + default: + throw new Error(`Environment variable ${name} must be a boolean (true/false).`); + } +} + +function resolveThinkingLevel(value: string | undefined): ThinkingLevel { + const normalized = (value ?? "off").toLowerCase(); + const valid: ThinkingLevel[] = ["off", "minimal", "low", "medium", "high", "xhigh"]; + if (!valid.includes(normalized as ThinkingLevel)) { + throw new Error( + `Invalid THINKING_LEVEL "${value}". Expected one of: ${valid.join(", ")}.`, + ); + } + return normalized as ThinkingLevel; +} + +export function loadConfig(): AppConfig { + const cwd = path.resolve(env("CWD") ?? "/app"); + + const runModeRaw = (env("RUN_MODE") ?? "gateway").toLowerCase(); + const runMode = runModeRaw === "single" ? "single" : "gateway"; + + return { + agent: { + providerName: env("PROVIDER"), + modelId: env("MODEL"), + apiKey: env("API_KEY"), + ollamaBaseUrl: env("OLLAMA_BASE_URL") ?? "http://host.docker.internal:11434/v1", + ollamaContextWindow: parseIntEnv("OLLAMA_CONTEXT_WINDOW", 32768), + ollamaMaxTokens: parseIntEnv("OLLAMA_MAX_TOKENS", 8192), + thinkingLevel: resolveThinkingLevel(env("THINKING_LEVEL")), + toolsEnv: env("TOOLS") ?? "all", + systemPrompt: env("SYSTEM_PROMPT"), + appendSystemPrompt: env("APPEND_SYSTEM_PROMPT"), + promptText: env("PROMPT"), + cwd, + sessionPersist: parseBoolEnv("SESSION_PERSIST", false), + verboseTools: parseBoolEnv("VERBOSE_TOOLS", false), + }, + gateway: { + runMode, + host: env("GATEWAY_HOST") ?? "0.0.0.0", + port: parseIntEnv("GATEWAY_PORT", 8787), + corsOrigin: env("GATEWAY_CORS_ORIGIN"), + authToken: env("GATEWAY_AUTH_TOKEN"), + enableWebUi: parseBoolEnv("GATEWAY_ENABLE_WEB_UI", true), + }, + }; +} + +export function readStdin(): Promise { + return new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + process.stdin.on("data", (chunk) => chunks.push(chunk)); + process.stdin.on("end", () => resolve(Buffer.concat(chunks).toString("utf8").trim())); + process.stdin.on("error", reject); + }); +} diff --git a/src/conversation-manager.ts b/src/conversation-manager.ts new file mode 100644 index 0000000..1ad0ba2 --- /dev/null +++ b/src/conversation-manager.ts @@ -0,0 +1,410 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import { randomUUID } from "node:crypto"; +import { + SessionManager, + type AgentSession, + type AgentSessionEvent, +} from "@mariozechner/pi-coding-agent"; +import type { ImageContent } from "@mariozechner/pi-ai"; +import type { AgentConfig } from "./config.js"; +import { AgentSessionFactory } from "./agent-session-factory.js"; + +export interface ChatRequest { + conversationId?: string; + message: string; + images?: ImageContent[]; + streamingBehavior?: "steer" | "followUp"; +} + +export interface ChatExecutionOptions { + includeEvents?: boolean; + onEvent?: (event: AgentSessionEvent) => void; +} + +export interface ChatResult { + conversationId: string; + sessionId: string; + sessionFile?: string; + assistantText: string; + eventCount: number; + events?: AgentSessionEvent[]; +} + +export interface ConversationSummary { + id: string; + createdAt: string; + updatedAt: string; + sessionId?: string; + sessionFile?: string; + isLoaded: boolean; + isStreaming: boolean; +} + +interface PersistedConversation { + id: string; + createdAt: string; + updatedAt: string; + sessionFile?: string; + sessionId?: string; +} + +interface PersistedConversationIndex { + version: 1; + conversations: PersistedConversation[]; +} + +interface ConversationRecord extends PersistedConversation { + queue: Promise; + session?: AgentSession; + sessionInit?: Promise; +} + +export class ConversationManager { + private readonly sessionFactory: AgentSessionFactory; + private readonly conversations = new Map(); + private initialized = false; + private persistQueue: Promise = Promise.resolve(); + + private readonly gatewayRootDir: string; + private readonly sessionDir: string; + private readonly indexPath: string; + + constructor(private readonly config: AgentConfig) { + this.sessionFactory = new AgentSessionFactory(config); + this.gatewayRootDir = path.join(this.config.cwd, ".gateway"); + this.sessionDir = path.join(this.gatewayRootDir, "sessions"); + this.indexPath = path.join(this.gatewayRootDir, "conversations.json"); + } + + async init(): Promise { + if (this.initialized) { + return; + } + + if (!this.config.sessionPersist) { + this.initialized = true; + return; + } + + try { + const raw = await fs.readFile(this.indexPath, "utf8"); + const parsed = JSON.parse(raw) as PersistedConversationIndex; + if (parsed.version !== 1 || !Array.isArray(parsed.conversations)) { + throw new Error("Unsupported conversation index format."); + } + + for (const item of parsed.conversations) { + if (!item.id || typeof item.id !== "string") { + continue; + } + this.conversations.set(item.id, { + id: item.id, + createdAt: item.createdAt ?? new Date().toISOString(), + updatedAt: item.updatedAt ?? new Date().toISOString(), + sessionFile: item.sessionFile, + sessionId: item.sessionId, + queue: Promise.resolve(), + }); + } + + console.error(`[gateway] Restored ${this.conversations.size} persisted conversation(s).`); + } catch (error) { + const nodeError = error as NodeJS.ErrnoException; + if (nodeError.code !== "ENOENT") { + console.warn(`[gateway] Failed to load conversation index: ${nodeError.message}`); + } + } + + this.initialized = true; + } + + listConversations(): ConversationSummary[] { + return [...this.conversations.values()] + .map((record) => this.toSummary(record)) + .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt)); + } + + getConversation(conversationId: string): ConversationSummary | undefined { + const record = this.conversations.get(conversationId); + return record ? this.toSummary(record) : undefined; + } + + async createConversation(preferredId?: string): Promise { + await this.init(); + const id = this.resolveConversationId(preferredId); + const record = this.getOrCreateRecord(id); + await this.persistIndex(); + return this.toSummary(record); + } + + async deleteConversation(conversationId: string): Promise { + await this.init(); + const record = this.conversations.get(conversationId); + if (!record) { + return false; + } + + if (record.session) { + record.session.dispose(); + record.session = undefined; + } + + if (record.sessionFile && this.config.sessionPersist) { + try { + await fs.unlink(record.sessionFile); + } catch (error) { + const nodeError = error as NodeJS.ErrnoException; + if (nodeError.code !== "ENOENT") { + console.warn(`[gateway] Failed to delete session file ${record.sessionFile}: ${nodeError.message}`); + } + } + } + + this.conversations.delete(conversationId); + await this.persistIndex(); + return true; + } + + async abortConversation(conversationId: string): Promise { + await this.init(); + const record = this.conversations.get(conversationId); + if (!record?.session) { + return; + } + await record.session.abort(); + } + + async chat(request: ChatRequest, options: ChatExecutionOptions = {}): Promise { + await this.init(); + + const message = request.message.trim(); + if (!message) { + throw new Error("message must be a non-empty string."); + } + + const conversationId = this.resolveConversationId(request.conversationId); + const record = this.getOrCreateRecord(conversationId); + + return this.enqueue(record, async () => { + const session = await this.ensureSession(record); + const collectedEvents: AgentSessionEvent[] | undefined = options.includeEvents ? [] : undefined; + const assistantTextChunks: string[] = []; + + const unsubscribe = session.subscribe((event) => { + if (collectedEvents) { + collectedEvents.push(event); + } + options.onEvent?.(event); + + if (event.type === "message_update" && event.assistantMessageEvent.type === "text_delta") { + assistantTextChunks.push(event.assistantMessageEvent.delta); + } + }); + + try { + const streamingBehavior = request.streamingBehavior ?? (session.isStreaming ? "followUp" : undefined); + + await session.prompt(message, { + ...(request.images ? { images: request.images } : {}), + ...(streamingBehavior ? { streamingBehavior } : {}), + }); + } finally { + unsubscribe(); + } + + record.updatedAt = new Date().toISOString(); + record.sessionFile = session.sessionFile; + record.sessionId = session.sessionId; + await this.persistIndex(); + + const assistantText = assistantTextChunks.join("") || this.getLastAssistantText(session); + + return { + conversationId: record.id, + sessionId: session.sessionId, + sessionFile: session.sessionFile, + assistantText, + eventCount: collectedEvents?.length ?? 0, + ...(collectedEvents ? { events: collectedEvents } : {}), + }; + }); + } + + async shutdown(): Promise { + for (const record of this.conversations.values()) { + if (record.session) { + record.session.dispose(); + record.session = undefined; + } + } + await this.persistIndex(); + } + + private toSummary(record: ConversationRecord): ConversationSummary { + return { + id: record.id, + createdAt: record.createdAt, + updatedAt: record.updatedAt, + sessionId: record.session?.sessionId ?? record.sessionId, + sessionFile: record.session?.sessionFile ?? record.sessionFile, + isLoaded: record.session !== undefined, + isStreaming: record.session?.isStreaming ?? false, + }; + } + + private resolveConversationId(preferredId?: string): string { + if (!preferredId) { + return randomUUID(); + } + + const normalized = preferredId.trim(); + if (!normalized) { + return randomUUID(); + } + + if (normalized.length > 200) { + throw new Error("conversationId must be 200 characters or fewer."); + } + + if (/\r|\n/.test(normalized)) { + throw new Error("conversationId must not contain newlines."); + } + + return normalized; + } + + private getOrCreateRecord(conversationId: string): ConversationRecord { + const existing = this.conversations.get(conversationId); + if (existing) { + return existing; + } + + const now = new Date().toISOString(); + const created: ConversationRecord = { + id: conversationId, + createdAt: now, + updatedAt: now, + queue: Promise.resolve(), + }; + + this.conversations.set(conversationId, created); + return created; + } + + private enqueue(record: ConversationRecord, task: () => Promise): Promise { + const run = record.queue.then(task, task); + record.queue = run.then( + () => undefined, + () => undefined, + ); + return run; + } + + private async ensureSession(record: ConversationRecord): Promise { + if (record.session) { + return record.session; + } + + if (!record.sessionInit) { + record.sessionInit = this.createSessionForRecord(record) + .then((session) => { + record.session = session; + return session; + }) + .finally(() => { + record.sessionInit = undefined; + }); + } + + return record.sessionInit; + } + + private async createSessionForRecord(record: ConversationRecord): Promise { + let sessionManager: SessionManager; + + if (!this.config.sessionPersist) { + sessionManager = SessionManager.inMemory(); + } else if (record.sessionFile) { + try { + sessionManager = SessionManager.open(record.sessionFile); + } catch (error) { + console.warn( + `[gateway] Failed to open persisted session ${record.sessionFile}, creating a new session.`, + error, + ); + sessionManager = SessionManager.create(this.config.cwd, this.sessionDir); + } + } else { + sessionManager = SessionManager.create(this.config.cwd, this.sessionDir); + } + + const session = await this.sessionFactory.createSession(sessionManager); + + if (this.config.verboseTools) { + session.subscribe((event) => { + if (event.type === "tool_execution_start") { + process.stderr.write(`\n[${record.id}] [tool:start] ${event.toolName}\n`); + } else if (event.type === "tool_execution_end") { + process.stderr.write(`[${record.id}] [tool:end] ${event.toolName}\n`); + } + }); + } + + record.sessionFile = session.sessionFile; + record.sessionId = session.sessionId; + + await this.persistIndex(); + + return session; + } + + private getLastAssistantText(session: AgentSession): string { + for (let i = session.messages.length - 1; i >= 0; i -= 1) { + const message = session.messages[i]; + if (message.role !== "assistant") { + continue; + } + + return message.content + .filter((contentPart): contentPart is { type: "text"; text: string } => contentPart.type === "text") + .map((contentPart) => contentPart.text) + .join(""); + } + + return ""; + } + + private async persistIndex(): Promise { + if (!this.config.sessionPersist) { + return; + } + + this.persistQueue = this.persistQueue + .catch((error) => { + console.warn("[gateway] Previous persistence operation failed:", error); + }) + .then(async () => { + await fs.mkdir(this.gatewayRootDir, { recursive: true }); + await fs.mkdir(this.sessionDir, { recursive: true }); + + const payload: PersistedConversationIndex = { + version: 1, + conversations: [...this.conversations.values()].map((record) => ({ + id: record.id, + createdAt: record.createdAt, + updatedAt: record.updatedAt, + sessionFile: record.session?.sessionFile ?? record.sessionFile, + sessionId: record.session?.sessionId ?? record.sessionId, + })), + }; + + const serialized = `${JSON.stringify(payload, null, 2)}\n`; + const tempPath = `${this.indexPath}.tmp`; + await fs.writeFile(tempPath, serialized, "utf8"); + await fs.rename(tempPath, this.indexPath); + }); + + await this.persistQueue; + } +} diff --git a/src/gateway/events.ts b/src/gateway/events.ts new file mode 100644 index 0000000..a6e2162 --- /dev/null +++ b/src/gateway/events.ts @@ -0,0 +1,119 @@ +import type { AgentSessionEvent } from "@mariozechner/pi-coding-agent"; + +export interface GatewayEvent { + type: string; + data: Record; +} + +export function toGatewayEvents(event: AgentSessionEvent): GatewayEvent[] { + switch (event.type) { + case "message_update": { + if (event.assistantMessageEvent.type === "text_delta") { + return [ + { + type: "assistant_text_delta", + data: { delta: event.assistantMessageEvent.delta }, + }, + ]; + } + + if (event.assistantMessageEvent.type === "thinking_delta") { + return [ + { + type: "assistant_thinking_delta", + data: { delta: event.assistantMessageEvent.delta }, + }, + ]; + } + + return [ + { + type: "assistant_message_update", + data: { updateType: event.assistantMessageEvent.type }, + }, + ]; + } + + case "tool_execution_start": + return [ + { + type: "tool_start", + data: { + toolName: event.toolName, + toolCallId: event.toolCallId, + }, + }, + ]; + + case "tool_execution_update": + return [ + { + type: "tool_update", + data: { + toolName: event.toolName, + toolCallId: event.toolCallId, + }, + }, + ]; + + case "tool_execution_end": + return [ + { + type: "tool_end", + data: { + toolName: event.toolName, + toolCallId: event.toolCallId, + isError: event.isError, + }, + }, + ]; + + case "agent_start": + return [{ type: "agent_start", data: {} }]; + + case "agent_end": + return [{ type: "agent_end", data: {} }]; + + case "auto_retry_start": + return [ + { + type: "retry_start", + data: { + attempt: event.attempt, + maxAttempts: event.maxAttempts, + delayMs: event.delayMs, + }, + }, + ]; + + case "auto_retry_end": + return [ + { + type: "retry_end", + data: { + success: event.success, + attempt: event.attempt, + finalError: event.finalError, + }, + }, + ]; + + case "auto_compaction_start": + return [{ type: "compaction_start", data: { reason: event.reason } }]; + + case "auto_compaction_end": + return [ + { + type: "compaction_end", + data: { + aborted: event.aborted, + willRetry: event.willRetry, + errorMessage: event.errorMessage, + }, + }, + ]; + + default: + return [{ type: event.type, data: {} }]; + } +} diff --git a/src/gateway/server.ts b/src/gateway/server.ts new file mode 100644 index 0000000..e79daf3 --- /dev/null +++ b/src/gateway/server.ts @@ -0,0 +1,432 @@ +import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http"; +import { URL } from "node:url"; +import type { ImageContent } from "@mariozechner/pi-ai"; +import type { GatewayConfig } from "../config.js"; +import type { ChatRequest } from "../conversation-manager.js"; +import { ConversationManager } from "../conversation-manager.js"; +import { toGatewayEvents } from "./events.js"; +import { getWebUiHtml } from "./web-ui.js"; + +interface ChatRequestBody { + conversationId?: unknown; + message?: unknown; + images?: unknown; + streamingBehavior?: unknown; + includeEvents?: unknown; +} + +interface AdapterChatRequestBody extends ChatRequestBody { + source?: unknown; + workspaceId?: unknown; + channelId?: unknown; + threadId?: unknown; + userId?: unknown; +} + +class HttpError extends Error { + constructor( + readonly statusCode: number, + message: string, + ) { + super(message); + } +} + +export class GatewayHttpServer { + private readonly server: Server; + + constructor( + private readonly manager: ConversationManager, + private readonly config: GatewayConfig, + ) { + this.server = createServer((req, res) => { + this.handleRequest(req, res).catch((error) => { + if (error instanceof HttpError) { + this.sendJson(res, error.statusCode, { error: error.message }); + return; + } + + console.error("[gateway] Unhandled request error", error); + if (!res.headersSent) { + this.sendJson(res, 500, { error: "Internal server error" }); + } else { + res.end(); + } + }); + }); + } + + async start(): Promise { + await new Promise((resolve, reject) => { + this.server.once("error", reject); + this.server.listen(this.config.port, this.config.host, () => { + this.server.off("error", reject); + resolve(); + }); + }); + + console.error(`[gateway] Listening on http://${this.config.host}:${this.config.port}`); + } + + async stop(): Promise { + await new Promise((resolve, reject) => { + this.server.close((error) => { + if (error) { + reject(error); + return; + } + resolve(); + }); + }); + } + + private async handleRequest(req: IncomingMessage, res: ServerResponse): Promise { + if (this.handleCors(req, res)) { + return; + } + + if (!this.isAuthorized(req, res)) { + return; + } + + const url = new URL(req.url ?? "/", `http://${req.headers.host ?? "localhost"}`); + const { pathname } = url; + const method = req.method ?? "GET"; + + if (method === "GET" && pathname === "/health") { + this.sendJson(res, 200, { ok: true }); + return; + } + + if (method === "GET" && pathname === "/") { + if (!this.config.enableWebUi) { + this.sendJson(res, 404, { error: "Web UI disabled" }); + return; + } + + this.sendHtml(res, 200, getWebUiHtml()); + return; + } + + if (method === "GET" && pathname === "/v1/conversations") { + this.sendJson(res, 200, { conversations: this.manager.listConversations() }); + return; + } + + if (method === "POST" && pathname === "/v1/conversations") { + const body = (await this.readJsonBody(req)) as { conversationId?: unknown }; + if (body.conversationId !== undefined && typeof body.conversationId !== "string") { + throw new HttpError(400, "conversationId must be a string when provided."); + } + + const created = await this.manager.createConversation(body.conversationId as string | undefined); + this.sendJson(res, 201, created); + return; + } + + const conversationMatch = pathname.match(/^\/v1\/conversations\/([^/]+)$/); + if (method === "GET" && conversationMatch) { + const conversationId = this.decodePathSegment(conversationMatch[1]); + const conversation = this.manager.getConversation(conversationId); + if (!conversation) { + this.sendJson(res, 404, { error: "Conversation not found" }); + return; + } + this.sendJson(res, 200, conversation); + return; + } + + if (method === "DELETE" && conversationMatch) { + const conversationId = this.decodePathSegment(conversationMatch[1]); + const deleted = await this.manager.deleteConversation(conversationId); + this.sendJson(res, deleted ? 200 : 404, { deleted }); + return; + } + + const abortMatch = pathname.match(/^\/v1\/conversations\/([^/]+)\/abort$/); + if (method === "POST" && abortMatch) { + const conversationId = this.decodePathSegment(abortMatch[1]); + await this.manager.abortConversation(conversationId); + this.sendJson(res, 200, { aborted: true }); + return; + } + + if (method === "POST" && pathname === "/v1/adapters/chat") { + const body = (await this.readJsonBody(req)) as AdapterChatRequestBody; + const { chatRequest, adapterKey } = this.parseAdapterChatRequest(body); + const includeEvents = body.includeEvents === true; + + const result = await this.manager.chat(chatRequest, { includeEvents }); + this.sendJson(res, 200, { adapterKey, ...result }); + return; + } + + if (method === "POST" && pathname === "/v1/adapters/chat/stream") { + const body = (await this.readJsonBody(req)) as AdapterChatRequestBody; + const { chatRequest, adapterKey } = this.parseAdapterChatRequest(body); + await this.handleChatStream(req, res, chatRequest, { adapterKey }); + return; + } + + if (method === "POST" && pathname === "/v1/chat") { + const body = (await this.readJsonBody(req)) as ChatRequestBody; + const chatRequest = this.parseChatRequest(body); + const includeEvents = body.includeEvents === true; + + const result = await this.manager.chat(chatRequest, { includeEvents }); + this.sendJson(res, 200, result); + return; + } + + if (method === "POST" && pathname === "/v1/chat/stream") { + const body = (await this.readJsonBody(req)) as ChatRequestBody; + const chatRequest = this.parseChatRequest(body); + await this.handleChatStream(req, res, chatRequest); + return; + } + + this.sendJson(res, 404, { error: "Not found" }); + } + + private async handleChatStream( + req: IncomingMessage, + res: ServerResponse, + chatRequest: ChatRequest, + metadata: Record = {}, + ): Promise { + this.writeSseHeaders(res); + this.writeSse(res, "ready", { ok: true }); + + let finished = false; + const providedConversationId = chatRequest.conversationId; + + req.on("close", () => { + if (!finished && providedConversationId) { + void this.manager.abortConversation(providedConversationId).catch((error) => { + console.warn("[gateway] Failed to abort conversation after client disconnect", error); + }); + } + }); + + try { + const result = await this.manager.chat(chatRequest, { + onEvent: (event) => { + for (const gatewayEvent of toGatewayEvents(event)) { + this.writeSse(res, gatewayEvent.type, gatewayEvent.data); + } + }, + }); + + this.writeSse(res, "done", { + ...metadata, + conversationId: result.conversationId, + sessionId: result.sessionId, + sessionFile: result.sessionFile, + assistantText: result.assistantText, + }); + + finished = true; + res.end(); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + this.writeSse(res, "error", { message }); + finished = true; + res.end(); + } + } + + private parseChatRequest(body: ChatRequestBody): ChatRequest { + if (typeof body.message !== "string") { + throw new HttpError(400, "message is required and must be a string."); + } + + if (!body.message.trim()) { + throw new HttpError(400, "message must not be empty."); + } + + const conversationId = + typeof body.conversationId === "string" ? body.conversationId : undefined; + + if (conversationId) { + if (conversationId.length > 200) { + throw new HttpError(400, "conversationId must be 200 characters or fewer."); + } + if (/\r|\n/.test(conversationId)) { + throw new HttpError(400, "conversationId must not contain newlines."); + } + } + + let images: ImageContent[] | undefined; + if (body.images !== undefined) { + if (!Array.isArray(body.images)) { + throw new HttpError(400, "images must be an array when provided."); + } + images = body.images as ImageContent[]; + } + + let streamingBehavior: "steer" | "followUp" | undefined; + if (body.streamingBehavior !== undefined) { + if (body.streamingBehavior === "steer" || body.streamingBehavior === "followUp") { + streamingBehavior = body.streamingBehavior; + } else { + throw new HttpError(400, 'streamingBehavior must be "steer" or "followUp".'); + } + } + + return { + conversationId, + message: body.message, + ...(images ? { images } : {}), + ...(streamingBehavior ? { streamingBehavior } : {}), + }; + } + + private parseAdapterChatRequest(body: AdapterChatRequestBody): { + chatRequest: ChatRequest; + adapterKey: string; + } { + const source = this.parseOptionalSegment(body.source, "source") ?? "generic"; + const workspaceId = this.parseOptionalSegment(body.workspaceId, "workspaceId") ?? "default"; + const channelId = this.parseRequiredSegment(body.channelId, "channelId"); + const threadId = this.parseOptionalSegment(body.threadId, "threadId") ?? "root"; + const userId = this.parseOptionalSegment(body.userId, "userId") ?? "anonymous"; + + const normalizedBody: ChatRequestBody = { + ...body, + conversationId: `${source}:${workspaceId}:${channelId}:${threadId}`, + }; + + return { + adapterKey: `${source}:${workspaceId}:${channelId}:${threadId}:${userId}`, + chatRequest: this.parseChatRequest(normalizedBody), + }; + } + + private parseOptionalSegment(value: unknown, fieldName: string): string | undefined { + if (value === undefined || value === null) { + return undefined; + } + + if (typeof value !== "string") { + throw new HttpError(400, `${fieldName} must be a string when provided.`); + } + + const trimmed = value.trim(); + if (!trimmed) { + return undefined; + } + + if (trimmed.includes(":")) { + throw new HttpError(400, `${fieldName} must not contain ':'.`); + } + + return trimmed; + } + + private parseRequiredSegment(value: unknown, fieldName: string): string { + const segment = this.parseOptionalSegment(value, fieldName); + if (!segment) { + throw new HttpError(400, `${fieldName} is required.`); + } + return segment; + } + + private async readJsonBody(req: IncomingMessage, maxBytes = 1024 * 1024): Promise { + const chunks: Buffer[] = []; + let total = 0; + + for await (const chunk of req) { + const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + total += buffer.length; + if (total > maxBytes) { + throw new HttpError(413, `Request body exceeds ${maxBytes} bytes.`); + } + chunks.push(buffer); + } + + if (chunks.length === 0) { + return {}; + } + + const raw = Buffer.concat(chunks).toString("utf8").trim(); + if (!raw) { + return {}; + } + + try { + return JSON.parse(raw); + } catch { + throw new HttpError(400, "Request body must be valid JSON."); + } + } + + private handleCors(req: IncomingMessage, res: ServerResponse): boolean { + if (!this.config.corsOrigin) { + return false; + } + + res.setHeader("Access-Control-Allow-Origin", this.config.corsOrigin); + res.setHeader("Access-Control-Allow-Headers", "Content-Type, Authorization"); + res.setHeader("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS"); + + if (req.method === "OPTIONS") { + res.statusCode = 204; + res.end(); + return true; + } + + return false; + } + + private isAuthorized(req: IncomingMessage, res: ServerResponse): boolean { + if (!this.config.authToken) { + return true; + } + + const authHeader = req.headers.authorization; + if (authHeader === `Bearer ${this.config.authToken}`) { + return true; + } + + this.sendJson(res, 401, { error: "Unauthorized" }); + return false; + } + + private decodePathSegment(value: string): string { + try { + return decodeURIComponent(value); + } catch { + throw new HttpError(400, "Invalid path segment encoding."); + } + } + + private sendJson(res: ServerResponse, statusCode: number, body: unknown): void { + res.statusCode = statusCode; + res.setHeader("Content-Type", "application/json; charset=utf-8"); + res.end(`${JSON.stringify(body)}\n`); + } + + private sendHtml(res: ServerResponse, statusCode: number, html: string): void { + res.statusCode = statusCode; + res.setHeader("Content-Type", "text/html; charset=utf-8"); + res.end(html); + } + + private writeSseHeaders(res: ServerResponse): void { + res.statusCode = 200; + res.setHeader("Content-Type", "text/event-stream; charset=utf-8"); + res.setHeader("Cache-Control", "no-cache, no-transform"); + res.setHeader("Connection", "keep-alive"); + res.setHeader("X-Accel-Buffering", "no"); + } + + private writeSse(res: ServerResponse, event: string, data: unknown): void { + if (res.writableEnded || res.destroyed) { + return; + } + + const payload = typeof data === "string" ? data : JSON.stringify(data); + res.write(`event: ${event}\n`); + res.write(`data: ${payload}\n\n`); + } +} diff --git a/src/gateway/web-ui.ts b/src/gateway/web-ui.ts new file mode 100644 index 0000000..99120d5 --- /dev/null +++ b/src/gateway/web-ui.ts @@ -0,0 +1,296 @@ +export function getWebUiHtml(): string { + return ` + + + + + Pi Gateway Chat + + + +
+
+
Gateway endpoint: /v1/chat/stream
+
+ + +
+
+ + +
+
+ +
+ +
+
+ + +
+
+
Idle
+ +
+
+
+ + + +`; +} diff --git a/src/index.ts b/src/index.ts index ba22bdf..f3dc76c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,306 +1,45 @@ -/** - * MoA Agent – pi coding agent SDK wrapper - * - * All behaviour is driven by environment variables so the container - * works as a drop-in Docker service without any code changes. - * - * Environment variables - * ───────────────────── - * PROVIDER Provider name (anthropic | openai | google | mistral | - * groq | cerebras | xai | openrouter | ollama | …) - * When set together with MODEL, the exact model is resolved. - * If omitted the session uses whatever model is already in - * settings / the first available one. - * - * ── Ollama (local) ────────────────────────────────────── - * Set PROVIDER=ollama and MODEL= (e.g. llama3.2, - * qwen2.5-coder, mistral, deepseek-r1:14b …). - * No API key is needed. - * OLLAMA_BASE_URL URL of the Ollama HTTP API. - * Default in Docker: - * http://host.docker.internal:11434/v1 - * Default locally: - * http://localhost:11434/v1 - * OLLAMA_CONTEXT_WINDOW Context size in tokens (default: 32768) - * OLLAMA_MAX_TOKENS Max output tokens (default: 8192) - * - * MODEL Model ID as understood by getModel(), e.g. - * "claude-sonnet-4-20250514" or "gpt-4o". - * - * API_KEY Generic API key for the selected PROVIDER. - * Stored as a runtime (non-persisted) key. - * Provider-specific env vars (ANTHROPIC_API_KEY, - * OPENAI_API_KEY, GEMINI_API_KEY, …) are also honoured - * automatically by the underlying pi-ai library. - * Not required when PROVIDER=ollama. - * - * THINKING_LEVEL off | minimal | low | medium | high | xhigh (default: off) - * - * TOOLS all – read, bash, edit, write (default) - * readonly – read only - * none – no built-in tools - * Or a comma-separated list: read,bash,edit,write - * - * SYSTEM_PROMPT Completely replaces the default system prompt. - * APPEND_SYSTEM_PROMPT Appended to the (possibly overridden) system prompt. - * - * PROMPT The user prompt to send. If omitted the agent reads from - * stdin. Exactly one of the two must be provided. - * - * CWD Working directory the agent operates in. - * Defaults to /app (the Docker volume mount point). - * - * SESSION_PERSIST Set to "true" to persist the session under CWD. - * Defaults to in-memory (ephemeral). - * - * VERBOSE_TOOLS Set to "true" to log tool start/end events to stderr. - */ +import { loadConfig } from "./config.js"; +import { runSingleShot } from "./single-shot.js"; +import { ConversationManager } from "./conversation-manager.js"; +import { GatewayHttpServer } from "./gateway/server.js"; -import { getModel, type Model } from "@mariozechner/pi-ai"; -import { - AuthStorage, - createAgentSession, - DefaultResourceLoader, - ModelRegistry, - SessionManager, - codingTools, - readOnlyTools, - readTool, - bashTool, - editTool, - writeTool, -} from "@mariozechner/pi-coding-agent"; +async function run(): Promise { + const config = loadConfig(); -// ─── helpers ──────────────────────────────────────────────────────────────── - -function env(name: string): string | undefined { - const v = process.env[name]; - return v === "" ? undefined : v; -} - -function requiredEnv(name: string): string { - const v = env(name); - if (!v) { - console.error(`[agent] Required environment variable ${name} is not set.`); - process.exit(1); + if (config.gateway.runMode === "single") { + await runSingleShot(config.agent); + return; } - return v; -} -function readStdin(): Promise { - return new Promise((resolve, reject) => { - const chunks: Buffer[] = []; - process.stdin.on("data", (chunk) => chunks.push(chunk)); - process.stdin.on("end", () => resolve(Buffer.concat(chunks).toString("utf8").trim())); - process.stdin.on("error", reject); + const manager = new ConversationManager(config.agent); + await manager.init(); + + const server = new GatewayHttpServer(manager, config.gateway); + await server.start(); + + let shuttingDown = false; + const shutdown = async (signal: string): Promise => { + if (shuttingDown) { + return; + } + shuttingDown = true; + + console.error(`\n[gateway] Received ${signal}, shutting down...`); + await server.stop(); + await manager.shutdown(); + process.exit(0); + }; + + process.on("SIGINT", () => { + void shutdown("SIGINT"); + }); + + process.on("SIGTERM", () => { + void shutdown("SIGTERM"); }); } -// ─── configuration ────────────────────────────────────────────────────────── - -const providerName = env("PROVIDER"); -const modelId = env("MODEL"); -const apiKey = env("API_KEY"); -const ollamaBaseUrl = env("OLLAMA_BASE_URL") ?? "http://host.docker.internal:11434/v1"; -const ollamaCtxWin = parseInt(env("OLLAMA_CONTEXT_WINDOW") ?? "32768", 10); -const ollamaMaxTok = parseInt(env("OLLAMA_MAX_TOKENS") ?? "8192", 10); -const thinkingLevel = (env("THINKING_LEVEL") ?? "off") as - "off" | "minimal" | "low" | "medium" | "high" | "xhigh"; -const toolsEnv = env("TOOLS") ?? "all"; -const systemPrompt = env("SYSTEM_PROMPT"); -const appendPrompt = env("APPEND_SYSTEM_PROMPT"); -const promptText = env("PROMPT"); -const cwd = env("CWD") ?? "/app"; -const sessionPersist = env("SESSION_PERSIST") === "true"; -const verboseTools = env("VERBOSE_TOOLS") === "true"; - -// ─── auth & model registry ────────────────────────────────────────────────── - -const authStorage = AuthStorage.create(); - -// Inject the generic API_KEY for the chosen provider at runtime so it is -// never written to disk inside the container. -if (providerName && apiKey) { - authStorage.setRuntimeApiKey(providerName, apiKey); -} - -// Ollama doesn't require a real key; set a dummy to satisfy the auth layer. -if (providerName?.toLowerCase() === "ollama" && !apiKey) { - authStorage.setRuntimeApiKey("ollama", "ollama"); -} - -const modelRegistry = new ModelRegistry(authStorage); - -// ─── model resolution ─────────────────────────────────────────────────────── - -let model: Model | undefined; - -if (providerName?.toLowerCase() === "ollama") { - // Ollama is OpenAI-compatible; build a custom model object. - if (!modelId) { - console.error("[agent] PROVIDER=ollama requires MODEL to be set (e.g. MODEL=llama3.2)."); - process.exit(1); - } - model = { - id: modelId, - name: modelId, - api: "openai-completions", - provider: "ollama", - baseUrl: ollamaBaseUrl, - reasoning: false, - input: ["text"], - cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, - contextWindow: ollamaCtxWin, - maxTokens: ollamaMaxTok, - compat: { - supportsStore: false, - supportsDeveloperRole: false, - supportsReasoningEffort: false, - supportsStrictMode: false, - maxTokensField: "max_tokens", - }, - } satisfies Model<"openai-completions">; - console.error(`[agent] Using Ollama model "${modelId}" at ${ollamaBaseUrl}`); - -} else if (providerName && modelId) { - model = (getModel as (provider: string, model: string) => Model | undefined)(providerName, modelId); - if (!model) { - console.error( - `[agent] Model "${providerName}/${modelId}" not found. ` + - `Run "pi --list-models" to check available models.` - ); - process.exit(1); - } - console.error(`[agent] Using model: ${model.provider}/${model.id}`); - -} else if (providerName || modelId) { - console.error( - "[agent] Both PROVIDER and MODEL must be set together. " + - "Falling back to default model." - ); -} - -// ─── tools ────────────────────────────────────────────────────────────────── - -// eslint-disable-next-line @typescript-eslint/no-explicit-any -const toolMap: Record = { - read: readTool, - bash: bashTool, - edit: editTool, - write: writeTool, -}; - -let tools: typeof codingTools; - -switch (toolsEnv.toLowerCase()) { - case "all": - tools = codingTools; - break; - case "readonly": - case "read-only": - tools = readOnlyTools; - break; - case "none": - tools = []; - break; - default: - tools = toolsEnv - .split(",") - .map((t) => t.trim().toLowerCase()) - .filter(Boolean) - .map((name) => { - const tool = toolMap[name]; - if (!tool) { - console.error(`[agent] Unknown tool "${name}", ignoring.`); - return null; - } - return tool; - }) - .filter((t): t is typeof readTool => t !== null); -} - -// ─── resource loader (system prompt) ──────────────────────────────────────── - -let resourceLoader: DefaultResourceLoader | undefined; - -if (systemPrompt !== undefined || appendPrompt !== undefined) { - resourceLoader = new DefaultResourceLoader({ - systemPromptOverride: (base: string | undefined) => { - let result = systemPrompt !== undefined ? systemPrompt : base; - if (appendPrompt) { - result = result ? `${result}\n\n${appendPrompt}` : appendPrompt; - } - return result; - }, - }); - await resourceLoader.reload(); -} - -// ─── session manager ──────────────────────────────────────────────────────── - -const sessionManager = sessionPersist - ? SessionManager.create(cwd) - : SessionManager.inMemory(); - -// ─── create session ────────────────────────────────────────────────────────── - -const { session } = await createAgentSession({ - ...(model ? { model } : {}), - ...(resourceLoader ? { resourceLoader } : {}), - thinkingLevel, - tools, - sessionManager, - authStorage, - modelRegistry, - cwd, -}); - -// ─── event handling ────────────────────────────────────────────────────────── - -session.subscribe((event) => { - switch (event.type) { - case "message_update": - if (event.assistantMessageEvent.type === "text_delta") { - process.stdout.write(event.assistantMessageEvent.delta); - } - break; - - case "tool_execution_start": - if (verboseTools) { - process.stderr.write(`\n[tool:start] ${event.toolName}\n`); - } - break; - - case "tool_execution_end": - if (verboseTools) { - process.stderr.write(`[tool:end] ${event.toolName}\n`); - } - break; - - case "agent_end": - process.stdout.write("\n"); - break; - } -}); - -// ─── run prompt ────────────────────────────────────────────────────────────── - -let userPrompt: string; - -if (promptText) { - userPrompt = promptText; -} else if (!process.stdin.isTTY) { - userPrompt = await readStdin(); - if (!userPrompt) { - console.error("[agent] Stdin was empty. Set PROMPT or pipe a non-empty message."); - process.exit(1); - } -} else { - console.error( - "[agent] No prompt provided.\n" + - " Set the PROMPT environment variable, or pipe input via stdin." - ); +run().catch((error) => { + console.error("[app] Fatal error:", error); process.exit(1); -} - -await session.prompt(userPrompt); +}); diff --git a/src/single-shot.ts b/src/single-shot.ts new file mode 100644 index 0000000..ae563cd --- /dev/null +++ b/src/single-shot.ts @@ -0,0 +1,54 @@ +import { SessionManager } from "@mariozechner/pi-coding-agent"; +import type { AgentConfig } from "./config.js"; +import { readStdin } from "./config.js"; +import { AgentSessionFactory } from "./agent-session-factory.js"; + +export async function runSingleShot(config: AgentConfig): Promise { + const sessionManager = config.sessionPersist + ? SessionManager.create(config.cwd) + : SessionManager.inMemory(); + + const factory = new AgentSessionFactory(config); + const session = await factory.createSession(sessionManager); + + session.subscribe((event) => { + switch (event.type) { + case "message_update": + if (event.assistantMessageEvent.type === "text_delta") { + process.stdout.write(event.assistantMessageEvent.delta); + } + break; + + case "tool_execution_start": + if (config.verboseTools) { + process.stderr.write(`\n[tool:start] ${event.toolName}\n`); + } + break; + + case "tool_execution_end": + if (config.verboseTools) { + process.stderr.write(`[tool:end] ${event.toolName}\n`); + } + break; + + case "agent_end": + process.stdout.write("\n"); + break; + } + }); + + let userPrompt: string; + + if (config.promptText) { + userPrompt = config.promptText; + } else if (!process.stdin.isTTY) { + userPrompt = await readStdin(); + if (!userPrompt) { + throw new Error("Stdin was empty. Set PROMPT or pipe a non-empty message."); + } + } else { + throw new Error("No prompt provided. Set PROMPT or pipe input via stdin."); + } + + await session.prompt(userPrompt); +}