More progress

Author: cte

apps/cloud-agents/.env.example

@@ -2,3 +2,4 @@ DATABASE_URL=postgresql://postgres:password@localhost:5433/cloud_agents
 REDIS_URL=redis://localhost:6380
 GITHUB_WEBHOOK_SECRET=your-webhook-secret-here
 OPENROUTER_API_KEY=your-openrouter-api-key
+SLACK_API_TOKEN=xoxb-...

apps/cloud-agents/src/lib/job.ts

@@ -2,7 +2,7 @@ import { eq } from "drizzle-orm"
 import { Job } from "bullmq"
 
 import { db, cloudJobs, type UpdateCloudJob } from "@/db"
-import type { JobTypes, JobType, JobStatus, JobParams } from "@/types"
+import type { JobType, JobStatus, JobParams } from "@/types"
 
 import { fixGitHubIssue } from "./jobs/fixGitHubIssue"
 
@@ -15,7 +15,7 @@ export async function processJob<T extends JobType>({ data: { type, payload, job
 
 		switch (type) {
 			case "github.issue.fix":
-				result = await fixGitHubIssue(jobId, payload as JobTypes["github.issue.fix"])
+				result = await fixGitHubIssue(payload)
 				break
 			default:
 				throw new Error(`Unknown job type: ${type}`)

apps/cloud-agents/src/lib/jobs/fixGitHubIssue.ts

@@ -1,46 +1,48 @@
 import * as path from "path"
 import * as os from "node:os"
 
-import type { JobTypes } from "@/types"
+import type { JobType, JobPayload } from "@/types"
 
 import { runTask } from "../runTask"
 import { Logger } from "../logger"
 
-export async function fixGitHubIssue(
-	jobId: number,
-	payload: JobTypes["github.issue.fix"],
-): Promise<{
+const jobType: JobType = "github.issue.fix"
+
+type FixGitHubIssueJobPayload = JobPayload<"github.issue.fix">
+
+export async function fixGitHubIssue(jobPayload: FixGitHubIssueJobPayload): Promise<{
 	repo: string
 	issue: number
 	result: unknown
 }> {
 	const prompt = `
 Fix the following GitHub issue:
 
-Repository: ${payload.repo}
-Issue #${payload.issue}: ${payload.title}
+Repository: ${jobPayload.repo}
+Issue #${jobPayload.issue}: ${jobPayload.title}
 
 Description:
-${payload.body}
+${jobPayload.body}
 
-${payload.labels && payload.labels.length > 0 ? `Labels: ${payload.labels.join(", ")}` : ""}
+${jobPayload.labels && jobPayload.labels.length > 0 ? `Labels: ${jobPayload.labels.join(", ")}` : ""}
 
 Please analyze the issue, understand what needs to be fixed, and implement a solution.
-If you're reasonably satisfied with the solution then create and submit a pull request using the "gh" command line tool.
-You'll first need to create a new branch for the pull request.
 
-Make sure to reference the issue in the pull request description.
+If you're reasonably satisfied with the solution then create and submit a pull request using the "gh" command line tool:
+gh pr create --title "Fixes #${jobPayload.issue}\n\n[Your PR description here.]" --fill --template "pull_request_template.md"
+
+You'll first need to create a new branch for the pull request.
 `.trim()
 
+	const { repo, issue } = jobPayload
+
 	const result = await runTask({
+		jobType,
+		jobPayload,
 		prompt,
 		publish: async () => {},
-		logger: new Logger({
-			logDir: path.resolve(os.tmpdir(), "logs"),
-			filename: "worker.log",
-			tag: "worker",
-		}),
+		logger: new Logger({ logDir: path.resolve(os.tmpdir(), "logs"), filename: "worker.log", tag: "worker" }),
 	})
 
-	return { repo: payload.repo, issue: payload.issue, result }
+	return { repo, issue, result }
 }

apps/cloud-agents/src/lib/runTask.ts

@@ -8,8 +8,11 @@ import { execa } from "execa"
 import { type TaskEvent, TaskCommandName, RooCodeEventName, IpcMessageType, EVALS_SETTINGS } from "@roo-code/types"
 import { IpcClient } from "@roo-code/ipc"
 
+import type { JobPayload, JobType } from "@/types"
+
 import { Logger } from "./logger"
 import { isDockerContainer } from "./utils"
+import { SlackNotifier } from "./slack"
 
 const TIMEOUT = 30 * 60 * 1_000
 
@@ -20,13 +23,21 @@ class SubprocessTimeoutError extends Error {
 	}
 }
 
-type RunTaskOptions = {
+type RunTaskOptions<T extends JobType> = {
+	jobType: T
+	jobPayload: JobPayload<T>
 	prompt: string
 	publish: (taskEvent: TaskEvent) => Promise<void>
 	logger: Logger
 }
 
-export const runTask = async ({ prompt, publish, logger }: RunTaskOptions) => {
+export const runTask = async <T extends JobType>({
+	jobType,
+	jobPayload,
+	prompt,
+	publish,
+	logger,
+}: RunTaskOptions<T>) => {
 	const workspacePath = "/Users/cte/Documents/Roomote-Control" // findGitRoot(process.cwd())
 	const ipcSocketPath = path.resolve(os.tmpdir(), `${crypto.randomUUID().slice(0, 8)}.sock`)
 	const env = { ROO_CODE_IPC_SOCKET_PATH: ipcSocketPath }
@@ -73,13 +84,16 @@ export const runTask = async ({ prompt, publish, logger }: RunTaskOptions) => {
 		}
 	}
 
-	let taskStartedAt = Date.now() // eslint-disable-line @typescript-eslint/no-unused-vars
+	let taskStartedAt = Date.now()
 	let taskFinishedAt: number | undefined
 	let taskAbortedAt: number | undefined
 	let taskTimedOut: boolean = false
 	let rooTaskId: string | undefined
 	let isClientDisconnected = false
 
+	const slackNotifier = new SlackNotifier(logger)
+	let slackThreadTs: string | null = null
+
 	const ignoreEvents: Record<"broadcast" | "log", RooCodeEventName[]> = {
 		broadcast: [RooCodeEventName.Message],
 		log: [RooCodeEventName.TaskTokenUsageUpdated, RooCodeEventName.TaskAskResponded],
@@ -105,14 +119,23 @@ export const runTask = async ({ prompt, publish, logger }: RunTaskOptions) => {
 		if (eventName === RooCodeEventName.TaskStarted) {
 			taskStartedAt = Date.now()
 			rooTaskId = payload[0]
+			slackThreadTs = await slackNotifier.postTaskStarted({ jobType, jobPayload, rooTaskId })
 		}
 
 		if (eventName === RooCodeEventName.TaskAborted) {
 			taskAbortedAt = Date.now()
+
+			if (slackThreadTs) {
+				await slackNotifier.postTaskUpdated(slackThreadTs, "Task was aborted", "warning")
+			}
 		}
 
 		if (eventName === RooCodeEventName.TaskCompleted) {
 			taskFinishedAt = Date.now()
+
+			if (slackThreadTs) {
+				await slackNotifier.postTaskCompleted(slackThreadTs, true, taskFinishedAt - taskStartedAt, rooTaskId)
+			}
 		}
 	})
 
@@ -142,6 +165,10 @@ export const runTask = async ({ prompt, publish, logger }: RunTaskOptions) => {
 		taskTimedOut = true
 		logger.error("time limit reached")
 
+		if (slackThreadTs) {
+			await slackNotifier.postTaskUpdated(slackThreadTs, "Task timed out after 30 minutes", "error")
+		}
+
 		if (rooTaskId && !isClientDisconnected) {
 			logger.info("cancelling task")
 			client.sendCommand({ commandName: TaskCommandName.CancelTask, data: rooTaskId })
@@ -153,6 +180,11 @@ export const runTask = async ({ prompt, publish, logger }: RunTaskOptions) => {
 
 	if (!taskFinishedAt && !taskTimedOut) {
 		logger.error("client disconnected before task finished")
+
+		if (slackThreadTs) {
+			await slackNotifier.postTaskUpdated(slackThreadTs, "Client disconnected before task completion", "error")
+		}
+
 		throw new Error("Client disconnected before task completion.")
 	}
 

apps/cloud-agents/src/lib/slack.ts

@@ -0,0 +1,170 @@
+import { JobPayload, JobType } from "@/types"
+import { Logger } from "./logger"
+
+export interface SlackMessage {
+	text: string
+	blocks?: unknown[]
+	attachments?: unknown[]
+	thread_ts?: string
+	channel?: string
+}
+
+// Example response:
+// {
+//     "ok": true,
+//     "channel": "C123ABC456",
+//     "ts": "1503435956.000247",
+//     "message": {
+//         "text": "Here's a message for you",
+//         "username": "ecto1",
+//         "bot_id": "B123ABC456",
+//         "attachments": [
+//             {
+//                 "text": "This is an attachment",
+//                 "id": 1,
+//                 "fallback": "This is an attachment's fallback"
+//             }
+//         ],
+//         "type": "message",
+//         "subtype": "bot_message",
+//         "ts": "1503435956.000247"
+//     }
+// }
+
+export interface SlackResponse {
+	ok: boolean
+	channel?: string
+	ts?: string
+	error?: string
+	message?: Record<string, unknown>
+}
+
+export class SlackNotifier {
+	private readonly logger: Logger
+	private readonly token: string
+
+	constructor(logger: Logger, token: string = process.env.SLACK_API_TOKEN!) {
+		this.logger = logger
+		this.token = token
+	}
+
+	private async postMessage(message: SlackMessage): Promise<string | null> {
+		try {
+			const messageWithChannel = { ...message, channel: message.channel || "#roomote-control" }
+
+			const response = await fetch("https://slack.com/api/chat.postMessage", {
+				method: "POST",
+				headers: { "Content-Type": "application/json", Authorization: `Bearer ${this.token}` },
+				body: JSON.stringify(messageWithChannel),
+			})
+
+			if (!response.ok) {
+				this.logger.error(`Slack API failed: ${response.status} ${response.statusText}`)
+				return null
+			}
+
+			const result: SlackResponse = await response.json()
+
+			if (!result.ok) {
+				if (result.error === "not_in_channel") {
+					this.logger.error(
+						`Slack bot is not a member of channel "${messageWithChannel.channel}". ` +
+							`Please add the bot to the channel or ensure the channel exists. ` +
+							`Error: ${result.error}`,
+					)
+				} else {
+					this.logger.error(`Slack API error: ${result.error}`)
+				}
+				return null
+			}
+
+			return result.ts ?? null
+		} catch (error) {
+			this.logger.error("Failed to send Slack message:", error)
+			return null
+		}
+	}
+
+	public async postTaskStarted<T extends JobType>({
+		jobType,
+		jobPayload,
+		rooTaskId,
+	}: {
+		jobType: T
+		jobPayload: JobPayload<T>
+		rooTaskId: string
+	}) {
+		switch (jobType) {
+			case "github.issue.fix":
+				return await this.postMessage({
+					text: `🚀 Task Started`,
+					blocks: [
+						{
+							type: "header",
+							text: { type: "plain_text", text: "🚀 Roo Code Task Started" },
+						},
+						{
+							type: "section",
+							text: {
+								type: "mrkdwn",
+								text: `Creating a pull request for <https://github.com/RooCodeInc/Roo-Code/issues/${jobPayload.issue}|GitHub Issue #${jobPayload.issue}>`,
+							},
+						},
+						{
+							type: "context",
+							elements: [
+								{
+									type: "mrkdwn",
+									text: `jobType: ${jobType}, rooTaskId: ${rooTaskId}`,
+								},
+							],
+						},
+					],
+				})
+			default:
+				throw new Error(`Unknown job type: ${jobType}`)
+		}
+	}
+
+	public async postTaskUpdated(
+		threadTs: string,
+		text: string,
+		status?: "info" | "success" | "warning" | "error",
+	): Promise<void> {
+		const emoji = { info: "ℹ️", success: "✅", warning: "⚠️", error: "❌" }[status || "info"]
+		await this.postMessage({ text: `${emoji} ${text}`, thread_ts: threadTs })
+	}
+
+	public async postTaskCompleted(
+		threadTs: string,
+		success: boolean,
+		duration: number,
+		taskId?: string,
+	): Promise<void> {
+		const status = success ? "✅ Completed" : "❌ Failed"
+		const durationText = `${Math.round(duration / 1000)}s`
+
+		await this.postMessage({
+			text: `${status} Task finished in ${durationText}`,
+			blocks: [
+				{
+					type: "section",
+					text: {
+						type: "mrkdwn",
+						text: `*${status}*\n*Task ID:* ${taskId || "Unknown"}\n*Duration:* ${durationText}`,
+					},
+				},
+				{
+					type: "context",
+					elements: [
+						{
+							type: "mrkdwn",
+							text: `Finished at: <!date^${Math.floor(Date.now() / 1000)}^{date_short_pretty} at {time}|${new Date().toISOString()}>`,
+						},
+					],
+				},
+			],
+			thread_ts: threadTs,
+		})
+	}
+}