pikasTech-HWLAB/internal/agent/runtime.mjs

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");
}