// Pure utility functions for the team orchestrator // Extracted from team-orchestrator.js for testability import { mkdir, readFile, rename, rm, writeFile } from "node:fs/promises" import path from "node:path" export const REGISTRY_VERSION = 1 export const LOCK_TIMEOUT_MS = 5000 export const LOCK_RETRY_MS = 50 export const LOCK_STALE_MS = 30 * 1000 export function now() { return new Date().toISOString() } export function createTaskID() { const random = Math.random().toString(36).slice(2, 8) return `team_${Date.now().toString(36)}_${random}` } export function stateRoot(worktree) { return path.join(worktree, ".opencode", "state") } export function artifactRoot(worktree) { return path.join(stateRoot(worktree), "artifacts") } export function registryFile(worktree) { return path.join(stateRoot(worktree), "tasks.json") } export function lockDirectory(worktree) { return path.join(stateRoot(worktree), "tasks.lock") } export function temporaryRegistryFile(worktree) { return path.join(stateRoot(worktree), `tasks.${process.pid}.tmp`) } export function projectRootFromContext(context) { if (context.worktree && context.worktree !== "/") { return context.worktree } return context.directory } export function isErrno(error, code) { return Boolean(error && typeof error === "object" && "code" in error && error.code === code) } export function errorMessage(error) { if (error instanceof Error) return error.message return String(error) } export function isPidAlive(pid) { try { process.kill(pid, 0) return true } catch (error) { return false } } export async function sleep(ms) { await new Promise(resolve => setTimeout(resolve, ms)) } function lockMetaFile(worktree) { return path.join(lockDirectory(worktree), "meta.json") } export async function acquireLock(worktree) { await ensureState(worktree) const deadline = Date.now() + LOCK_TIMEOUT_MS const lockPath = lockDirectory(worktree) while (Date.now() < deadline) { try { await mkdir(lockPath) const meta = { owner: process.pid, createdAt: Date.now() } await writeFile(lockMetaFile(worktree), JSON.stringify(meta), "utf8") return } catch (error) { if (!isErrno(error, "EEXIST")) throw error await recoverStaleLock(worktree) await sleep(LOCK_RETRY_MS) } } throw new Error("Timed out acquiring the team task registry lock") } export async function recoverStaleLock(worktree) { const lockPath = lockDirectory(worktree) const metaPath = lockMetaFile(worktree) try { const raw = await readFile(metaPath, "utf8") const meta = JSON.parse(raw) if (typeof meta.createdAt === "number" && Date.now() - meta.createdAt < LOCK_STALE_MS) { if (typeof meta.owner === "number" && isPidAlive(meta.owner)) { return } } } catch (error) { if (isErrno(error, "ENOENT")) { // meta file missing — safe to reclaim } else { return // corrupt meta — don't reclaim } } try { await rm(lockPath, { recursive: true, force: true }) } catch (error) { if (!isErrno(error, "ENOENT")) { throw error } } } export async function withLockedRegistry(worktree, mutate) { await acquireLock(worktree) try { const registry = await loadRegistry(worktree) const result = await mutate(registry) await saveRegistry(worktree, registry) return result } finally { await rm(lockDirectory(worktree), { recursive: true, force: true }) } } export async function ensureState(worktree) { await mkdir(artifactRoot(worktree), { recursive: true }) } export async function loadRegistry(worktree) { await ensureState(worktree) const file = registryFile(worktree) try { const raw = await readFile(file, "utf8") const parsed = JSON.parse(raw) if (!Array.isArray(parsed.tasks)) { return { version: REGISTRY_VERSION, tasks: [] } } return parsed } catch (error) { if (error && typeof error === "object" && "code" in error && error.code === "ENOENT") { return { version: REGISTRY_VERSION, tasks: [] } } throw error } } export async function saveRegistry(worktree, registry) { await ensureState(worktree) const file = registryFile(worktree) const temporaryFile = temporaryRegistryFile(worktree) await writeFile( temporaryFile, `${JSON.stringify({ version: REGISTRY_VERSION, tasks: registry.tasks }, null, 2)}\n`, "utf8", ) await rename(temporaryFile, file) } export function upsertTask(registry, task) { const index = registry.tasks.findIndex(existing => existing.id === task.id) if (index === -1) { registry.tasks.push(task) return } registry.tasks[index] = task } export function getTask(registry, taskID) { return registry.tasks.find(task => task.id === taskID) } export async function addTask(worktree, task) { await withLockedRegistry(worktree, registry => { upsertTask(registry, task) }) } export async function readTask(worktree, taskID) { const registry = await loadRegistry(worktree) return getTask(registry, taskID) } export async function updateTask(worktree, taskID, mutate) { return withLockedRegistry(worktree, registry => { const task = getTask(registry, taskID) if (!task) { throw new Error(`Unknown task_id: ${taskID}`) } return mutate(task, registry) }) } export function buildWorkerPrompt(input) { const sections = [ "You are a delegated worker in a multi-agent team orchestration prototype.", "Stay focused on the assigned scope and produce a concise final result that a coordinator can synthesize.", "", `## Task ID\n${input.taskID}`, "", `## Title\n${input.title}`, ] if (input.goal) { sections.push("", `## Goal\n${input.goal}`) } if (input.files.length > 0) { sections.push("", `## Suggested Files\n${input.files.map(file => `- ${file}`).join("\n")}`) } sections.push( "", "## Instructions", input.prompt, "", "## Response Contract", "Return a final response with these sections:", "## Summary", "## Files", "- file paths you changed or inspected", "## Risks", "- remaining risks or unknowns", "## Next Steps", "- any follow-up the coordinator should consider", ) return sections.join("\n") } export function extractTextFromParts(parts) { return parts .filter(part => part.type === "text" && typeof part.text === "string") .map(part => part.text.trim()) .filter(Boolean) .join("\n\n") } export const STATUS_ICONS = { running: "▸", ready: "◦", done: "✓", cancelled: "○", error: "✗", } export function formatTaskLine(task, liveStatus) { const status = liveStatus || task.status const icon = STATUS_ICONS[status] || "•" return `${icon} ${task.id} ${status} ${task.agent} ${task.title}` } export function artifactPathForDisplay(worktree, artifactPath) { if (!artifactPath) return "" return path.relative(worktree, artifactPath) || artifactPath } export async function recordTaskError(worktree, taskID, error) { return updateTask(worktree, taskID, task => { if (task.status === "cancelled") { return { ...task } } task.status = "error" task.updatedAt = now() task.error = errorMessage(error) return { ...task } }) } export async function recordTaskCompletion(worktree, taskID, latest, artifactPath) { return updateTask(worktree, taskID, task => { if (task.status === "cancelled") { return { cancelled: true, task: { ...task }, } } task.status = "done" task.updatedAt = now() task.summary = latest.text || undefined task.artifactPath = artifactPath task.lastMessageID = latest.messageID delete task.error return { cancelled: false, task: { ...task }, } }) }