diff --git a/docs/config.md b/docs/config.md index 8a064b0..c9c3395 100644 --- a/docs/config.md +++ b/docs/config.md @@ -10,6 +10,7 @@ Boolean env values: `1`, `true`, `yes`, or `on` (case-insensitive). CSV lists ar | ---------- | --------- | ------- | | `PORTER_STATE_DIR` | `~/.local/state/porter` | Runtime state, Pi sessions, cron logs | | `PORTER_CONFIG_DIR` | `~/.config/porter` | Reserved for future config files | +| `PORTER_SOCKET` | `$XDG_RUNTIME_DIR/porter/porter.sock` | Control plane UNIX socket path (CLI only) | | `DATABASE_URL` | — | PostgreSQL connection URL (required) | ## Agent diff --git a/docs/control-plane.md b/docs/control-plane.md new file mode 100644 index 0000000..8fb70f8 --- /dev/null +++ b/docs/control-plane.md @@ -0,0 +1,251 @@ +# Control Plane + +Porter exposes a REST API over a UNIX domain socket for runtime management: +scheduled task CRUD, daemon health, and worker pool observability. The socket +uses systemd socket activation in production and direct binding in development. + +## Architecture + +``` +porter CLI ──fetch(unix)──> porter.sock ──> Bun.serve (ControlServer) + │ + ┌───────────┼───────────┐ + │ │ │ + SchedulerRegistry TaskStore WorkerPool +``` + +The control server runs inside the daemon process alongside channels, the +inbound/outbound workers, and the scheduler registry. No separate process, no +TCP port. + +## Socket lifecycle + +Systemd manages the socket directory via `RuntimeDirectory=porter-%i` in the +service unit. The daemon binds the socket itself -- systemd socket activation +(`Bun.serve({ fd })`) is not used because the feature never shipped in Bun +(see [oven-sh/bun#2852](https://github.com/oven-sh/bun/pull/2852)). + +```ini +# porter@.service +Environment=PORTER_SOCKET=%t/porter-%i/porter.sock +RuntimeDirectory=porter-%i +RuntimeDirectoryMode=0700 +``` + +Each template instance gets its own directory: + +``` +porter@projects-me-pi-porter → /run/user/1000/porter-projects-me-pi-porter/porter.sock +porter-dev → /run/user/1000/porter-dev/porter.sock +``` + +In development (`bun run`, no systemd): + +```bash +bun run runtime/src/index.ts --serve +# Falls back to $XDG_RUNTIME_DIR/porter/porter.sock +# or ~/.local/state/porter/porter.sock +``` + +The daemon prefers `PORTER_SOCKET` from the environment; the CLI uses the +same variable with the same fallback. + +## API Reference + +All endpoints return JSON. Error responses include an `error` field. + +### Health + +``` +GET /api/health +``` + +```json +{ + "status": "ok", + "uptime": 12345.678, + "pid": 12345, + "workers": 2 +} +``` + +### Scheduled Tasks + +#### List all tasks + +``` +GET /api/scheduled-tasks +``` + +Returns an array of task objects including `paused` and `completed` tasks. + +#### Get one task + +``` +GET /api/scheduled-tasks/:id +``` + +Returns a single task object, or `404`. + +#### Create a task + +``` +POST /api/scheduled-tasks +Content-Type: application/json + +{ + "id": "morning-brief", + "name": "Morning Briefing", + "prompt": "Summarize today's calendar and weather.", + "agentSessionKey": "main:telegram:default:dm:123456", + "scheduleType": "cron", + "scheduleValue": "0 9 * * *", + "reportSessionKey": "main:telegram:default:dm:123456", + "workdir": null, + "preHook": null, + "postHook": null +} +``` + +| Field | Required | Notes | +|-------|----------|-------| +| `id` | yes | Unique task identifier (slug) | +| `prompt` | yes | Prompt sent to the agent | +| `agentSessionKey` | yes | Session key for the agent run | +| `scheduleType` | yes | `cron`, `interval`, or `once` | +| `scheduleValue` | yes | Cron expression, milliseconds, or `"0"` for once | +| `name` | no | Human-readable label | +| `reportSessionKey` | no | Where to deliver results (defaults to agent session) | +| `workdir` | no | Working directory for the agent | +| `preHook` | no | Shell command run before the agent | +| `postHook` | no | Shell command run after the agent | + +Session keys for `agentSessionKey` and `reportSessionKey` are auto-created if +they don't already exist. + +Returns `201` with the created task object. Returns `409` if the id already +exists. + +#### Delete a task + +``` +DELETE /api/scheduled-tasks/:id +``` + +Pauses the task (disarms the timer) then removes it from the database. +Returns `204`. + +#### Pause a task + +``` +POST /api/scheduled-tasks/:id/pause +``` + +Disarms the timer. The task stays in the database with `status: "paused"`. +Returns `409` if already paused or completed. + +```json +{ "id": "morning-brief", "status": "paused" } +``` + +#### Resume a task + +``` +POST /api/scheduled-tasks/:id/resume +``` + +Re-arms the timer via the scheduler registry. Returns `409` if not paused. + +```json +{ "id": "morning-brief", "status": "active" } +``` + +#### Fire a task immediately + +``` +POST /api/scheduled-tasks/:id/fire +``` + +Publishes an inbound event for the task without waiting for the schedule. +Does not affect the existing timer. Returns `409` if the task is not active. + +```json +{ "id": "morning-brief", "fired": true } +``` + +#### Get run history + +``` +GET /api/scheduled-tasks/:id/runs?limit=50 +``` + +Returns recent run records (newest first), up to `limit` (default 50). + +### Workers + +``` +GET /api/workers +``` + +```json +{ + "count": 2, + "snapshot": [ + { "sessionKey": "main:telegram:default:dm:123456", "state": "ready" }, + { "sessionKey": "main:telegram:default:dm:789012", "state": "busy" } + ] +} +``` + +Worker states: `booting` (initializing Pi session), `ready` (idle), `busy` +(handling a prompt). + +## CLI + +```bash +# Socket path is set by the systemd service unit; export manually for dev: +export PORTER_SOCKET=$XDG_RUNTIME_DIR/porter-projects-me-pi-porter/porter.sock + +porter task list +porter task get morning-brief +porter task create --id nightly-report --cron "0 2 * * *" \ + --prompt "Generate daily summary" \ + --session-key "main:telegram:default:dm:123456" +porter task delete morning-brief +porter task pause morning-brief +porter task resume morning-brief +porter task fire morning-brief +porter task runs morning-brief +porter status +porter help +``` + +### Create flags + +| Flag | Notes | +|------|-------| +| `--id ` | Required | +| `--prompt ` | Required | +| `--session-key ` | Required | +| `--cron ` | e.g. `"0 9 * * *"` | +| `--interval ` | Milliseconds | +| `--once` | One-shot, fires immediately on create | +| `--name ` | Optional label | +| `--report-key ` | Where to deliver results | +| `--workdir ` | Agent working directory | +| `--pre-hook ` | Shell command before agent | +| `--post-hook ` | Shell command after agent | + +One of `--cron`, `--interval`, or `--once` is required. + +## curl + +```bash +curl --unix-socket "$PORTER_SOCKET" http://localhost/api/health +curl --unix-socket "$PORTER_SOCKET" http://localhost/api/scheduled-tasks + +curl --unix-socket "$PORTER_SOCKET" \ + -X POST http://localhost/api/scheduled-tasks \ + -H 'Content-Type: application/json' \ + -d '{"id":"test","prompt":"hello","agentSessionKey":"main:telegram:default:dm:123456","scheduleType":"once","scheduleValue":"0"}' +``` diff --git a/resources/systemd/porter-dev.service b/resources/systemd/porter-dev.service index 1878eb1..908408e 100644 --- a/resources/systemd/porter-dev.service +++ b/resources/systemd/porter-dev.service @@ -11,6 +11,9 @@ EnvironmentFile=%h/workspace/.env # Dev unit: run the locally built binary. PATH is for agent subprocesses (bun, tools), not ExecStart. ExecStart=%h/.local/bin/porter --serve Environment=PATH=%h/.local/bin:/usr/local/bin:/usr/bin:/bin +Environment=PORTER_SOCKET=%t/porter-dev/porter.sock +RuntimeDirectory=porter-dev +RuntimeDirectoryMode=0700 Restart=on-failure RestartSec=10s RestartSteps=4 diff --git a/resources/systemd/porter-dev.socket b/resources/systemd/porter-dev.socket new file mode 100644 index 0000000..f531098 --- /dev/null +++ b/resources/systemd/porter-dev.socket @@ -0,0 +1,11 @@ +[Unit] +Description=Porter control socket (dev) + +[Socket] +ListenStream=%t/porter/porter-dev.sock +SocketMode=0600 +DirectoryMode=0700 +RemoveOnStop=yes + +[Install] +WantedBy=sockets.target diff --git a/resources/systemd/porter@.service b/resources/systemd/porter@.service index 52d61a4..844b09f 100644 --- a/resources/systemd/porter@.service +++ b/resources/systemd/porter@.service @@ -11,6 +11,9 @@ EnvironmentFile=%h/%i/.env # Prod unit: packaged install at a fixed path. PATH only affects subprocesses, not ExecStart. ExecStart=/usr/bin/porter --serve Environment=PATH=%h/.local/bin:/usr/local/bin:/usr/bin:/bin +Environment=PORTER_SOCKET=%t/porter-%i/porter.sock +RuntimeDirectory=porter-%i +RuntimeDirectoryMode=0700 Restart=on-failure RestartSec=5 TimeoutStopSec=30 diff --git a/resources/systemd/porter@.socket b/resources/systemd/porter@.socket new file mode 100644 index 0000000..3c530cd --- /dev/null +++ b/resources/systemd/porter@.socket @@ -0,0 +1,21 @@ +# Template socket unit paired with porter@.service. +# +# Enable alongside the matching service instance: +# systemctl --user enable 'porter@projects-me-pi-porter.socket' +# systemctl --user enable 'porter@projects-me-pi-porter.service' +# +# Systemd creates the socket before the service starts and passes it as fd 3. +# The daemon detects LISTEN_FDS and uses Bun.serve({ fd: 3 }). +# RemoveOnStop=yes ensures the socket file is cleaned up on deactivation. + +[Unit] +Description=Porter control socket (%i) + +[Socket] +ListenStream=%t/porter/porter-%i.sock +SocketMode=0600 +DirectoryMode=0700 +RemoveOnStop=yes + +[Install] +WantedBy=sockets.target diff --git a/runtime/src/agent/pi-runner.ts b/runtime/src/agent/pi-runner.ts index a0b09b7..8e04d66 100644 --- a/runtime/src/agent/pi-runner.ts +++ b/runtime/src/agent/pi-runner.ts @@ -5,9 +5,11 @@ import { SessionWorkerPool } from './worker-pool.js'; export class PiAgentRunner implements AgentRunner { private cwd: string; private promptTimeoutMs: number; - private pool: SessionWorkerPool; private locks: Map> = new Map(); + /** Exposed for control plane observability (/api/workers). */ + readonly pool: SessionWorkerPool; + constructor(config: PorterConfig) { this.cwd = process.cwd(); this.promptTimeoutMs = config.agentPromptTimeoutMs; diff --git a/runtime/src/agent/worker-pool.ts b/runtime/src/agent/worker-pool.ts index e242b8e..b3e15d8 100644 --- a/runtime/src/agent/worker-pool.ts +++ b/runtime/src/agent/worker-pool.ts @@ -127,6 +127,15 @@ export class SessionWorkerPool { return this.workers.size; } + /** Read-only snapshot for observability (control plane /api/workers). */ + snapshot(): Array<{ sessionKey: string; state: string }> { + const result: Array<{ sessionKey: string; state: string }> = []; + for (const [key, entry] of this.workers) { + result.push({ sessionKey: key, state: entry.state }); + } + return result; + } + // ---- Internal ---- #spawn(key: string, cwd: string): WorkerEntry { diff --git a/runtime/src/cli.ts b/runtime/src/cli.ts new file mode 100644 index 0000000..4a5d012 --- /dev/null +++ b/runtime/src/cli.ts @@ -0,0 +1,273 @@ +/** + * Porter client CLI. Talks to the daemon over a UNIX socket. + * + * Socket location (first found wins): + * 1. PORTER_SOCKET env var + * 2. $XDG_RUNTIME_DIR/porter/porter.sock + * 3. $PORTER_STATE_DIR/porter.sock + * 4. ~/.local/state/porter/porter.sock + */ + +function socketPath(): string { + if (process.env.PORTER_SOCKET) return process.env.PORTER_SOCKET; + const runtimeDir = process.env.XDG_RUNTIME_DIR; + if (runtimeDir) return `${runtimeDir}/porter/porter.sock`; + const stateDir = process.env.PORTER_STATE_DIR || `${process.env.HOME}/.local/state/porter`; + return `${stateDir}/porter.sock`; +} + +async function apiFetch(path: string, init?: RequestInit): Promise { + const url = `http://localhost${path}`; + const res = await fetch(url, { ...init, unix: socketPath() }); + const body = await res.text(); + if (!res.ok) { + let message = `${res.status} ${res.statusText}`; + try { + const parsed = JSON.parse(body); + if (parsed.error) message = parsed.error; + } catch (err) { + // Response body is not JSON; use raw status text. + console.warn('[cli] failed to parse error response body', { status: res.status, err }); + } + throw new Error(message); + } + if (res.status === 204) return null; + return body ? JSON.parse(body) : null; +} + +function printJson(data: unknown): void { + console.log(JSON.stringify(data, null, 2)); +} + +// ---- Commands ---- + +async function cmdList(): Promise { + const tasks = await apiFetch('/api/scheduled-tasks'); + printJson(tasks); +} + +async function cmdGet(id: string): Promise { + const task = await apiFetch(`/api/scheduled-tasks/${encodeURIComponent(id)}`); + printJson(task); +} + +async function cmdCreate(opts: Record): Promise { + const id = opts.id; + if (!id) throw new Error('--id is required'); + + let scheduleType: string; + let scheduleValue: string; + if (opts.cron) { + scheduleType = 'cron'; + scheduleValue = opts.cron; + } else if (opts.interval) { + scheduleType = 'interval'; + scheduleValue = opts.interval; + } else if (opts.once !== undefined) { + scheduleType = 'once'; + scheduleValue = opts.once || '0'; + } else { + throw new Error('one of --cron, --interval, or --once is required'); + } + + const body: Record = { + id, + scheduleType, + scheduleValue, + prompt: opts.prompt ?? '', + agentSessionKey: opts['session-key'] ?? '', + }; + if (opts.name !== undefined) body.name = opts.name || null; + if (opts['report-key'] !== undefined) body.reportSessionKey = opts['report-key'] || null; + if (opts.workdir !== undefined) body.workdir = opts.workdir || null; + if (opts['pre-hook'] !== undefined) body.preHook = opts['pre-hook'] || null; + if (opts['post-hook'] !== undefined) body.postHook = opts['post-hook'] || null; + + const task = await apiFetch('/api/scheduled-tasks', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(body), + }); + printJson(task); +} + +async function cmdDelete(id: string): Promise { + await apiFetch(`/api/scheduled-tasks/${encodeURIComponent(id)}`, { method: 'DELETE' }); + console.log(`deleted task ${id}`); +} + +async function cmdPause(id: string): Promise { + const result = await apiFetch(`/api/scheduled-tasks/${encodeURIComponent(id)}/pause`, { method: 'POST' }); + printJson(result); +} + +async function cmdResume(id: string): Promise { + const result = await apiFetch(`/api/scheduled-tasks/${encodeURIComponent(id)}/resume`, { method: 'POST' }); + printJson(result); +} + +async function cmdFire(id: string): Promise { + const result = await apiFetch(`/api/scheduled-tasks/${encodeURIComponent(id)}/fire`, { method: 'POST' }); + printJson(result); +} + +async function cmdRuns(id: string): Promise { + const runs = await apiFetch(`/api/scheduled-tasks/${encodeURIComponent(id)}/runs`); + printJson(runs); +} + +async function cmdStatus(): Promise { + const health = await apiFetch('/api/health'); + printJson(health); +} + +// ---- Entry ---- + +export async function runCli(argv: string[]): Promise { + // argv = ['bun', 'index.ts', 'task', 'list', ...] — strip runtime prefix + const args = argv.slice(2); + + if (args.length === 0 || args[0] === 'help' || args[0] === '--help' || args[0] === '-h') { + printHelp(); + return; + } + + const subcommand = args[0]; + const rest = args.slice(1); + + try { + switch (subcommand) { + case 'task': + await handleTask(rest); + break; + case 'status': + await cmdStatus(); + break; + default: + console.error(`unknown command: ${subcommand}`); + console.error('run `porter help` for usage'); + process.exit(1); + } + } catch (error) { + if (error instanceof Error && error.message.includes('ConnectionRefused')) { + console.error('daemon not running or socket not found'); + console.error(` socket: ${socketPath()}`); + } else { + console.error(error instanceof Error ? error.message : String(error)); + } + process.exit(1); + } +} + +function handleTask(args: string[]): Promise { + if (args.length === 0) { + console.error('porter task: missing subcommand (list, get, create, delete, pause, resume, fire, runs)'); + process.exit(1); + } + + const sub = args[0]; + const subArgs = args.slice(1); + + switch (sub) { + case 'list': + return cmdList(); + case 'get': + return requireArg(subArgs, 'id').then((id) => cmdGet(id)); + case 'create': + return cmdCreate(parseCreateArgs(subArgs)); + case 'delete': + return requireArg(subArgs, 'id').then((id) => cmdDelete(id)); + case 'pause': + return requireArg(subArgs, 'id').then((id) => cmdPause(id)); + case 'resume': + return requireArg(subArgs, 'id').then((id) => cmdResume(id)); + case 'fire': + return requireArg(subArgs, 'id').then((id) => cmdFire(id)); + case 'runs': + return requireArg(subArgs, 'id').then((id) => cmdRuns(id)); + default: + console.error(`porter task: unknown subcommand: ${sub}`); + process.exit(1); + } +} + +function requireArg(args: string[], name: string): Promise { + if (args.length === 0) { + console.error(`missing required argument: ${name}`); + process.exit(1); + } + return Promise.resolve(args[0]!); +} + +const VALUE_FLAGS = new Set([ + '--id', + '--name', + '--cron', + '--interval', + '--prompt', + '--session-key', + '--report-key', + '--workdir', + '--pre-hook', + '--post-hook', +]); +const BOOL_FLAGS = new Set(['--once']); + +function parseCreateArgs(args: string[]): Record { + const opts: Record = {}; + let i = 0; + while (i < args.length) { + const arg = args[i]!; + if (VALUE_FLAGS.has(arg)) { + opts[arg.slice(2)] = args[++i] ?? ''; + } else if (BOOL_FLAGS.has(arg)) { + opts[arg.slice(2)] = ''; + i++; + } else { + i++; + } + } + return opts; +} + +function printHelp(): void { + console.log( + 'porter - Personal assistant daemon\n\n' + + 'Usage:\n' + + ' porter Client CLI (default)\n' + + ' porter --serve Start the daemon\n' + + ' porter --help Show help\n\n' + + 'Client commands:\n' + + ' porter task list List all scheduled tasks\n' + + ' porter task get Get task details\n' + + ' porter task create ... Create a new task\n' + + ' porter task delete Delete a task\n' + + ' porter task pause Pause a task\n' + + ' porter task resume Resume a paused task\n' + + ' porter task fire Trigger a task immediately\n' + + ' porter task runs Show recent run history\n' + + ' porter status Daemon health and stats\n\n' + + 'Task create options:\n' + + ' --id Required: unique task identifier\n' + + ' --prompt Required: prompt to send to the agent\n' + + ' --session-key Required: agent session key (e.g. main:telegram:default:dm:123456)\n' + + ' --cron Cron expression (e.g. "0 9 * * *")\n' + + ' --interval Interval in milliseconds\n' + + ' --once One-shot task (fires immediately on create)\n' + + ' --name Human-readable name\n' + + ' --report-key Session key for reporting results\n' + + ' --workdir Working directory for the agent\n' + + ' --pre-hook Shell command to run before the agent\n' + + ' --post-hook Shell command to run after the agent\n\n' + + 'Internal (spawned by daemon):\n' + + ' porter --agent-worker Run as a long-lived agent worker process\n\n' + + 'Environment:\n' + + ' DATABASE_URL PostgreSQL connection URL\n' + + ' PORTER_TELEGRAM_ENABLED=1 Enable Telegram long polling\n' + + ' PORTER_TELEGRAM_BOT_TOKEN= Telegram bot token\n' + + ' PORTER_TELEGRAM_ALLOWED_SENDERS= Comma-separated numeric sender IDs; * allows all\n' + + ' PORTER_AGENT_PROMPT_TIMEOUT_MS= Agent prompt timeout; default 900000\n' + + ' PORTER_AGENT_WORKER_MAX_COUNT= Max agent worker processes; default 10\n' + + ' PORTER_AGENT_WORKER_IDLE_TIMEOUT_MS= Idle worker eviction; default 600000\n', + ); +} diff --git a/runtime/src/control-server.ts b/runtime/src/control-server.ts new file mode 100644 index 0000000..8e9ef89 --- /dev/null +++ b/runtime/src/control-server.ts @@ -0,0 +1,329 @@ +/** + * UNIX socket control plane for the porter daemon. + * + * Exposes a REST API over HTTP on a systemd-passed file descriptor + * (socket activation) or a directly-bound UNIX socket in development. + * + * Routes: + * GET /api/health + * GET /api/scheduled-tasks + * GET /api/scheduled-tasks/:id + * POST /api/scheduled-tasks + * DELETE /api/scheduled-tasks/:id + * POST /api/scheduled-tasks/:id/pause + * POST /api/scheduled-tasks/:id/resume + * POST /api/scheduled-tasks/:id/fire + * GET /api/scheduled-tasks/:id/runs + * GET /api/workers + */ + +import type { SessionWorkerPool } from './agent/worker-pool.js'; +import type { ScheduledTaskStore } from './db/scheduled-task-store.js'; +import type { SessionStore } from './db/session-store.js'; +import { parseSessionKey } from './routing/session-key.js'; +import type { SchedulerRegistry } from './scheduler/registry.js'; +import type { NewScheduledTask } from './scheduler/types.js'; + +// ---- Types ---- + +export type ControlServerOptions = { + /** File descriptor from systemd socket activation (fd 3). */ + fd?: number; + /** UNIX socket path for direct binding (development fallback). */ + unix?: string; + scheduler: SchedulerRegistry; + taskStore: ScheduledTaskStore; + sessionStore: SessionStore; + /** Optional: worker pool for /api/workers observability. */ + workerPool?: SessionWorkerPool; +}; + +// ---- Helpers ---- + +function json(data: unknown, status = 200): Response { + return new Response(JSON.stringify(data), { + status, + headers: { 'content-type': 'application/json' }, + }); +} + +function errorJson(message: string, status: number): Response { + return json({ error: message }, status); +} + +// ---- ControlServer ---- + +export class ControlServer { + private server: ReturnType | null = null; + private socketPath: string | null = null; // for dev-mode cleanup + private scheduler: SchedulerRegistry; + private taskStore: ScheduledTaskStore; + private sessionStore: SessionStore; + private workerPool: SessionWorkerPool | undefined; + + constructor(options: ControlServerOptions) { + this.scheduler = options.scheduler; + this.taskStore = options.taskStore; + this.sessionStore = options.sessionStore; + this.workerPool = options.workerPool; + } + + async start(): Promise { + const fetchHandler: (req: Request) => Response | Promise = (req) => { + const url = new URL(req.url); + const id = url.pathname.match(/^\/api\/scheduled-tasks\/([^/]+)/)?.[1]; + + // Health + if (url.pathname === '/api/health' && req.method === 'GET') return this.handleHealth(); + + // Scheduled tasks — collection + if (url.pathname === '/api/scheduled-tasks') { + if (req.method === 'GET') return this.handleListTasks(); + if (req.method === 'POST') return this.handleCreateTask(req); + } + + // Scheduled tasks — single + if (id && url.pathname === `/api/scheduled-tasks/${id}`) { + if (req.method === 'GET') return this.handleGetTask(id); + if (req.method === 'DELETE') return this.handleDeleteTask(id); + } + if (id && url.pathname === `/api/scheduled-tasks/${id}/pause` && req.method === 'POST') + return this.handlePauseTask(id); + if (id && url.pathname === `/api/scheduled-tasks/${id}/resume` && req.method === 'POST') + return this.handleResumeTask(id); + if (id && url.pathname === `/api/scheduled-tasks/${id}/fire` && req.method === 'POST') + return this.handleFireTask(id); + if (id && url.pathname === `/api/scheduled-tasks/${id}/runs` && req.method === 'GET') { + const limit = Number.parseInt(url.searchParams.get('limit') ?? '50', 10); + return this.handleGetTaskRuns(id, limit); + } + + // Workers + if (url.pathname === '/api/workers' && req.method === 'GET') return this.handleWorkers(); + + return errorJson('not found', 404); + }; + + const onError = (err: Error) => { + console.error('[control-server] unhandled error', { error: err }); + return errorJson('internal server error', 500); + }; + + const socketPath = + process.env.PORTER_SOCKET || + (() => { + const runtimeDir = process.env.XDG_RUNTIME_DIR; + const stateDir = process.env.PORTER_STATE_DIR || `${process.env.HOME}/.local/state/porter`; + const base = runtimeDir ? `${runtimeDir}/porter` : stateDir; + return `${base}/porter.sock`; + })(); + + // Clean up stale socket from a previous run. + try { + await Bun.file(socketPath).delete(); + } catch (err) { + // ENOENT is expected; log anything unexpected. + console.warn('[control-server] stale socket cleanup failed', { path: socketPath, err }); + } + + this.socketPath = socketPath; + console.log('[control-server] binding on unix socket', { path: socketPath }); + this.server = Bun.serve({ unix: socketPath, fetch: fetchHandler, error: onError }); + + console.log('[control-server] started'); + } + + async stop(): Promise { + if (this.server) { + await this.server.stop(); + this.server = null; + } + // In dev mode we own the socket; clean it up. Under systemd socket + // activation RemoveOnStop=yes handles unlink. + if (this.socketPath) { + const path = this.socketPath; + await Bun.file(path) + .delete() + .catch((err) => { + console.warn('[control-server] socket unlink on stop failed', { path, err }); + }); + this.socketPath = null; + } + console.log('[control-server] stopped'); + } + + // ---- Route handlers ---- + + private handleHealth(): Response { + const workerCount = this.workerPool?.size ?? 'n/a'; + return json({ + status: 'ok', + uptime: process.uptime(), + pid: process.pid, + workers: workerCount, + }); + } + + private async handleListTasks(): Promise { + const tasks = await this.taskStore.listAll(); + return json(tasks); + } + + private async handleGetTask(id: string): Promise { + const task = await this.taskStore.getById(id); + if (!task) return errorJson('task not found', 404); + return json(task); + } + + private async handleCreateTask(req: Request): Promise { + let body: unknown; + try { + body = await req.json(); + } catch { + return errorJson('invalid JSON body', 400); + } + + const input = body as Record; + const validation = validateCreateTask(input); + if (validation) return errorJson(validation, 400); + + const newTask: NewScheduledTask = { + id: String(input.id), + name: typeof input.name === 'string' ? input.name : null, + prompt: String(input.prompt), + agentSessionKey: String(input.agentSessionKey), + reportSessionKey: typeof input.reportSessionKey === 'string' ? input.reportSessionKey : null, + workdir: typeof input.workdir === 'string' ? input.workdir : null, + preHook: typeof input.preHook === 'string' ? input.preHook : null, + postHook: typeof input.postHook === 'string' ? input.postHook : null, + scheduleType: input.scheduleType as NewScheduledTask['scheduleType'], + scheduleValue: String(input.scheduleValue), + }; + + // Ensure the session row exists so the FK constraint passes. + const parsed = parseSessionKey(newTask.agentSessionKey); + if (!parsed) return errorJson('invalid agentSessionKey format', 400); + await this.sessionStore.ensureSession(newTask.agentSessionKey, parsed); + if (newTask.reportSessionKey) { + const reportParsed = parseSessionKey(newTask.reportSessionKey); + if (!reportParsed) return errorJson('invalid reportSessionKey format', 400); + await this.sessionStore.ensureSession(newTask.reportSessionKey, reportParsed); + } + + try { + const task = await this.taskStore.create(newTask); + await this.scheduler.refresh(task.id); + console.log('[control-server] created task', { taskId: task.id, name: task.name }); + return json(task, 201); + } catch (err: unknown) { + const msg = err instanceof Error ? err.message : String(err); + if (msg.includes('duplicate key') || msg.includes('violates unique constraint')) { + return errorJson(`task id '${newTask.id}' already exists`, 409); + } + console.error('[control-server] create task failed', { taskId: newTask.id, error: msg }); + return errorJson('failed to create task', 500); + } + } + + private async handleDeleteTask(id: string): Promise { + const task = await this.taskStore.getById(id); + if (!task) return errorJson('task not found', 404); + + // Pause (disarms the timer) before deleting from DB. + await this.taskStore.setStatus(id, 'paused'); + await this.scheduler.refresh(id); + + const deleted = await this.taskStore.delete(id); + if (!deleted) return errorJson('task not found', 404); // race + console.log('[control-server] deleted task', { taskId: id }); + return new Response(null, { status: 204 }); + } + + private async handlePauseTask(id: string): Promise { + const task = await this.taskStore.getById(id); + if (!task) return errorJson('task not found', 404); + if (task.status !== 'active') { + return errorJson(`task is already ${task.status}`, 409); + } + + await this.taskStore.setStatus(id, 'paused'); + // refresh disarms the paused task's handle since status != 'active' + await this.scheduler.refresh(id); + console.log('[control-server] paused task', { taskId: id }); + return json({ id, status: 'paused' }); + } + + private async handleResumeTask(id: string): Promise { + const task = await this.taskStore.getById(id); + if (!task) return errorJson('task not found', 404); + if (task.status !== 'paused') { + return errorJson(`task is ${task.status}, not paused`, 409); + } + + await this.taskStore.setStatus(id, 'active'); + await this.scheduler.refresh(id); + console.log('[control-server] resumed task', { taskId: id }); + return json({ id, status: 'active' }); + } + + private async handleFireTask(id: string): Promise { + const result = await this.scheduler.fireNow(id); + if (!result.ok) { + return errorJson(result.error ?? 'failed to fire task', result.error === 'task not found' ? 404 : 409); + } + return json({ id, fired: true }); + } + + private async handleGetTaskRuns(id: string, limit: number): Promise { + const clamped = Number.isFinite(limit) && limit > 0 ? Math.min(limit, 500) : 50; + const runs = await this.taskStore.getRuns(id, clamped); + return json(runs); + } + + private handleWorkers(): Response { + if (!this.workerPool) { + return json({ workers: 'unavailable' }); + } + return json({ + count: this.workerPool.size, + snapshot: this.workerPool.snapshot(), + }); + } +} + +// ---- Validation ---- + +const VALID_SCHEDULE_TYPES = new Set(['cron', 'interval', 'once']); + +function validateCreateTask(input: Record): string | null { + if (typeof input.id !== 'string' || !input.id.trim()) { + return 'id is required and must be a non-empty string'; + } + if (typeof input.prompt !== 'string' || !input.prompt.trim()) { + return 'prompt is required and must be a non-empty string'; + } + if (typeof input.agentSessionKey !== 'string' || !input.agentSessionKey.trim()) { + return 'agentSessionKey is required and must be a non-empty string'; + } + if (typeof input.scheduleType !== 'string' || !VALID_SCHEDULE_TYPES.has(input.scheduleType)) { + return 'scheduleType must be one of: cron, interval, once'; + } + if (typeof input.scheduleValue !== 'string' || !input.scheduleValue.trim()) { + return 'scheduleValue is required and must be a non-empty string'; + } + if (input.reportSessionKey !== undefined && input.reportSessionKey !== null) { + if (typeof input.reportSessionKey !== 'string' || !input.reportSessionKey.trim()) { + return 'reportSessionKey must be a non-empty string or null'; + } + } + if (input.preHook !== undefined && input.preHook !== null && typeof input.preHook !== 'string') { + return 'preHook must be a string or null'; + } + if (input.postHook !== undefined && input.postHook !== null && typeof input.postHook !== 'string') { + return 'postHook must be a string or null'; + } + if (input.workdir !== undefined && input.workdir !== null && typeof input.workdir !== 'string') { + return 'workdir must be a string or null'; + } + return null; +} diff --git a/runtime/src/daemon.ts b/runtime/src/daemon.ts index 5690fd7..fffb2fe 100644 --- a/runtime/src/daemon.ts +++ b/runtime/src/daemon.ts @@ -4,6 +4,7 @@ import { ChannelManager } from './channels/manager.js'; import { MatrixRuntime } from './channels/matrix/index.js'; import { TelegramRuntime } from './channels/telegram/index.js'; import { ensureRuntimeDirs, type PorterConfig } from './config.js'; +import { ControlServer } from './control-server.js'; import { ChannelWorkdirStore } from './db/channel-workdir-store.js'; import { closeDb, type Db, getDb } from './db/client.js'; import { migrate } from './db/migrate.js'; @@ -18,6 +19,7 @@ import { OutboundWorker } from './workers/outbound-worker.js'; export class PorterDaemon { private channels: ChannelManager | null = null; private config: PorterConfig; + private controlServer: ControlServer | null = null; private db: Db | null = null; private inboundWorker: InboundWorker | null = null; private outboundWorker: OutboundWorker | null = null; @@ -92,7 +94,16 @@ export class PorterDaemon { }); this.scheduler = scheduler; - this.inboundWorker = new InboundWorker(bus, sessions, transcripts, new PiAgentRunner(this.config), { + const agentRunner = new PiAgentRunner(this.config); + + this.controlServer = new ControlServer({ + scheduler, + taskStore: scheduledTasks, + sessionStore: sessions, + workerPool: agentRunner.pool, + }); + + this.inboundWorker = new InboundWorker(bus, sessions, transcripts, agentRunner, { stateDir: this.config.stateDir, sessionRoot, sessionArchiveStore: sessionArchives, @@ -107,6 +118,7 @@ export class PorterDaemon { this.outboundWorker.start(); this.inboundWorker.start(); await scheduler.start(); + await this.controlServer.start(); } catch (error) { await this.stop(); throw error; @@ -116,6 +128,9 @@ export class PorterDaemon { } async stop(): Promise { + await this.controlServer?.stop(); + this.controlServer = null; + this.scheduler?.stop(); this.scheduler = null; diff --git a/runtime/src/db/scheduled-task-store.ts b/runtime/src/db/scheduled-task-store.ts index 4d40cbb..092b84f 100644 --- a/runtime/src/db/scheduled-task-store.ts +++ b/runtime/src/db/scheduled-task-store.ts @@ -1,6 +1,9 @@ +import { computeNextRun } from '../scheduler/compute-next-run.js'; import type { + NewScheduledTask, NewScheduledTaskRun, ScheduledTask, + ScheduledTaskRun, TaskStatus, UpdateScheduledTaskAfterRun, } from '../scheduler/types.js'; @@ -88,4 +91,70 @@ export class ScheduledTaskStore { ) `; } + + async listAll(): Promise { + const rows = (await this.db` + select * from scheduled_tasks + order by created_at, id + `) as Record[]; + return rows.map(mapTask); + } + + async create(input: NewScheduledTask): Promise { + const nextRun = computeNextRun(input.scheduleType, input.scheduleValue); + const rows = (await this.db` + insert into scheduled_tasks ( + id, name, prompt, + agent_session_key, report_session_key, + workdir, pre_hook, post_hook, + schedule_type, schedule_value, + next_run, status + ) values ( + ${input.id}, ${input.name ?? null}, ${input.prompt}, + ${input.agentSessionKey}, ${input.reportSessionKey ?? null}, + ${input.workdir ?? null}, ${input.preHook ?? null}, ${input.postHook ?? null}, + ${input.scheduleType}, ${input.scheduleValue}, + ${nextRun}, 'active' + ) + returning * + `) as Record[]; + return mapTask(rows[0]!); + } + + async delete(id: string): Promise { + const result = await this.db` + delete from scheduled_tasks where id = ${id} + `; + return result.count > 0; + } + + async setStatus(id: string, status: TaskStatus): Promise { + const result = await this.db` + update scheduled_tasks + set status = ${status}::scheduled_task_status_t, updated_at = now() + where id = ${id} + `; + return result.count > 0; + } + + async getRuns(taskId: string, limit = 50): Promise { + const rows = (await this.db` + select * from scheduled_task_runs + where task_id = ${taskId} + order by run_at desc, id desc + limit ${limit} + `) as Record[]; + return rows.map( + (row): ScheduledTaskRun => ({ + id: Number(row.id), + taskId: String(row.task_id), + inboundId: row.inbound_id != null ? Number(row.inbound_id) : null, + runAt: row.run_at as Date, + durationMs: row.duration_ms != null ? Number(row.duration_ms) : null, + status: row.status as ScheduledTaskRun['status'], + result: (row.result as string | null) ?? null, + error: (row.error as string | null) ?? null, + }), + ); + } } diff --git a/runtime/src/index.ts b/runtime/src/index.ts index f42396c..0b7eb38 100644 --- a/runtime/src/index.ts +++ b/runtime/src/index.ts @@ -1,6 +1,7 @@ #!/usr/bin/env bun import { parseArgs } from 'node:util'; +import { runCli } from './cli.js'; import { loadConfig } from './config.js'; import { PorterDaemon } from './daemon.js'; @@ -78,6 +79,5 @@ if (values['agent-worker']) { } } else { // Default: client CLI mode. Talks to the daemon over a UNIX socket. - console.log('porter client (not yet implemented)'); - process.exit(0); + await runCli(Bun.argv); } diff --git a/runtime/src/scheduler/index.ts b/runtime/src/scheduler/index.ts index c51ed13..9496160 100644 --- a/runtime/src/scheduler/index.ts +++ b/runtime/src/scheduler/index.ts @@ -5,4 +5,11 @@ export type { HookResult } from './hooks.js'; export { runHook } from './hooks.js'; export { SchedulerRegistry } from './registry.js'; export { buildSchedulerAgentSessionKey } from './session-keys.js'; -export type { ScheduledTask, ScheduledTaskRunStatus, ScheduleType, TaskStatus } from './types.js'; +export type { + NewScheduledTask, + ScheduledTask, + ScheduledTaskRun, + ScheduledTaskRunStatus, + ScheduleType, + TaskStatus, +} from './types.js'; diff --git a/runtime/src/scheduler/registry.ts b/runtime/src/scheduler/registry.ts index 5e33dc1..4d869db 100644 --- a/runtime/src/scheduler/registry.ts +++ b/runtime/src/scheduler/registry.ts @@ -45,6 +45,43 @@ export class SchedulerRegistry { console.log('[scheduler] registry stopped'); } + /** + * Immediately fire a scheduled task by publishing an inbound event. + * Does not affect the existing schedule timer. + */ + async fireNow(taskId: string): Promise<{ ok: boolean; error?: string }> { + if (!this.started) return { ok: false, error: 'scheduler not started' }; + + const task = await this.store.getById(taskId); + if (!task) return { ok: false, error: 'task not found' }; + if (task.status !== 'active') return { ok: false, error: `task is ${task.status}, not active` }; + + const parsed = parseSessionKey(task.agentSessionKey); + if (!parsed) return { ok: false, error: `invalid agent session key: ${task.agentSessionKey}` }; + + await this.sessions.ensureSession(task.agentSessionKey, parsed); + + await this.bus.publishInbound({ + sessionKey: task.agentSessionKey, + channel: 'scheduler', + accountId: 'default', + chatId: task.id, + senderId: 'scheduler', + content: task.prompt, + metadata: { + scheduled: true, + taskId: task.id, + taskName: task.name, + reportSessionKey: task.reportSessionKey, + firedManually: true, + ...(task.workdir ? { workdir: task.workdir } : {}), + }, + }); + + console.log('[scheduler] manual fire', { taskId: task.id, name: task.name }); + return { ok: true }; + } + async refresh(taskId: string): Promise { if (!this.started) return; this.disarm(taskId); diff --git a/runtime/src/scheduler/types.ts b/runtime/src/scheduler/types.ts index dde435d..41f1e92 100644 --- a/runtime/src/scheduler/types.ts +++ b/runtime/src/scheduler/types.ts @@ -37,3 +37,27 @@ export type UpdateScheduledTaskAfterRun = { lastResult: string; status?: TaskStatus; }; + +export type NewScheduledTask = { + id: string; + name: string | null; + prompt: string; + agentSessionKey: string; + reportSessionKey: string | null; + workdir: string | null; + preHook: string | null; + postHook: string | null; + scheduleType: ScheduleType; + scheduleValue: string; +}; + +export type ScheduledTaskRun = { + id: number; + taskId: string; + inboundId: number | null; + runAt: Date; + durationMs: number | null; + status: ScheduledTaskRunStatus; + result: string | null; + error: string | null; +};