From b906cb23774d9631ec86eecee07bab88ecce884a Mon Sep 17 00:00:00 2001 From: Rob Colbert Date: Sun, 17 May 2026 00:48:39 -0400 Subject: [PATCH] =?UTF-8?q?feat(gadget-tasks):=20course=20correction=20?= =?UTF-8?q?=E2=80=94=20headless=20IDE=20client=20architecture?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Converts gadget-tasks from a duplicated codebase with direct MongoDB access to a headless IDE client that uses gadget-code's REST API and Socket.IO protocol to submit and process task prompts through the existing pipeline. Deleted (no longer needed): - gadget-tasks/src/models/ (5 Mongoose models — duplicated from gadget-code) - gadget-tasks/data/prompts/ (11 prompt templates — duplicated from gadget-code) - gadget-tasks/src/services/executor.ts (629-line AWL loop — duplicated from gadget-drone) - gadget-tasks/src/services/ai.ts (direct AI API calls — drones do this now) - gadget-tasks/src/services/workspace.ts (workspace management — drones do this now) Added: - gadget-tasks/src/services/platform.ts — REST API + Socket.IO headless IDE client with session lock, workspace mode, prompt submission, work order tracking Rewritten: - gadget-tasks/src/gadget-tasks.ts — new startup: user login → auth → drone selection → Socket.IO connect → schedule tasks (no MongoDB) - gadget-tasks/src/services/scheduler.ts — delegates to PlatformService.executeTask() - gadget-tasks/src/config/env.ts — platform.baseUrl config, no mongodb/google gadget-code changes: - Added PATCH /api/v1/projects/:projectId/tasks/:taskId/lastRun route - Added ProjectService.updateTaskLastRun() for atomic task field update Config changes: - GadgetTasksConfig: replaced mongodb with platform.baseUrl, removed google.cse - Dependencies: removed mongoose, @gadget/ai, @gadget/ai-toolbox, dayjs, simple-git, nanoid; added socket.io-client, @inquirer/prompts --- gadget-code/src/controllers/api/v1/project.ts | 44 ++ gadget-code/src/services/project.ts | 13 + gadget-tasks/.gitignore | 1 + gadget-tasks/gadget-tasks.yaml | 13 + gadget-tasks/package.json | 39 ++ gadget-tasks/src/config/env.ts | 88 +++ gadget-tasks/src/gadget-tasks.ts | 191 ++++++ gadget-tasks/src/lib/process.ts | 19 + gadget-tasks/src/lib/service.ts | 19 + gadget-tasks/src/services/lock.ts | 138 +++++ gadget-tasks/src/services/platform.ts | 572 ++++++++++++++++++ gadget-tasks/src/services/scheduler.ts | 133 ++++ gadget-tasks/tsconfig.json | 23 + packages/config/src/loader.ts | 9 +- packages/config/src/types.ts | 34 ++ pnpm-lock.yaml | 31 + 16 files changed, 1366 insertions(+), 1 deletion(-) create mode 100644 gadget-tasks/.gitignore create mode 100644 gadget-tasks/gadget-tasks.yaml create mode 100644 gadget-tasks/package.json create mode 100644 gadget-tasks/src/config/env.ts create mode 100644 gadget-tasks/src/gadget-tasks.ts create mode 100644 gadget-tasks/src/lib/process.ts create mode 100644 gadget-tasks/src/lib/service.ts create mode 100644 gadget-tasks/src/services/lock.ts create mode 100644 gadget-tasks/src/services/platform.ts create mode 100644 gadget-tasks/src/services/scheduler.ts create mode 100644 gadget-tasks/tsconfig.json diff --git a/gadget-code/src/controllers/api/v1/project.ts b/gadget-code/src/controllers/api/v1/project.ts index bce58b4..df35d76 100644 --- a/gadget-code/src/controllers/api/v1/project.ts +++ b/gadget-code/src/controllers/api/v1/project.ts @@ -32,6 +32,7 @@ export class ProjectApiControllerV1 extends DtpController { this.router.get("/:projectId", this.getProject.bind(this)); this.router.put("/:projectId", this.updateProject.bind(this)); this.router.delete("/:projectId", this.deleteProject.bind(this)); + this.router.patch("/:projectId/tasks/:taskId/lastRun", this.updateTaskLastRun.bind(this)); } async getProjects(req: Request, res: Response): Promise { @@ -256,6 +257,49 @@ export class ProjectApiControllerV1 extends DtpController { }); } } + + async updateTaskLastRun(req: Request, res: Response): Promise { + try { + const projectId = req.params.projectId as string; + const taskId = req.params.taskId as string; + const { sessionId } = req.body; + + if (!sessionId) { + res.status(400).json({ + success: false, + message: "sessionId is required", + }); + return; + } + + const project = await projectService.findById(projectId); + if (!project) { + res.status(404).json({ + success: false, + message: "project not found", + }); + return; + } + + const user = project.user as IUser; + if (user._id !== req.user._id) { + res.status(403).json({ + success: false, + message: "access denied", + }); + return; + } + + await projectService.updateTaskLastRun(projectId, taskId, sessionId); + res.status(200).json({ success: true }); + } catch (error) { + this.log.error("failed to update task lastRun", { error }); + res.status(500).json({ + success: false, + message: "failed to update task lastRun", + }); + } + } } export default ProjectApiControllerV1; diff --git a/gadget-code/src/services/project.ts b/gadget-code/src/services/project.ts index e01a479..1310d8e 100644 --- a/gadget-code/src/services/project.ts +++ b/gadget-code/src/services/project.ts @@ -257,6 +257,19 @@ class ProjectService extends DtpService { ); } + async updateTaskLastRun(projectId: string, taskId: string, sessionId: string): Promise { + const result = await Project.updateOne( + { _id: projectId, "tasks._id": taskId }, + { $set: { "tasks.$.lastRun": sessionId } }, + ); + + if (result.matchedCount === 0) { + throw new Error(`Task ${taskId} not found in project ${projectId}`); + } + + this.log.info("task lastRun updated", { projectId, taskId, sessionId }); + } + async delete(project: IProject): Promise { await Project.deleteOne({ _id: project._id }); } diff --git a/gadget-tasks/.gitignore b/gadget-tasks/.gitignore new file mode 100644 index 0000000..1521c8b --- /dev/null +++ b/gadget-tasks/.gitignore @@ -0,0 +1 @@ +dist diff --git a/gadget-tasks/gadget-tasks.yaml b/gadget-tasks/gadget-tasks.yaml new file mode 100644 index 0000000..25d2ae7 --- /dev/null +++ b/gadget-tasks/gadget-tasks.yaml @@ -0,0 +1,13 @@ +timezone: America/New_York +platform: + baseUrl: https://code-dev.g4dge7.com:5174 +redis: + host: localhost + port: 6379 +concurrency: 1 +logging: + console: + enabled: true + file: + enabled: true + path: ~/logs/gadget-tasks diff --git a/gadget-tasks/package.json b/gadget-tasks/package.json new file mode 100644 index 0000000..0f103ed --- /dev/null +++ b/gadget-tasks/package.json @@ -0,0 +1,39 @@ +{ + "name": "gadget-tasks", + "version": "1.0.0", + "description": "Gadget Code scheduled task worker process", + "type": "module", + "bin": { + "gadget-tasks": "./dist/gadget-tasks.js" + }, + "main": "./dist/gadget-tasks.js", + "scripts": { + "dev": "tsx src/gadget-tasks.ts", + "dev:watch": "tsx watch src/gadget-tasks.ts", + "build": "tsc", + "start": "node dist/gadget-tasks.js" + }, + "keywords": [ + "gadget", + "tasks", + "cron", + "scheduler", + "worker" + ], + "author": "Rob Colbert", + "license": "Apache-2.0", + "packageManager": "pnpm@10.33.2+sha512.a90faf6feeab71ad6c6e57f94e0fe1a12f5dcc22cd754db40ae9593eb6a3e0b6b12e3540218bb37ae083404b1f2ce6db2a4121e979829b4aff94b99f49da1cf8", + "dependencies": { + "@gadget/api": "workspace:*", + "@gadget/config": "workspace:*", + "@inquirer/prompts": "^8.4.2", + "cron": "^4.3.1", + "ioredis": "^5.6.0", + "socket.io-client": "^4.8.1" + }, + "devDependencies": { + "@types/node": "^25.6.0", + "tsx": "^4.21.0", + "typescript": "6.0.3" + } +} diff --git a/gadget-tasks/src/config/env.ts b/gadget-tasks/src/config/env.ts new file mode 100644 index 0000000..efc279d --- /dev/null +++ b/gadget-tasks/src/config/env.ts @@ -0,0 +1,88 @@ +// config/env.ts +// Copyright (C) 2026 Rob Colbert +// Licensed under the Apache License, Version 2.0 + +import path, { dirname } from "node:path"; +import { fileURLToPath } from "node:url"; +import { loadGadgetTasksConfig, resolvePath } from "@gadget/config"; +import type PackageJson from "../../package.json"; +import { GadgetLog, GadgetLogFile } from "@gadget/api"; + +const __dirname = dirname(fileURLToPath(import.meta.url)); + +// INSTALL_DIR: where the package is installed (for loading assets, etc.) +export const INSTALL_DIR = path.resolve(__dirname, "..", ".."); + +// Load YAML configuration +const yamlConfig = loadGadgetTasksConfig(); + +// Validate required fields +if (!yamlConfig.platform?.baseUrl) { + throw new Error( + "Configuration error: platform.baseUrl is required in gadget-tasks.yaml\n" + + "See documentation: ./docs/configuration.md", + ); +} + +async function readJsonFile(filePath: string): Promise { + const fs = await import("node:fs"); + const file = await fs.promises.readFile(filePath); + return JSON.parse(file.toString("utf-8")) as T; +} + +/* eslint-disable no-process-env */ +export default { + NODE_ENV: process.env.NODE_ENV || "develop", + timezone: yamlConfig.timezone || "America/New_York", + installDir: INSTALL_DIR, + pkg: await readJsonFile( + path.join(INSTALL_DIR, "package.json"), + ), + platform: { + baseUrl: yamlConfig.platform.baseUrl, + }, + redis: { + host: yamlConfig.redis?.host || "localhost", + port: yamlConfig.redis?.port || 6379, + password: yamlConfig.redis?.password, + keyPrefix: yamlConfig.redis?.keyPrefix || "gadget:", + }, + concurrency: yamlConfig.concurrency || 1, + log: { + console: { + enabled: yamlConfig.logging?.console?.enabled === true, + }, + file: { + enabled: yamlConfig.logging?.file?.enabled === true, + path: yamlConfig.logging?.file?.path + ? resolvePath(yamlConfig.logging.file.path) + : path.join(INSTALL_DIR, "logs"), + name: yamlConfig.logging?.file?.name || "gadget-tasks", + maxWritesPerFile: yamlConfig.logging?.file?.maxWritesPerFile || 10000, + maxFiles: yamlConfig.logging?.file?.maxFiles || 10, + }, + levels: { + debug: yamlConfig.logging?.levels?.debug === true, + info: yamlConfig.logging?.levels?.info === true, + warn: yamlConfig.logging?.levels?.warn === true, + }, + }, +}; + +// Configure GadgetLog for this package +GadgetLog.consoleEnabled = yamlConfig.logging?.console?.enabled === true; +if (yamlConfig.logging?.file?.enabled === true) { + const logFileOptions = { + basePath: yamlConfig.logging.file.path + ? resolvePath(yamlConfig.logging.file.path) + : path.join(INSTALL_DIR, "logs"), + name: yamlConfig.logging.file.name || "gadget-tasks", + maxWritesPerFile: yamlConfig.logging.file.maxWritesPerFile || 10000, + maxFiles: yamlConfig.logging.file.maxFiles || 10, + }; + const defaultLogFile = new GadgetLogFile(logFileOptions); + defaultLogFile.open(); + GadgetLog.defaultFile = defaultLogFile; +} + +/* eslint-enable no-process-env */ diff --git a/gadget-tasks/src/gadget-tasks.ts b/gadget-tasks/src/gadget-tasks.ts new file mode 100644 index 0000000..4688dea --- /dev/null +++ b/gadget-tasks/src/gadget-tasks.ts @@ -0,0 +1,191 @@ +// src/gadget-tasks.ts +// Copyright (C) 2026 Rob Colbert +// Licensed under the Apache License, Version 2.0 + +import env from "./config/env.ts"; + +import PlatformService from "./services/platform.ts"; +import SchedulerService from "./services/scheduler.ts"; +import TaskLockService from "./services/lock.ts"; + +import { GadgetLog, type IDroneRegistration, type IProject } from "@gadget/api"; +import { GadgetProcess } from "./lib/process.ts"; + +const log = new GadgetLog({ name: "GadgetTasks", slug: "gadget-tasks" }); + +interface UserCredentials { + email: string; + password: string; +} + +class GadgetTasks extends GadgetProcess { + private selectedDrone: IDroneRegistration | null = null; + + get name(): string { + return "GadgetTasks"; + } + + get slug(): string { + return "gadget-tasks"; + } + + constructor() { + super(); + } + + async start(): Promise { + this.hookProcessSignals(); + + // 1. Acquire singleton lock via Redis + const lockAcquired = await TaskLockService.acquire(); + if (!lockAcquired) { + this.log.error("another instance of gadget-tasks is already running"); + process.exit(1); + } + + // 2. Get user credentials (CLI args or interactive prompt) + const credentials = await this.getUserCredentials(); + + // 3. Authenticate with platform + await PlatformService.start(); + await PlatformService.authenticate(credentials.email, credentials.password); + this.log.info("authenticated with platform", { email: credentials.email }); + + // 4. Select a drone + this.selectedDrone = await this.selectDrone(); + PlatformService.setSelectedDrone(this.selectedDrone); + this.log.info("drone selected", { + droneId: this.selectedDrone._id, + hostname: this.selectedDrone.hostname, + }); + + // 5. Connect Socket.IO + await PlatformService.connectSocket(); + + // 6. Start scheduler + await SchedulerService.start(); + + // 7. Fetch and schedule active tasks + await this.scheduleAllTasks(); + + this.log.info(`gadget-tasks v${env.pkg.version} started, ${SchedulerService.scheduledCount} tasks scheduled`); + } + + async stop(): Promise { + this.log.info(`gadget-tasks v${env.pkg.version} shutting down`); + + await SchedulerService.stop(); + await PlatformService.stop(); + await TaskLockService.release(); + + this.log.info("shutdown complete"); + return 0; + } + + /** + * Query all active projects and schedule their enabled tasks. + */ + async scheduleAllTasks(): Promise { + const projects = await PlatformService.getProjects(); + const activeProjects = projects.filter((p) => p.status === "active"); + + let taskCount = 0; + for (const project of activeProjects) { + for (const task of project.tasks) { + if (!task.enabled) continue; + SchedulerService.scheduleTask(task, project as unknown as IProject); + taskCount++; + } + } + + this.log.info("tasks scheduled", { + projectCount: activeProjects.length, + taskCount, + }); + } + + /** + * Get user credentials from CLI args or interactive prompt. + * Same pattern as gadget-drone startup. + */ + async getUserCredentials(): Promise { + const args = process.argv.slice(2); + const userArg = args.find((a) => a.startsWith("--user=")); + const passArg = args.find((a) => a.startsWith("--password=")); + + if (userArg && passArg) { + return { + email: userArg.split("=")[1], + password: passArg.split("=")[1], + }; + } + + const { input: inqInput, password: inqPassword } = await import("@inquirer/prompts"); + return { + email: await inqInput({ message: "📧 Enter Email: " }), + password: await inqPassword({ message: "🔑 Enter Password: " }), + }; + } + + /** + * Select a drone for task execution. If only one is available, + * auto-select it. Otherwise, present an interactive selection. + */ + async selectDrone(): Promise { + const drones = await PlatformService.getAvailableDrones(); + if (drones.length === 0) { + throw new Error("No available drones. Start a gadget-drone instance first."); + } + if (drones.length === 1) { + this.log.info("auto-selecting only available drone", { droneId: drones[0]._id }); + return drones[0]; + } + + console.log("\nAvailable Drones:"); + drones.forEach((d, i) => { + console.log(` ${i + 1}. ${d.hostname} (${d.workspaceId}) - ${d.status}`); + }); + + const { select: inqSelect } = await import("@inquirer/prompts"); + const selection = await inqSelect({ + message: "Select a drone for task execution:", + choices: drones.map((d) => ({ + name: `${d.hostname} (${d.workspaceId})`, + value: d, + })), + }); + + return selection; + } + + hookProcessSignals(): void { + process.title = this.name; + + process.on("unhandledRejection", async (error: Error) => { + this.log.error("Unhandled rejection", { error, stack: error.stack }); + const exitCode = await this.stop(); + process.exit(exitCode); + }); + + process.on("warning", (error) => { + if (error.name === "DeprecationWarning") return; + this.log.alert("warning", { error }); + }); + + process.on("SIGINT", async () => { + this.log.info("SIGINT received"); + const exitCode = await this.stop(); + process.exit(exitCode); + }); + } +} + +(async () => { + try { + const tasks = new GadgetTasks(); + await tasks.start(); + } catch (error) { + console.error("failed to start gadget-tasks", error); + process.exit(-1); + } +})(); diff --git a/gadget-tasks/src/lib/process.ts b/gadget-tasks/src/lib/process.ts new file mode 100644 index 0000000..9f658ad --- /dev/null +++ b/gadget-tasks/src/lib/process.ts @@ -0,0 +1,19 @@ +// src/lib/process.ts +// Copyright (C) 2026 Rob Colbert +// Licensed under the Apache License, Version 2.0 + +import { GadgetComponent, GadgetLog } from "@gadget/api"; + +export abstract class GadgetProcess implements GadgetComponent { + protected log: GadgetLog; + + constructor() { + this.log = new GadgetLog(this); + } + + abstract get name(): string; + abstract get slug(): string; + + abstract start(): Promise; + abstract stop(): Promise; +} diff --git a/gadget-tasks/src/lib/service.ts b/gadget-tasks/src/lib/service.ts new file mode 100644 index 0000000..00304ba --- /dev/null +++ b/gadget-tasks/src/lib/service.ts @@ -0,0 +1,19 @@ +// src/lib/service.ts +// Copyright (C) 2026 Rob Colbert +// Licensed under the Apache License, Version 2.0 + +import { GadgetComponent, GadgetLog } from "@gadget/api"; + +export abstract class GadgetService implements GadgetComponent { + public log: GadgetLog; + + constructor() { + this.log = new GadgetLog(this); + } + + abstract get name(): string; + abstract get slug(): string; + + abstract start(): Promise; + abstract stop(): Promise; +} diff --git a/gadget-tasks/src/services/lock.ts b/gadget-tasks/src/services/lock.ts new file mode 100644 index 0000000..4e2e280 --- /dev/null +++ b/gadget-tasks/src/services/lock.ts @@ -0,0 +1,138 @@ +// src/services/lock.ts +// Copyright (C) 2026 Rob Colbert +// Licensed under the Apache License, Version 2.0 + +import os from "node:os"; +import { Redis } from "ioredis"; +import env from "../config/env.ts"; +import { GadgetLog } from "@gadget/api"; + +const log = new GadgetLog({ name: "TaskLock", slug: "task-lock" }); + +const LOCK_KEY = "gadget:tasks:lock"; +const LOCK_TTL_SECONDS = 60; +const HEARTBEAT_INTERVAL_MS = 30000; // 30 seconds + +export interface TaskLockData { + pid: number; + hostname: string; + startedAt: number; + heartbeatAt: number; +} + +class TaskLockService { + private redis: Redis | null = null; + private heartbeatTimer: ReturnType | null = null; + + async connect(): Promise { + this.redis = new Redis({ + host: env.redis.host, + port: env.redis.port, + password: env.redis.password || undefined, + keyPrefix: env.redis.keyPrefix, + lazyConnect: true, + }); + await this.redis.connect(); + log.info("connected to Redis", { + host: env.redis.host, + port: env.redis.port, + }); + } + + /** + * Attempt to acquire the singleton lock. + * Returns true if acquired, false if another instance is running. + */ + async acquire(): Promise { + if (!this.redis) { + throw new Error("Redis connection not initialized"); + } + + const data: TaskLockData = { + pid: process.pid, + hostname: os.hostname(), + startedAt: Date.now(), + heartbeatAt: Date.now(), + }; + + // SET gadget:tasks:lock EX 60 NX + const result = await this.redis.set( + LOCK_KEY, + JSON.stringify(data), + "EX", + LOCK_TTL_SECONDS, + "NX", + ); + + if (result === "OK") { + this.startHeartbeat(); + return true; + } + + // Lock exists — check if stale + const existing = await this.redis.get(LOCK_KEY); + if (existing) { + const info = JSON.parse(existing) as TaskLockData; + const age = Date.now() - info.heartbeatAt; + if (age > LOCK_TTL_SECONDS * 1000) { + // Stale — reclaim + log.warn("stale lock detected, reclaiming", { existingPid: info.pid }); + await this.redis.del(LOCK_KEY); + return this.acquire(); // retry + } + log.error("gadget-tasks is already running", { + pid: info.pid, + hostname: info.hostname, + startedAt: new Date(info.startedAt).toISOString(), + }); + return false; + } + + // Lock disappeared — retry + return this.acquire(); + } + + /** + * Release the lock on shutdown. + */ + async release(): Promise { + if (this.heartbeatTimer) { + clearInterval(this.heartbeatTimer); + this.heartbeatTimer = null; + } + if (this.redis) { + try { + await this.redis.del(LOCK_KEY); + } catch { + // best effort + } + try { + await this.redis.quit(); + } catch { + // best effort + } + this.redis = null; + } + } + + /** + * Start heartbeat timer to refresh lock TTL. + */ + private startHeartbeat(): void { + this.heartbeatTimer = setInterval(async () => { + if (!this.redis) return; + try { + const existing = await this.redis.get(LOCK_KEY); + if (existing) { + const data = JSON.parse(existing) as TaskLockData; + data.heartbeatAt = Date.now(); + await this.redis.set(LOCK_KEY, JSON.stringify(data), "EX", LOCK_TTL_SECONDS); + } + } catch (error) { + log.error("heartbeat failed", { error }); + } + }, HEARTBEAT_INTERVAL_MS); + } +} + +export default new TaskLockService(); diff --git a/gadget-tasks/src/services/platform.ts b/gadget-tasks/src/services/platform.ts new file mode 100644 index 0000000..407f044 --- /dev/null +++ b/gadget-tasks/src/services/platform.ts @@ -0,0 +1,572 @@ +// src/services/platform.ts +// Copyright (C) 2026 Rob Colbert +// Licensed under the Apache License, Version 2.0 + +import { io, Socket } from "socket.io-client"; +import { GadgetService } from "../lib/service.ts"; +import env from "../config/env.ts"; +import { + type IChatSession, + type IProject, + type IDroneRegistration, + type IProjectTask, + type IUser, + WorkspaceMode, + type SubmitPromptCallbackData, +} from "@gadget/api"; + +/** + * REST API response envelope + */ +interface ApiResponse { + success: boolean; + message?: string; + data?: T; +} + +/** + * Work order completion result + */ +interface WorkOrderResult { + success: boolean; + message?: string; +} + +/** + * Pending work order tracker — maps turnId to a Promise resolver + */ +interface PendingWorkOrder { + resolve: (value: WorkOrderResult) => void; + reject: (reason: Error) => void; +} + +class PlatformService extends GadgetService { + private jwt: string | null = null; + private socket: Socket | null = null; + private selectedDrone: IDroneRegistration | null = null; + + /** + * Tracks in-flight work orders by turnId. When the server emits + * "workOrderComplete", the corresponding Promise is resolved. + */ + private pendingWorkOrders: Map = new Map(); + + /** + * Interval timer for session heartbeats sent to the drone + * via the platform. Prevents the drone's 120s heartbeat timeout. + */ + private heartbeatInterval: ReturnType | null = null; + + get name(): string { + return "PlatformService"; + } + get slug(): string { + return "svc:platform"; + } + + async start(): Promise { + this.log.info("started"); + } + + async stop(): Promise { + this.stopHeartbeat(); + + if (this.socket) { + this.socket.disconnect(); + this.socket = null; + } + + // Reject any pending work orders + for (const [turnId, pending] of this.pendingWorkOrders) { + pending.reject(new Error("PlatformService shutting down")); + } + this.pendingWorkOrders.clear(); + + this.log.info("stopped"); + } + + setSelectedDrone(drone: IDroneRegistration): void { + this.selectedDrone = drone; + } + + getSelectedDrone(): IDroneRegistration | null { + return this.selectedDrone; + } + + // ─── REST API Methods ─────────────────────────────────────────────── + + /** + * Authenticate with the gadget-code platform. + * POST /api/v1/auth/sign-in + * Returns: JWT token string + */ + async authenticate(email: string, password: string): Promise { + const json = await this.apiRequest<{ token: string; user: IUser }>( + "POST", + "/auth/sign-in", + { email, password }, + ); + + if (!json.data?.token) { + throw new Error("Authentication response did not contain a token"); + } + + this.jwt = json.data.token; + this.log.info("authenticated with platform", { email }); + return this.jwt; + } + + /** + * Get all projects for the authenticated user. + * GET /api/v1/projects + */ + async getProjects(): Promise { + const json = await this.apiRequest("GET", "/projects"); + return json.data ?? []; + } + + /** + * Create a new chat session. + * POST /api/v1/chat-sessions + */ + async createSession(opts: { + projectId: string; + providerId: string; + selectedModel: string; + mode: string; + name?: string; + }): Promise { + const json = await this.apiRequest("POST", "/chat-sessions", { + projectId: opts.projectId, + providerId: opts.providerId, + selectedModel: opts.selectedModel, + mode: opts.mode, + name: opts.name, + }); + + if (!json.data) { + throw new Error("Create session response did not contain data"); + } + + return json.data; + } + + /** + * Get available drone registrations. + * GET /api/v1/drone/registration + */ + async getAvailableDrones(): Promise { + const json = await this.apiRequest( + "GET", + "/drone/registration", + ); + return json.data ?? []; + } + + /** + * Update a task's lastRun field. + * PATCH /api/v1/projects/:projectId/tasks/:taskId/lastRun + */ + async updateTaskLastRun( + projectId: string, + taskId: string, + sessionId: string, + ): Promise { + await this.apiRequest( + "PATCH", + `/projects/${projectId}/tasks/${taskId}/lastRun`, + { sessionId }, + ); + this.log.info("task lastRun updated", { projectId, taskId, sessionId }); + } + + // ─── REST API Helper ──────────────────────────────────────────────── + + private async apiRequest( + method: string, + path: string, + body?: unknown, + ): Promise> { + const url = `${env.platform.baseUrl}/api/v1${path}`; + const headers: Record = { + Accept: "application/json", + "Content-Type": "application/json", + }; + if (this.jwt) { + headers["Authorization"] = `Bearer ${this.jwt}`; + } + + const response = await fetch(url, { + method, + headers, + body: body ? JSON.stringify(body) : undefined, + }); + + const json = (await response.json()) as ApiResponse; + if (!json.success) { + const error = new Error(json.message || "API request failed"); + (error as any).statusCode = response.status; + throw error; + } + + return json; + } + + // ─── Socket.IO Methods ────────────────────────────────────────────── + + /** + * Connect to the platform via Socket.IO using JWT auth. + * Mirrors the browser IDE's socket connection behavior. + */ + async connectSocket(): Promise { + return new Promise((resolve, reject) => { + if (!this.jwt) { + reject(new Error("Cannot connect socket: not authenticated")); + return; + } + + const options = { + auth: { token: this.jwt }, + reconnection: true, + reconnectionAttempts: 10, + reconnectionDelay: 1000, + reconnectionDelayMax: 5000, + timeout: 5000, + transports: ["websocket"], + }; + + this.socket = io(env.platform.baseUrl, options); + + this.socket.on("connect_error", (err: Error) => { + this.log.error("socket connect error", { err: err.message }); + reject(err); + }); + + this.socket.on("connect", () => { + this.log.info("connected to platform via Socket.IO"); + this.startHeartbeat(); + resolve(); + }); + + // Listen for work order completion from the server + this.socket.on( + "workOrderComplete", + (turnId: string, success: boolean, message?: string) => { + this.onWorkOrderComplete(turnId, success, message); + }, + ); + + this.socket.on("disconnect", (reason: string) => { + this.log.info("socket disconnected", { reason }); + }); + + this.socket.io.on("reconnect", (attempt: number) => { + this.log.info("socket reconnected", { attempt }); + }); + + this.socket.io.on("reconnect_attempt", (attempt: number) => { + this.log.info("socket reconnect attempt", { attempt }); + }); + + this.socket.io.on("reconnect_failed", () => { + this.log.error("socket reconnection failed after max attempts"); + }); + }); + } + + /** + * Lock the selected drone to a session. + * Socket emit: requestSessionLock(registration, project, session, callback) + */ + async requestSessionLock( + project: IProject, + session: IChatSession, + ): Promise { + if (!this.socket?.connected || !this.selectedDrone) { + this.log.warn("cannot request session lock: socket not connected or no drone selected"); + return false; + } + + return new Promise((resolve) => { + this.socket!.emit( + "requestSessionLock", + this.selectedDrone, + project, + session, + (success: boolean, _chatSessionId: string) => { + if (success) { + this.log.info("session lock acquired", { + droneId: this.selectedDrone!._id, + sessionId: session._id, + }); + } else { + this.log.warn("session lock denied", { + droneId: this.selectedDrone!._id, + }); + } + resolve(success); + }, + ); + }); + } + + /** + * Set the drone's workspace mode. + * Socket emit: requestWorkspaceMode(registration, project, session, mode, callback) + */ + async requestWorkspaceMode( + project: IProject, + session: IChatSession, + mode: WorkspaceMode, + ): Promise { + if (!this.socket?.connected || !this.selectedDrone) { + this.log.warn("cannot request workspace mode: socket not connected or no drone selected"); + return false; + } + + return new Promise((resolve) => { + this.socket!.emit( + "requestWorkspaceMode", + this.selectedDrone, + project, + session, + mode, + (success: boolean, currentMode: WorkspaceMode, reason?: string) => { + if (success) { + this.log.info("workspace mode set", { + mode: currentMode, + }); + } else { + this.log.warn("workspace mode request failed", { + mode, + currentMode, + reason, + }); + } + resolve(success); + }, + ); + }); + } + + /** + * Submit a prompt and wait for the work order to complete. + * Socket emit: submitPrompt(content, callback) + * Then waits for "workOrderComplete" server event. + */ + async submitPromptAndWait( + content: string, + ): Promise { + if (!this.socket?.connected) { + throw new Error("Socket not connected"); + } + + return new Promise((resolve, reject) => { + this.socket!.emit( + "submitPrompt", + content, + (success: boolean, data: SubmitPromptCallbackData) => { + if (!success) { + reject( + new Error(data.message || "Prompt submission failed"), + ); + return; + } + + const turnId = data.turnId; + if (!turnId) { + reject(new Error("No turnId returned from submitPrompt")); + return; + } + + // Store the resolver — it will be called when workOrderComplete fires + this.pendingWorkOrders.set(turnId, { resolve, reject }); + this.log.info("prompt submitted, waiting for work order completion", { + turnId, + }); + }, + ); + }); + } + + /** + * Release the session lock on the drone. + * Socket emit: releaseSessionLock(registration, project, session, callback) + */ + async releaseSessionLock( + project: IProject, + session: IChatSession, + ): Promise { + if (!this.socket?.connected || !this.selectedDrone) { + this.log.warn("cannot release session lock: socket not connected or no drone selected"); + return; + } + + return new Promise((resolve) => { + this.socket!.emit( + "releaseSessionLock", + this.selectedDrone, + project, + session, + (success: boolean) => { + if (success) { + this.log.info("session lock released", { + droneId: this.selectedDrone!._id, + sessionId: session._id, + }); + } else { + this.log.warn("session lock release failed", { + droneId: this.selectedDrone!._id, + sessionId: session._id, + }); + } + resolve(); + }, + ); + }); + } + + // ─── Task Execution ───────────────────────────────────────────────── + + /** + * Execute a task through the full headless IDE pipeline: + * 1. Create ChatSession via REST API + * 2. Lock the drone to this session + * 3. Set workspace mode to Agent + * 4. Submit prompt and wait for completion + * 5. Release session lock + * 6. Update task.lastRun via dedicated PATCH endpoint + */ + async executeTask(task: IProjectTask, project: IProject): Promise { + this.log.info("task execution starting", { + taskId: task._id, + taskName: task.name, + projectSlug: project.slug, + mode: task.mode, + }); + + let session: IChatSession | null = null; + + try { + // 1. Create ChatSession via REST API + session = await this.createSession({ + projectId: project._id as string, + providerId: task.provider as string, + selectedModel: task.selectedModel as string, + mode: task.mode, + name: `Task: ${task.name}`, + }); + + // 2. Lock the drone to this session + const locked = await this.requestSessionLock(project, session); + if (!locked) { + throw new Error("Failed to lock drone for task execution"); + } + + // 3. Set workspace mode to Agent + const modeSet = await this.requestWorkspaceMode( + project, + session, + WorkspaceMode.Agent, + ); + if (!modeSet) { + throw new Error("Failed to set drone workspace mode"); + } + + // 4. Submit prompt and wait for completion + const result = await this.submitPromptAndWait(task.content); + + // 5. Release session lock + await this.releaseSessionLock(project, session); + + // 6. Update task.lastRun via dedicated PATCH endpoint + await this.updateTaskLastRun( + project._id as string, + task._id as string, + session._id as string, + ); + + this.log.info("task execution completed", { + taskId: task._id, + taskName: task.name, + sessionId: session._id, + success: result.success, + }); + } catch (error) { + const msg = error instanceof Error ? error.message : String(error); + this.log.error("task execution failed", { + taskId: task._id, + taskName: task.name, + error: msg, + }); + + // Try to release the session lock on failure + if (session && this.socket?.connected && this.selectedDrone) { + try { + await this.releaseSessionLock(project, session); + } catch { + // Best-effort release + } + } + + throw error; + } + } + + // ─── Work Order Completion ────────────────────────────────────────── + + /** + * Called when the server emits "workOrderComplete". + * Resolves the corresponding pending Promise. + */ + private onWorkOrderComplete( + turnId: string, + success: boolean, + message?: string, + ): void { + const pending = this.pendingWorkOrders.get(turnId); + if (pending) { + this.pendingWorkOrders.delete(turnId); + pending.resolve({ success, message }); + this.log.info("work order completed", { turnId, success, message }); + } else { + this.log.warn("received workOrderComplete for unknown turnId", { + turnId, + success, + }); + } + } + + // ─── Heartbeat ────────────────────────────────────────────────────── + + /** + * Start sending periodic session heartbeats to the drone + * via the platform. This prevents the drone's 120-second + * heartbeat timeout from firing while a task is active. + */ + private startHeartbeat(): void { + if (this.heartbeatInterval) return; // already running + + this.heartbeatInterval = setInterval(() => { + if (this.socket?.connected) { + this.socket.emit("sessionHeartbeat", (ack: boolean) => { + if (!ack) { + this.log.warn("sessionHeartbeat: drone did not acknowledge"); + } + }); + } + }, 19_000); // every 19 seconds, same as browser IDE + } + + /** + * Stop the heartbeat interval. + */ + private stopHeartbeat(): void { + if (this.heartbeatInterval) { + clearInterval(this.heartbeatInterval); + this.heartbeatInterval = null; + } + } +} + +export default new PlatformService(); diff --git a/gadget-tasks/src/services/scheduler.ts b/gadget-tasks/src/services/scheduler.ts new file mode 100644 index 0000000..a7864a8 --- /dev/null +++ b/gadget-tasks/src/services/scheduler.ts @@ -0,0 +1,133 @@ +// src/services/scheduler.ts +// Copyright (C) 2026 Rob Colbert +// Licensed under the Apache License, Version 2.0 + +import env from "../config/env.ts"; +import { CronJob } from "cron"; +import type { IProjectTask, IProject } from "@gadget/api"; +import { GadgetService } from "../lib/service.ts"; +import PlatformService from "./platform.ts"; + +class SchedulerService extends GadgetService { + private jobs: Map = new Map(); + private _activeCount: number = 0; + + get name(): string { + return "SchedulerService"; + } + get slug(): string { + return "svc:scheduler"; + } + + get activeCount(): number { + return this._activeCount; + } + + get scheduledCount(): number { + return this.jobs.size; + } + + async start(): Promise { + this.log.info("started"); + } + + async stop(): Promise { + for (const [taskId, job] of this.jobs) { + job.stop(); + this.log.info("stopped cron job", { taskId }); + } + this.jobs.clear(); + this.log.info("stopped"); + } + + /** + * Schedules a task by creating a CronJob. + * Key: task._id (string) + */ + scheduleTask(task: IProjectTask, project: IProject): void { + // If already scheduled, unschedule first + const taskId = String(task._id); + if (this.jobs.has(taskId)) { + this.unscheduleTask(taskId); + } + + try { + const job = new CronJob( + task.crontab, // cron expression + async () => { + await this.executeTask(task, project); + }, + null, // onComplete + true, // start + env.timezone, // timeZone + ); + + this.jobs.set(taskId, job); + this.log.info("task scheduled", { + taskId, + taskName: task.name, + crontab: task.crontab, + projectSlug: project.slug, + }); + } catch (error) { + const err = error as Error; + this.log.error("failed to schedule task", { + taskId, + crontab: task.crontab, + error: err.message, + }); + } + } + + /** + * Removes a scheduled task. + */ + unscheduleTask(taskId: string): void { + const job = this.jobs.get(taskId); + if (job) { + job.stop(); + this.jobs.delete(taskId); + this.log.info("task unscheduled", { taskId }); + } + } + + /** + * Execute a task with concurrency control. + * Delegates to PlatformService.executeTask() which handles + * the full headless IDE pipeline. + */ + private async executeTask(task: IProjectTask, project: IProject): Promise { + // Check concurrency + if (this._activeCount >= env.concurrency) { + this.log.warn("task skipped: concurrency limit reached", { + taskId: task._id, + taskName: task.name, + activeCount: this._activeCount, + maxConcurrency: env.concurrency, + }); + return; + } + + this._activeCount++; + try { + this.log.info("executing task", { + taskId: task._id, + taskName: task.name, + projectSlug: project.slug, + activeCount: this._activeCount, + }); + + await PlatformService.executeTask(task, project); + } catch (error) { + const err = error as Error; + this.log.error("task execution failed", { + taskId: task._id, + error: err.message, + }); + } finally { + this._activeCount--; + } + } +} + +export default new SchedulerService(); diff --git a/gadget-tasks/tsconfig.json b/gadget-tasks/tsconfig.json new file mode 100644 index 0000000..d4a77df --- /dev/null +++ b/gadget-tasks/tsconfig.json @@ -0,0 +1,23 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "NodeNext", + "moduleResolution": "NodeNext", + "lib": ["ES2022"], + "outDir": "./dist", + "rootDir": "./src", + "types": ["node"], + "strict": true, + "esModuleInterop": true, + "skipLibCheck": true, + "forceConsistentCasingInFileNames": true, + "resolveJsonModule": true, + "allowImportingTsExtensions": true, + "rewriteRelativeImportExtensions": true, + "declaration": true, + "declarationMap": true, + "sourceMap": true + }, + "include": ["src/**/*"], + "exclude": ["node_modules", "dist"] +} diff --git a/packages/config/src/loader.ts b/packages/config/src/loader.ts index 49e6c00..48d2066 100644 --- a/packages/config/src/loader.ts +++ b/packages/config/src/loader.ts @@ -8,7 +8,7 @@ import fs from "node:fs"; import path from "node:path"; import os from "node:os"; import yaml from "js-yaml"; -import type { GadgetCodeConfig, GadgetDroneConfig } from "./types.js"; +import type { GadgetCodeConfig, GadgetDroneConfig, GadgetTasksConfig } from "./types.js"; /** * Substitute environment variables in YAML values. @@ -121,6 +121,13 @@ export function loadGadgetDroneConfig(): GadgetDroneConfig { return loadYamlConfig("gadget-drone.yaml"); } +/** + * Load Gadget Tasks configuration + */ +export function loadGadgetTasksConfig(): GadgetTasksConfig { + return loadYamlConfig("gadget-tasks.yaml"); +} + /** * Resolve a path that may contain ~ for home directory */ diff --git a/packages/config/src/types.ts b/packages/config/src/types.ts index 97c47f1..81ed159 100644 --- a/packages/config/src/types.ts +++ b/packages/config/src/types.ts @@ -150,6 +150,40 @@ export interface GadgetDroneConfig { }; } +/** + * Gadget Tasks Worker Configuration + */ +export interface GadgetTasksConfig { + timezone?: string; + platform: { + baseUrl: string; + }; + redis: { + host?: string; + port?: number; + password?: string; + keyPrefix?: string; + }; + concurrency?: number; // max concurrent tasks, default 1 (sequential) + logging?: { + console?: { + enabled?: boolean; + }; + file?: { + enabled?: boolean; + path?: string; + name?: string; + maxWritesPerFile?: number; + maxFiles?: number; + }; + levels?: { + debug?: boolean; + info?: boolean; + warn?: boolean; + }; + }; +} + /** * Configuration search result */ diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 9f27360..6370ca5 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -446,6 +446,37 @@ importers: specifier: 4.1.5 version: 4.1.5(@types/node@25.6.0)(jsdom@29.0.2)(vite@8.0.10(@types/node@25.6.0)(esbuild@0.27.7)(jiti@2.6.1)(less@4.6.4)(tsx@4.21.0)) + gadget-tasks: + dependencies: + '@gadget/api': + specifier: workspace:* + version: link:../packages/api + '@gadget/config': + specifier: workspace:* + version: link:../packages/config + '@inquirer/prompts': + specifier: ^8.4.2 + version: 8.4.2(@types/node@25.6.0) + cron: + specifier: ^4.3.1 + version: 4.4.0 + ioredis: + specifier: ^5.6.0 + version: 5.10.1 + socket.io-client: + specifier: ^4.8.1 + version: 4.8.3 + devDependencies: + '@types/node': + specifier: ^25.6.0 + version: 25.6.0 + tsx: + specifier: ^4.21.0 + version: 4.21.0 + typescript: + specifier: 6.0.3 + version: 6.0.3 + packages/ai: dependencies: googleapis: