#!/usr/bin/env node

import os from "node:os";
import path from "node:path";
import { readFile } from "node:fs/promises";
import { fileURLToPath } from "node:url";

import { executeCapability } from "./capabilities.mjs";
import { assertProfile } from "./profile-policy.mjs";
import { createHostedSession, runHostedOnce } from "./hosted-runner.mjs";

const APP_ROOT = path.dirname(path.dirname(fileURLToPath(import.meta.url)));
const REPO_ROOT = APP_ROOT;

export async function runOnce(options = {}) {
  const bridgeUrl = new URL(options.bridgeUrl || "http://127.0.0.1:8787");
  if (options.hosted === true || !isLoopback(bridgeUrl.hostname)) {
    const result = await runHostedOnce({
      ...options,
      bridgeUrl,
      session: options.session || createHostedSession({ token: options.deviceSession, expiresAt: options.deviceSessionExpiresAt }),
      workspaceRoot: path.resolve(options.workspaceRoot || os.homedir()),
      homeRoot: path.resolve(options.homeRoot || os.homedir()),
    });
    return result.job;
  }
  const runnerId = options.runnerId || `local-${process.platform}-${process.arch}`;
  const deviceId = options.deviceId || "device-local";
  const stateDir = path.resolve(options.stateDir || path.join(APP_ROOT, ".data"));
  const runnerToken = options.runnerToken || (await readFile(path.join(stateDir, "runner-token"), "utf8")).trim();
  const enrolledProfile = assertProfile(options.profile || "monitoring_only");
  const job = (await runnerRequest(bridgeUrl, runnerToken, "/api/v1/runner/claim", {
    runnerId,
    deviceId,
    leaseSeconds: options.leaseSeconds || 900,
  })).job;
  if (!job) return null;

  assertExecutableJob(job, deviceId, enrolledProfile);
  await runnerRequest(bridgeUrl, runnerToken, "/api/v1/runner/start", {
    jobId: job.jobId,
    leaseId: job.lease.leaseId,
    runnerId,
  });

  const pairing = {
    // Execute against the immutable profile already bound into the job digest.
    // The exact match above proves that it is also the local enrollment value.
    profile: job.spec.profile,
    workspaceRoot: path.resolve(options.workspaceRoot || REPO_ROOT),
    homeRoot: path.resolve(options.homeRoot || os.homedir()),
    npmFeedFile: options.npmFeedFile ? path.resolve(options.npmFeedFile) : undefined,
    consents: job.spec.kind === "action" ? { [job.spec.capabilityId]: true } : {},
  };
  let result;
  try {
    result = job.spec.kind === "assessment"
      ? await executeAssessment(job, pairing, options.signal)
      : await executeCapability({ capabilityId: job.spec.capabilityId, parameters: job.spec.parameters, pairing, signal: options.signal });
  } catch (error) {
    const failure = {
      type: "talons.device-runner.failure.v1",
      code: error?.code || "capability_failed",
      message: error instanceof Error ? error.message : String(error),
    };
    await runnerRequest(bridgeUrl, runnerToken, "/api/v1/runner/finish", {
      jobId: job.jobId,
      leaseId: job.lease.leaseId,
      runnerId,
      outcome: options.signal?.aborted ? "cancelled" : "failed",
      result: failure,
    });
    return { ...job, status: options.signal?.aborted ? "cancelled" : "failed", result: failure };
  }

  // Settlement is deliberately outside the capability catch. A transport or
  // state error after the capability returned must never be rewritten as a
  // capability failure (or, worse, as proof that an already-applied mutation
  // was cancelled). The store records the actual completed outcome even when
  // a cancellation request raced too late to stop this MVP runner.
  return (await runnerRequest(bridgeUrl, runnerToken, "/api/v1/runner/finish", {
    jobId: job.jobId,
    leaseId: job.lease.leaseId,
    runnerId,
    outcome: "succeeded",
    result,
  })).job;
}

async function executeAssessment(job, pairing, signal) {
  const results = [];
  for (const capabilityId of job.spec.capabilityIds) {
    results.push(await executeCapability({ capabilityId, parameters: {}, pairing, signal }));
  }
  return {
    type: "talons.device-assessment-result.v1",
    profile: job.spec.profile,
    results,
  };
}

function assertExecutableJob(job, deviceId, enrolledProfile) {
  if (!job || job.status !== "leased" || !job.lease?.leaseId || !job.spec?.profile
    || job.spec.deviceId !== deviceId || job.lease.deviceId !== deviceId) {
    throw new Error("Runner received a malformed or non-leased enrolled-profile job.");
  }
  if (job.spec.profile !== enrolledProfile) {
    throw new Error(`Job profile ${job.spec.profile} does not match locally enrolled profile ${enrolledProfile}.`);
  }
  if (job.spec.kind === "assessment") {
    if (job.spec.operationClass !== "read_only" || !Array.isArray(job.spec.capabilityIds)) {
      throw new Error("Runner received an invalid assessment specification.");
    }
    return;
  }
  const receipt = job.consent?.receipt;
  if (job.spec.kind !== "action" || job.spec.operationClass !== "consent_required"
    || !receipt || receipt.jobId !== job.jobId || receipt.jobSpecDigest !== job.specDigest) {
    throw new Error("Runner received an action without an exact consent receipt.");
  }
}

async function runnerRequest(baseUrl, runnerToken, pathname, body) {
  const response = await fetch(new URL(pathname, baseUrl), {
    method: "POST",
    headers: { authorization: `Bearer ${runnerToken}`, "content-type": "application/json" },
    body: JSON.stringify(body),
  });
  const payload = await response.json();
  if (!response.ok) {
    const error = new Error(payload.message || `Runner API returned HTTP ${response.status}.`);
    error.code = payload.error;
    throw error;
  }
  return payload;
}

function isLoopback(hostname) {
  return hostname === "127.0.0.1" || hostname === "[::1]" || hostname === "::1" || hostname === "localhost";
}

async function main() {
  const once = process.argv.includes("--once");
  const hosted = process.env.TALONS_RUNNER_MODE === "hosted";
  const bridgeUrl = new URL(
    hosted
      ? process.env.TALONS_HOSTED_BRIDGE_URL || process.env.TALONS_BRIDGE_URL || "http://127.0.0.1:8788"
      : process.env.TALONS_BRIDGE_URL || "http://127.0.0.1:8787",
  );
  let hostedSession = null;
  if (hosted || !isLoopback(bridgeUrl.hostname)) {
    if (process.env.TALONS_RUNNER_SESSION_STDIN !== "1") {
      throw new Error("Hosted runner requires TALONS_RUNNER_SESSION_STDIN=1.");
    }
    hostedSession = await readHostedSessionFromStdin();
  }
  do {
    let job;
    if (hostedSession) {
      const result = await runHostedOnce({
        bridgeUrl,
        session: hostedSession,
        deviceId: process.env.TALONS_DEVICE_ID,
        workspaceRoot: process.env.TALONS_WORKSPACE_ROOT || os.homedir(),
        homeRoot: process.env.TALONS_HOME_ROOT || os.homedir(),
        npmFeedFile: process.env.TALONS_NPM_FEED_FILE,
        profile: process.env.TALONS_ENROLLED_PROFILE,
      });
      job = result.job;
      hostedSession = result.session;
    } else job = await runOnce({
      bridgeUrl,
      stateDir: process.env.TALONS_STATE_DIR,
      runnerId: process.env.TALONS_RUNNER_ID,
      deviceId: process.env.TALONS_DEVICE_ID,
      workspaceRoot: process.env.TALONS_WORKSPACE_ROOT,
      homeRoot: process.env.TALONS_HOME_ROOT,
      npmFeedFile: process.env.TALONS_NPM_FEED_FILE,
      profile: process.env.TALONS_ENROLLED_PROFILE,
    });
    if (job) process.stdout.write(`${JSON.stringify({ jobId: job.jobId, status: job.status })}\n`);
    if (once) break;
    await new Promise((resolve) => setTimeout(resolve, job ? 250 : 2_000));
  } while (true);
}

async function readHostedSessionFromStdin() {
  let bytes = "";
  process.stdin.setEncoding("utf8");
  for await (const chunk of process.stdin) {
    bytes += chunk;
    if (bytes.length > 8_192) throw new Error("Hosted runner bootstrap exceeds 8 KiB.");
    const newline = bytes.indexOf("\n");
    if (newline >= 0) {
      const value = JSON.parse(bytes.slice(0, newline));
      if (value?.type !== "talons.runner-session.v1") {
        throw new Error("Hosted runner received an unsupported session handoff.");
      }
      return createHostedSession({ token: value.deviceSession, expiresAt: value.expiresAt });
    }
  }
  throw new Error("Hosted runner requires one private device-session bootstrap line on stdin.");
}

if (process.argv[1] && path.resolve(process.argv[1]) === path.resolve(fileURLToPath(import.meta.url))) {
  main().catch((error) => {
    process.stderr.write(`${error instanceof Error ? error.stack : String(error)}\n`);
    process.exitCode = 1;
  });
}
