import crypto from "node:crypto"; import { constants as fsConstants } from "node:fs"; import { access, mkdir, readFile, rm, stat, writeFile } from "node:fs/promises"; import path from "node:path"; import { fileURLToPath, pathToFileURL } from "node:url"; import { AGENT_MGR_SERVICE_ID, AGENT_WORKER_SERVICE_ID, advanceAgentSession, advanceWorkerSession, createAgentSession, createEvidenceRecord, createTraceEvent, createWorkerSession, createWorkspaceVolume } from "./index.mjs"; import { ENVIRONMENT_DEV } from "../protocol/index.mjs"; const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../.."); export const DEFAULT_AGENT_RUNTIME_STATE_DIR = path.join(repoRoot, ".state", "agent-runtime"); export const LOCAL_DRY_RUN_MODE = "local-dry-run"; export function resolveAgentRuntimeStateDir(stateDir = process.env.HWLAB_AGENT_RUNTIME_STATE_DIR) { return path.resolve(stateDir || DEFAULT_AGENT_RUNTIME_STATE_DIR); } export function assessSkillsInjection({ skillCommitId = null, skillVersion = null, role = "runtime" } = {}) { const commitId = explicitValue(skillCommitId); const version = explicitValue(skillVersion); const missing = []; if (!commitId) missing.push("skillCommitId"); if (!version) missing.push("skillVersion"); return { role, status: missing.length === 0 ? "ready" : "degraded", commitId, version, missing, source: "explicit-field" }; } export function createRuntimeHealth({ stateDir, managerSkills = null, workerSkills = null, workspaceCleaned = false } = {}) { const missing = []; const mismatches = []; for (const skills of [managerSkills, workerSkills].filter(Boolean)) { for (const field of skills.missing) { missing.push(`${skills.role}.${field}`); } } if (managerSkills && workerSkills) { if (managerSkills.commitId && workerSkills.commitId && managerSkills.commitId !== workerSkills.commitId) { mismatches.push("skillCommitId"); } if (managerSkills.version && workerSkills.version && managerSkills.version !== workerSkills.version) { mismatches.push("skillVersion"); } } const status = missing.length === 0 && mismatches.length === 0 ? "ok" : "degraded"; return { serviceId: AGENT_MGR_SERVICE_ID, status, environment: ENVIRONMENT_DEV, runtimeMode: LOCAL_DRY_RUN_MODE, stateDir: stateDir ? resolveAgentRuntimeStateDir(stateDir) : undefined, workspaceCleaned, skills: { manager: managerSkills, worker: workerSkills, missing, mismatches } }; } export async function createLocalAgentSession({ stateDir, agentSessionId, projectId, goal = "", prompt = "", requestedBy = "usr_local", skillCommitId = null, skillVersion = null, now = () => new Date().toISOString() }) { requireId(agentSessionId, "agentSessionId"); requireId(projectId, "projectId"); const root = resolveAgentRuntimeStateDir(stateDir); const sessionDir = localAgentSessionDir(root, agentSessionId); if (await pathExists(sessionDir)) { throw new Error(`agent session already exists: ${agentSessionId}`); } const timestamp = now(); const traceId = `trc_${agentSessionId}`; const operationId = `op_${agentSessionId}`; const workerSessionId = `wkr_${agentSessionId}`; const taskId = `tsk_${agentSessionId}`; const workspacePath = path.join(root, "workspaces", agentSessionId); const managerSkills = assessSkillsInjection({ skillCommitId, skillVersion, role: "manager" }); const health = createRuntimeHealth({ stateDir: root, managerSkills }); const workspaceVolume = { ...createWorkspaceVolume({ agentSessionId, projectId, now }), mountPath: workspacePath, localPath: workspacePath }; const agentSession = createAgentSession({ agentSessionId, projectId, requestedBy, goal, now, metadata: { runtimeMode: LOCAL_DRY_RUN_MODE, traceId, operationId, workerSessionId, taskId, prompt, skills: managerSkills } }); const workerSession = createWorkerSession({ workerSessionId, agentSessionId, projectId, now, metadata: { runtimeMode: LOCAL_DRY_RUN_MODE, traceId, operationId, taskId, scopedProjectId: projectId, workspaceVolumeId: workspaceVolume.workspaceVolumeId, expectedSkills: managerSkills } }); const state = { schemaVersion: 1, environment: ENVIRONMENT_DEV, runtimeMode: LOCAL_DRY_RUN_MODE, stateDir: root, sessionDir, createdAt: timestamp, updatedAt: timestamp, health, task: { taskId, traceId, operationId, status: "queued", goal, prompt, createdAt: timestamp, updatedAt: timestamp }, agentSession, workerSession, workspaceVolume, workspace: { workspaceVolumeId: workspaceVolume.workspaceVolumeId, path: workspacePath, cleaned: false, existsAfterCleanup: null }, traceEvents: [], evidenceRecords: [], cleanup: null }; state.traceEvents.push( createRuntimeTraceEvent(state, { suffix: "create", serviceId: AGENT_MGR_SERVICE_ID, message: "Agent manager created local dry-run session", state: "create", now, metadata: { taskId, healthStatus: health.status, skills: managerSkills } }) ); await writeLocalAgentState(root, state); return summarizeLocalAgentSession(state); } export async function readLocalAgentSession({ stateDir, agentSessionId }) { requireId(agentSessionId, "agentSessionId"); const root = resolveAgentRuntimeStateDir(stateDir); return readLocalAgentState(root, agentSessionId); } export async function getLocalAgentStatus({ stateDir, agentSessionId }) { const state = await readLocalAgentSession({ stateDir, agentSessionId }); return summarizeLocalAgentSession(state); } export async function getLocalAgentTrace({ stateDir, agentSessionId }) { const state = await readLocalAgentSession({ stateDir, agentSessionId }); return { serviceId: AGENT_MGR_SERVICE_ID, agentSessionId, traceId: state.task.traceId, status: state.health.status, tracePath: localAgentTracePath(state.sessionDir), traceEvents: state.traceEvents }; } export async function getLocalAgentEvidence({ stateDir, agentSessionId }) { const state = await readLocalAgentSession({ stateDir, agentSessionId }); const evidenceHealth = state.evidenceRecords.some((record) => record.metadata?.health?.status === "degraded") ? "degraded" : state.health.status; return { serviceId: AGENT_MGR_SERVICE_ID, agentSessionId, operationId: state.task.operationId, status: evidenceHealth, evidenceDir: localAgentEvidenceDir(state.sessionDir), evidenceRecords: state.evidenceRecords }; } export async function cleanupLocalAgentSession({ stateDir, agentSessionId, actor = AGENT_MGR_SERVICE_ID, now = () => new Date().toISOString() }) { requireId(agentSessionId, "agentSessionId"); const root = resolveAgentRuntimeStateDir(stateDir); const state = await readLocalAgentState(root, agentSessionId); if (state.cleanup?.completed === true) { return { serviceId: actor, agentSessionId, idempotent: true, cleanup: state.cleanup, workspace: state.workspace, status: state.health.status }; } await cleanupWorkspaceForState(state, { actor, now }); await writeLocalAgentState(root, state); return { serviceId: actor, agentSessionId, idempotent: false, cleanup: state.cleanup, workspace: state.workspace, status: state.health.status }; } export async function runLocalWorkerDryRun({ stateDir, agentSessionId, skillCommitId = null, skillVersion = null, now = () => new Date().toISOString() }) { requireId(agentSessionId, "agentSessionId"); const root = resolveAgentRuntimeStateDir(stateDir); const state = await readLocalAgentState(root, agentSessionId); if (state.cleanup?.completed === true) { throw new Error(`agent session is already cleaned up: ${agentSessionId}`); } const workerSkills = assessSkillsInjection({ skillCommitId, skillVersion, role: "worker" }); const health = createRuntimeHealth({ stateDir: root, managerSkills: state.health.skills.manager, workerSkills }); const timestamp = now(); state.updatedAt = timestamp; state.health = health; state.task = { ...state.task, status: "claimed", claimedAt: timestamp, updatedAt: timestamp }; state.agentSession = advanceAgentSession(state.agentSession, "start", { now, extra: { workerSessionId: state.workerSession.workerSessionId, workspaceVolumeId: state.workspace.workspaceVolumeId, taskStatus: "claimed", healthStatus: health.status } }); state.workerSession = advanceWorkerSession(state.workerSession, "start", { now, extra: { workspaceVolumeId: state.workspace.workspaceVolumeId, workspacePath: state.workspace.path, taskStatus: "claimed", healthStatus: health.status, skills: workerSkills } }); await mkdir(state.workspace.path, { recursive: true }); state.traceEvents.push( createRuntimeTraceEvent(state, { suffix: "start", serviceId: AGENT_WORKER_SERVICE_ID, message: "Agent worker claimed local dry-run task", state: "start", now, metadata: { taskId: state.task.taskId, workspacePath: state.workspace.path, healthStatus: health.status, skills: workerSkills } }) ); const workspaceTracePath = path.join(state.workspace.path, "trace.ndjson"); const workspaceTrace = createRuntimeTraceEvent(state, { suffix: "trace", serviceId: AGENT_WORKER_SERVICE_ID, message: "Agent worker wrote local dry-run trace", state: "trace", now, metadata: { taskId: state.task.taskId, workspaceTracePath, healthStatus: health.status } }); state.agentSession = advanceAgentSession(state.agentSession, "trace", { now, extra: { taskStatus: "running", healthStatus: health.status } }); state.workerSession = advanceWorkerSession(state.workerSession, "trace", { now, extra: { taskStatus: "running", healthStatus: health.status } }); state.traceEvents.push(workspaceTrace); await writeFile(workspaceTracePath, `${JSON.stringify(workspaceTrace)}\n`, "utf8"); const evidencePayload = { schemaVersion: 1, environment: ENVIRONMENT_DEV, runtimeMode: LOCAL_DRY_RUN_MODE, agentSessionId, workerSessionId: state.workerSession.workerSessionId, taskId: state.task.taskId, traceId: state.task.traceId, operationId: state.task.operationId, goal: state.task.goal, prompt: state.task.prompt, health, traceEventIds: state.traceEvents.map((event) => event.traceEventId), cleanup: { workspacePath: state.workspace.path, workspaceWillBeRemoved: true }, createdAt: now() }; const evidenceContent = `${JSON.stringify(evidencePayload, null, 2)}\n`; const evidenceDir = localAgentEvidenceDir(state.sessionDir); const evidenceArtifactPath = path.join(evidenceDir, `worker-dry-run-${agentSessionId}.json`); await mkdir(evidenceDir, { recursive: true }); await writeFile(evidenceArtifactPath, evidenceContent, "utf8"); const evidenceStats = await stat(evidenceArtifactPath); const evidenceRecord = createEvidenceRecord({ evidenceId: `evi_${agentSessionId}`, projectId: state.agentSession.projectId, operationId: state.task.operationId, uri: pathToFileURL(evidenceArtifactPath).href, sha256: sha256Hex(evidenceContent), serviceId: AGENT_WORKER_SERVICE_ID, environment: ENVIRONMENT_DEV, kind: "report", agentSessionId, workerSessionId: state.workerSession.workerSessionId, mimeType: "application/json", sizeBytes: evidenceStats.size, metadata: { runtimeMode: LOCAL_DRY_RUN_MODE, traceId: state.task.traceId, taskId: state.task.taskId, workspaceVolumeId: state.workspace.workspaceVolumeId, health, skills: { manager: state.health.skills.manager, worker: workerSkills } }, now }); state.evidenceRecords.push(evidenceRecord); state.traceEvents.push( createRuntimeTraceEvent(state, { suffix: "finish", serviceId: AGENT_WORKER_SERVICE_ID, message: "Agent worker persisted local dry-run evidence", state: "finish", now, metadata: { taskId: state.task.taskId, evidenceId: evidenceRecord.evidenceId, healthStatus: health.status } }) ); state.task = { ...state.task, status: "completed", completedAt: now(), updatedAt: now() }; state.agentSession = advanceAgentSession(state.agentSession, "finish", { now, extra: { taskStatus: "completed", healthStatus: health.status } }); state.workerSession = advanceWorkerSession(state.workerSession, "finish", { now, extra: { taskStatus: "completed", evidenceId: evidenceRecord.evidenceId, healthStatus: health.status } }); await cleanupWorkspaceForState(state, { actor: AGENT_WORKER_SERVICE_ID, now }); await writeLocalAgentState(root, state); return { serviceId: AGENT_WORKER_SERVICE_ID, dryRun: true, agentSession: state.agentSession, workerSession: state.workerSession, task: state.task, health: state.health, traceEvents: state.traceEvents, evidenceRecord, evidencePath: evidenceArtifactPath, cleanup: state.cleanup, workspace: state.workspace }; } export function summarizeLocalAgentSession(state) { return { serviceId: AGENT_MGR_SERVICE_ID, agentSessionId: state.agentSession.agentSessionId, status: state.health.status, environment: state.environment, runtimeMode: state.runtimeMode, stateDir: state.stateDir, sessionDir: state.sessionDir, health: state.health, task: state.task, agentSession: state.agentSession, workerSession: state.workerSession, workspace: state.workspace, traceCount: state.traceEvents.length, evidenceCount: state.evidenceRecords.length, cleanup: state.cleanup }; } function createRuntimeTraceEvent(state, { suffix, serviceId, message, state: lifecycleState, metadata = {}, now }) { return createTraceEvent({ traceEventId: `tev_${state.agentSession.agentSessionId}_${suffix}`, traceId: state.task.traceId, projectId: state.agentSession.projectId, serviceId, level: metadata.healthStatus === "degraded" ? "warn" : "info", message, environment: ENVIRONMENT_DEV, agentSessionId: state.agentSession.agentSessionId, workerSessionId: state.workerSession.workerSessionId, operationId: state.task.operationId, metadata: { ...metadata, state: lifecycleState }, now }); } async function cleanupWorkspaceForState(state, { actor, now }) { const existedBeforeCleanup = await pathExists(state.workspace.path); await rm(state.workspace.path, { recursive: true, force: true }); const existsAfterCleanup = await pathExists(state.workspace.path); const timestamp = now(); state.cleanup = { cleanupId: `cln_${state.agentSession.agentSessionId}`, serviceId: actor, agentSessionId: state.agentSession.agentSessionId, workerSessionId: state.workerSession.workerSessionId, workspaceVolumeId: state.workspace.workspaceVolumeId, workspacePath: state.workspace.path, completed: true, existedBeforeCleanup, removed: !existsAfterCleanup, existsAfterCleanup, cleanedAt: timestamp }; state.workspace = { ...state.workspace, cleaned: true, existsAfterCleanup }; state.health = { ...state.health, workspaceCleaned: true }; state.agentSession = advanceAgentSession(state.agentSession, "cleanup", { now, extra: { cleanedUp: true, cleanupId: state.cleanup.cleanupId, healthStatus: state.health.status } }); state.workerSession = advanceWorkerSession(state.workerSession, "cleanup", { now, extra: { cleanedUp: true, cleanupId: state.cleanup.cleanupId, healthStatus: state.health.status } }); state.traceEvents.push( createRuntimeTraceEvent(state, { suffix: "cleanup", serviceId: actor, message: "Agent worker cleaned local dry-run workspace", state: "cleanup", now, metadata: { cleanupId: state.cleanup.cleanupId, workspaceVolumeId: state.workspace.workspaceVolumeId, workspacePath: state.workspace.path, removed: state.cleanup.removed, healthStatus: state.health.status } }) ); state.updatedAt = timestamp; } async function readLocalAgentState(root, agentSessionId) { const statePath = localAgentStatePath(localAgentSessionDir(root, agentSessionId)); try { const raw = await readFile(statePath, "utf8"); return JSON.parse(raw); } catch (error) { if (error.code === "ENOENT") { throw new Error(`agent session not found: ${agentSessionId}`); } throw error; } } async function writeLocalAgentState(root, state) { await mkdir(state.sessionDir, { recursive: true }); await mkdir(localAgentEvidenceDir(state.sessionDir), { recursive: true }); await writeFile(localAgentStatePath(state.sessionDir), `${JSON.stringify(state, null, 2)}\n`, "utf8"); const traceContent = state.traceEvents.map((event) => JSON.stringify(event)).join("\n"); await writeFile(localAgentTracePath(state.sessionDir), `${traceContent}${traceContent ? "\n" : ""}`, "utf8"); } function localAgentSessionDir(root, agentSessionId) { return path.join(root, "sessions", agentSessionId); } function localAgentStatePath(sessionDir) { return path.join(sessionDir, "session.json"); } function localAgentTracePath(sessionDir) { return path.join(sessionDir, "trace.ndjson"); } function localAgentEvidenceDir(sessionDir) { return path.join(sessionDir, "evidence"); } async function pathExists(targetPath) { try { await access(targetPath, fsConstants.F_OK); return true; } catch (error) { if (error.code === "ENOENT") return false; throw error; } } function requireId(value, fieldName) { if (!explicitValue(value)) { throw new Error(`${fieldName} is required`); } } function explicitValue(value) { if (typeof value !== "string") return null; const trimmed = value.trim(); return trimmed.length > 0 ? trimmed : null; } function sha256Hex(value) { return crypto.createHash("sha256").update(value).digest("hex"); }