feat(gadget-tasks): course correction — headless IDE client architecture

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
This commit is contained in:
Rob Colbert 2026-05-17 00:48:39 -04:00
parent 79a6f77659
commit b906cb2377
16 changed files with 1366 additions and 1 deletions

View File

@ -32,6 +32,7 @@ export class ProjectApiControllerV1 extends DtpController {
this.router.get("/:projectId", this.getProject.bind(this)); this.router.get("/:projectId", this.getProject.bind(this));
this.router.put("/:projectId", this.updateProject.bind(this)); this.router.put("/:projectId", this.updateProject.bind(this));
this.router.delete("/:projectId", this.deleteProject.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<void> { async getProjects(req: Request, res: Response): Promise<void> {
@ -256,6 +257,49 @@ export class ProjectApiControllerV1 extends DtpController {
}); });
} }
} }
async updateTaskLastRun(req: Request, res: Response): Promise<void> {
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; export default ProjectApiControllerV1;

View File

@ -257,6 +257,19 @@ class ProjectService extends DtpService {
); );
} }
async updateTaskLastRun(projectId: string, taskId: string, sessionId: string): Promise<void> {
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<void> { async delete(project: IProject): Promise<void> {
await Project.deleteOne({ _id: project._id }); await Project.deleteOne({ _id: project._id });
} }

1
gadget-tasks/.gitignore vendored Normal file
View File

@ -0,0 +1 @@
dist

View File

@ -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

39
gadget-tasks/package.json Normal file
View File

@ -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"
}
}

View File

@ -0,0 +1,88 @@
// config/env.ts
// Copyright (C) 2026 Rob Colbert <rob.colbert@openplatform.us>
// 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<T>(filePath: string): Promise<T> {
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<typeof PackageJson>(
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 */

View File

@ -0,0 +1,191 @@
// src/gadget-tasks.ts
// Copyright (C) 2026 Rob Colbert <rob.colbert@openplatform.us>
// 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<void> {
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<number> {
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<void> {
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<UserCredentials> {
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<IDroneRegistration> {
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);
}
})();

View File

@ -0,0 +1,19 @@
// src/lib/process.ts
// Copyright (C) 2026 Rob Colbert <rob.colbert@openplatform.us>
// 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<void>;
abstract stop(): Promise<number>;
}

View File

@ -0,0 +1,19 @@
// src/lib/service.ts
// Copyright (C) 2026 Rob Colbert <rob.colbert@openplatform.us>
// 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<void>;
abstract stop(): Promise<void>;
}

View File

@ -0,0 +1,138 @@
// src/services/lock.ts
// Copyright (C) 2026 Rob Colbert <rob.colbert@openplatform.us>
// 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<typeof setInterval> | null = null;
async connect(): Promise<void> {
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<boolean> {
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 <data> 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<void> {
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();

View File

@ -0,0 +1,572 @@
// src/services/platform.ts
// Copyright (C) 2026 Rob Colbert <rob.colbert@openplatform.us>
// 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<T = unknown> {
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<string, PendingWorkOrder> = new Map();
/**
* Interval timer for session heartbeats sent to the drone
* via the platform. Prevents the drone's 120s heartbeat timeout.
*/
private heartbeatInterval: ReturnType<typeof setInterval> | null = null;
get name(): string {
return "PlatformService";
}
get slug(): string {
return "svc:platform";
}
async start(): Promise<void> {
this.log.info("started");
}
async stop(): Promise<void> {
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<string> {
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<IProject[]> {
const json = await this.apiRequest<IProject[]>("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<IChatSession> {
const json = await this.apiRequest<IChatSession>("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<IDroneRegistration[]> {
const json = await this.apiRequest<IDroneRegistration[]>(
"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<void> {
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<T = unknown>(
method: string,
path: string,
body?: unknown,
): Promise<ApiResponse<T>> {
const url = `${env.platform.baseUrl}/api/v1${path}`;
const headers: Record<string, string> = {
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<T>;
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<void> {
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<boolean> {
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<boolean> {
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<WorkOrderResult> {
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<void> {
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<void> {
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();

View File

@ -0,0 +1,133 @@
// src/services/scheduler.ts
// Copyright (C) 2026 Rob Colbert <rob.colbert@openplatform.us>
// 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<string, CronJob> = 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<void> {
this.log.info("started");
}
async stop(): Promise<void> {
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<void> {
// 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();

View File

@ -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"]
}

View File

@ -8,7 +8,7 @@ import fs from "node:fs";
import path from "node:path"; import path from "node:path";
import os from "node:os"; import os from "node:os";
import yaml from "js-yaml"; 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. * Substitute environment variables in YAML values.
@ -121,6 +121,13 @@ export function loadGadgetDroneConfig(): GadgetDroneConfig {
return loadYamlConfig<GadgetDroneConfig>("gadget-drone.yaml"); return loadYamlConfig<GadgetDroneConfig>("gadget-drone.yaml");
} }
/**
* Load Gadget Tasks configuration
*/
export function loadGadgetTasksConfig(): GadgetTasksConfig {
return loadYamlConfig<GadgetTasksConfig>("gadget-tasks.yaml");
}
/** /**
* Resolve a path that may contain ~ for home directory * Resolve a path that may contain ~ for home directory
*/ */

View File

@ -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 * Configuration search result
*/ */

View File

@ -446,6 +446,37 @@ importers:
specifier: 4.1.5 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)) 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: packages/ai:
dependencies: dependencies:
googleapis: googleapis: