/*
<MODULE_CONTRACT>
<purpose>
RFC-1133: shared command-result cache block for the kernel executor. Owns the
`command_results` read/write consumed by `executeRegisteredCommand` — the single
choke point covering direct `executeKernelCommand` calls and both pipeline
executors — plus the flagsHash rule (CACHE_INVISIBLE_FLAGS denylist), the
RFC-0685 mtime fast path, and the RFC-1130 cacheBypassFlags gate.
</purpose>
<non-goals>
  <item>Do not implement cache storage — that lives in kernel/cache/cache-layer.ts and sqlite-cache-layer.ts.</item>
  <item>Do not implement pipeline scheduling or step orchestration — that lives in execute-pipeline.ts.</item>
  <item>Do not persist cross-pipeline cache hits — persistCacheHits stays in execute-pipeline.ts.</item>
</non-goals>
</MODULE_CONTRACT>
<KEY_DECISIONS>
  <item>Cache read/write runs inside executeRegisteredCommand — the single choke point; executors pass per-run state via context.resultCache.</item>
  <item>flagsHash denylist (CACHE_INVISIBLE_FLAGS) over resolved flags — presentation/bypass universals never fragment the cache.</item>
</KEY_DECISIONS>
<CHANGE_SUMMARY>
  <item>RFC-1133: module created — isCommandCacheable, hasCacheBypassFlag, tryMtimeFastPath, and the read/write orchestrators relocated from execute-pipeline.ts; computeFlagsHash + CACHE_INVISIBLE_FLAGS added; key gains flagsHash (schema v2).</item>
  <item>RFC-1133: cache direct executeKernelCommand executions with flag-keyed results</item>
  <item>RFC-1193: bound module hashing to src/os roots with constant fallbacks</item>
  <item>RFC-1193: review findings (discriminated union, dead condition, fixture)</item>
  <item>RFC-1303: review REVIEW-CODE-2026-10-16-02 — scanner moves to werkstatt-shared/node (IO-01), workpieceEnv hashes into envDigest, saveCache tolerates KERNEL-META-01, ENV-INPUT-02 coverage notice, ratchet path-arg tightening</item>
</CHANGE_SUMMARY>
*/

import { createHash } from "node:crypto";
import { dirname, join, relative, sep } from "node:path";

import { byteHash, stableJsonHash } from "@warpgogol/werkstatt-engine/fingerprint";
import type {
  KernelCommandDefinition,
  KernelExecutionReport,
  KernelResultCacheContext,
  KernelRuntimeContext,
} from "@warpgogol/werkstatt-shared/kernel";

import { createCacheLayer } from "../cache/cache-layer.ts";
import {
  COMMAND_RESULT_CACHE_SCHEMA_VERSION,
  computeInputsHash,
  computeModuleHash,
  getCachedCommandResult,
  setCachedCommandResult,
  type CommandResultCacheKey,
  type InputsMetadataEntry,
} from "../cache/command-result-cache.ts";
import { filterTreeIndex } from "../cache/workspace-tree-index.ts";
import { deriveModuleBasePath } from "./registry.ts";
import { createDefaultIO, type WorkspaceIO } from "@warpgogol/werkstatt-shared/kernel/workspace-io";

// @ai-invariant: Command-result caching lives only in executeRegisteredCommand — never read or write command_results from pipeline executors directly.

/**
 * RFC-1133: universal flags that never enter flagsHash. Presentation flags
 * (json/quiet/verbose/help) must not fragment the cache; bypass flags
 * (dry-run/force) are handled by the read/write gates, and `force` in
 * particular must stay out so a --force write refreshes the entry that
 * normal runs read (DNA-93). `root` is a locator, not a result input.
 *
 * `site`/`all` are deliberately NOT invisible: workspace-scoped commands carry
 * siteName: null in the key, so a --site value that survived alias resolution
 * must be keyed.
 */
export const CACHE_INVISIBLE_FLAGS: ReadonlySet<string> = new Set([
  "json",
  "quiet",
  "verbose",
  "help",
  "dry-run",
  "force",
  "root",
]);

/**
 * RFC-1133: stable hash over the cache-visible resolved flags.
 * Returns "" when no visible flags are present.
 */
export function computeFlagsHash(flags: Record<string, unknown>): string {
  const visible: Record<string, unknown> = {};
  for (const [name, value] of Object.entries(flags)) {
    if (!CACHE_INVISIBLE_FLAGS.has(name)) visible[name] = value;
  }
  return Object.keys(visible).length === 0 ? "" : stableJsonHash(visible);
}

/**
 * RFC-1303: stable digest over the command's declared `envInputs` —
 * `name=sha256(value||"")` pairs, sorted. Raw values are never hashed into
 * the persisted key material (only their per-name digests). Empty string
 * when the command declares no env inputs, keeping pre-RFC-1303 cache keys
 * valid for unaffected commands.
 */
export function computeEnvDigest(
  envInputs: string[] | undefined,
  workpieceEnv?: ReadonlyMap<string, string>,
): string {
  if (!envInputs || envInputs.length === 0) return "";
  const pairs = [...envInputs].sort().map(
    (name) =>
      `${name}=${createHash("sha256")
        // RFC-1126 resolution order: process.env wins, then the workpiece
        // .env channel — hash the effective value so a .env-only change
        // invalidates the cache like a process.env change (REVIEW-RFC-1303-02).
        .update(process.env[name] ?? workpieceEnv?.get(name) ?? "")
        .digest("hex")}`,
  );
  return stableJsonHash(pairs);
}

/** RFC-1303: mix the env digest into the base inputs hash. */
function withEnv(inputsHash: string, envDigest: string): string {
  return envDigest ? stableJsonHash({ inputsHash, envDigest }) : inputsHash;
}

/** RFC-1303: metaKey suffix — each env state owns its sidecar mapping. */
function metaKeyEnvSuffix(envDigest: string): string {
  return envDigest ? `:${envDigest}` : "";
}

/**
 * RFC-0390: a command is eligible for caching when `cacheable !== false` AND
 * it declares non-empty `reads`. Commands with `cacheable: false` or without
 * `reads` are always executed.
 */
export function isCommandCacheable(command: KernelCommandDefinition): boolean {
  if (command.cacheable === false) return false;
  const reads = command.reads ?? [];
  return reads.length > 0;
}

/**
 * RFC-1130: true when argv contains a flag listed in the command's
 * `cacheBypassFlags` (`--name` or `--name=value` forms). Flags that reduce
 * coverage (e.g. `--rfc`) must bypass both the cache read and the cache write
 * so a filtered run can neither serve a stale full-run report nor poison the
 * full-run entry.
 */
export function hasCacheBypassFlag(command: KernelCommandDefinition, argv: string[]): boolean {
  const names = command.cacheBypassFlags ?? [];
  return names.some((n) => argv.some((a) => a === `--${n}` || a.startsWith(`--${n}=`)));
}

/**
 * Everything the cache block needs for one invocation, computed once by
 * executeRegisteredCommand from the final resolved input — post
 * resolveSiteFlagAlias, post pipeline --site/--system injection — so read and
 * write always see identical flags.
 */
export interface CommandResultCacheCall {
  command: KernelCommandDefinition;
  /** Raw argv (pre-resolution) — used only for the cacheBypassFlags check. */
  argv: string[];
  /** Resolved input.flags — hashed via computeFlagsHash. */
  flags: Record<string, unknown>;
  baseDir: string;
  workspaceRoot: string;
  siteName: string | null;
  /** Workpiece .env channel (RFC-1126) — hashed into envDigest alongside process.env. */
  workpieceEnv?: ReadonlyMap<string, string>;
  force: boolean;
  dryRun: boolean;
}

export interface ResolvedResultCache {
  cache: KernelResultCacheContext;
  /** True when this call opened the layer — the caller must close() it. */
  owned: boolean;
}

/**
 * Resolve the cache carrier for one command execution. Returns null when the
 * invocation can neither read nor write the cache (non-cacheable command,
 * dry-run, or a cacheBypassFlags hit) — in that case no layer is opened at
 * all. `--force` still resolves: the read is skipped but the write refreshes
 * the entry (DNA-93).
 */
export async function resolveResultCache(
  context: KernelRuntimeContext,
  call: CommandResultCacheCall,
): Promise<ResolvedResultCache | null> {
  if (!isCommandCacheable(call.command)) return null;
  if (call.dryRun) return null;
  if (hasCacheBypassFlag(call.command, call.argv)) return null;
  if (context.resultCache) return { cache: context.resultCache, owned: false };
  const layer = await createCacheLayer(context.workspaceRoot);
  return { cache: { layer, moduleHashCache: new Map() }, owned: true };
}

/**
 * RFC-0390 + RFC-0685 + RFC-1133: attempt to read a cached result.
 * Returns the cached report (with `cached: true`) or null on miss.
 * Skips when `dryRun` or `force` is set or when the layer is unavailable.
 */
export async function readCommandResult(
  cache: KernelResultCacheContext,
  call: CommandResultCacheCall,
  io: WorkspaceIO = createDefaultIO().io,
): Promise<KernelExecutionReport | null> {
  if (call.dryRun || call.force) return null;
  if (!isCommandCacheable(call.command)) return null;
  if (hasCacheBypassFlag(call.command, call.argv)) return null;
  if (!cache.layer.available) return null;

  const reads = call.command.reads ?? [];
  const flagsHash = computeFlagsHash(call.flags);
  const envDigest = computeEnvDigest(call.command.envInputs, call.workpieceEnv);
  const moduleResolved = await resolveModuleSrcDir(call.command, call.workspaceRoot, io);
  const moduleHashCache = cache.moduleHashCache ?? new Map<string, string>();

  // RFC-0685: mtime fast path — requires the pipeline's tree index.
  // RFC-1193: no fast path when moduleSrcDir fell back — the hash is a constant.
  if (cache.treeIndex && moduleResolved.srcDir) {
    const fastPathResult = await tryMtimeFastPath(
      cache,
      call.command,
      reads,
      call.baseDir,
      call.workspaceRoot,
      call.siteName,
      moduleResolved.srcDir,
      moduleHashCache,
      flagsHash,
      envDigest,
    );
    if (fastPathResult) return fastPathResult;
  }

  const { hash: baseInputsHash } = await computeInputsHash(
    reads,
    call.baseDir,
    call.workspaceRoot,
    cache.treeIndex,
  );
  const inputsHash = withEnv(baseInputsHash, envDigest);
  const moduleHash = await getOrComputeModuleHash(
    moduleResolved,
    call.command,
    moduleHashCache,
    call.workspaceRoot,
    io,
  );

  const key: CommandResultCacheKey = {
    schemaVersion: COMMAND_RESULT_CACHE_SCHEMA_VERSION,
    commandName: call.command.name,
    siteName: call.siteName,
    inputsHash,
    moduleHash,
    flagsHash,
  };

  const entry = await getCachedCommandResult(cache.layer, key);
  if (!entry?.report) return null;

  // RFC-1057: verify output files still exist before trusting cache.
  const writes = call.command.writes ?? [];
  if (writes.length > 0) {
    for (const writePath of writes) {
      const abs = join(call.baseDir, writePath);
      if (!(await io.exists(abs))) {
        return null;
      }
    }
  }

  return entry.report;
}

/**
 * RFC-0390 + RFC-0685 + RFC-1133: store a successful command result.
 * Only stores when `ok: true`, not `dryRun`, and the command is cacheable.
 * On `--force`, still stores (refreshing entries). The `command_results_meta`
 * sidecar format is unchanged — inputsHash is flag-independent, so flag
 * variants share the entry benignly.
 */
export async function writeCommandResult(
  cache: KernelResultCacheContext,
  call: CommandResultCacheCall,
  report: KernelExecutionReport,
): Promise<void> {
  if (call.dryRun) return;
  if (!report.ok) return;
  if (!isCommandCacheable(call.command)) return;
  if (hasCacheBypassFlag(call.command, call.argv)) return;
  if (!cache.layer.available) return;

  const reads = call.command.reads ?? [];
  const flagsHash = computeFlagsHash(call.flags);
  const envDigest = computeEnvDigest(call.command.envInputs, call.workpieceEnv);
  const { hash: baseInputsHash, metadata } = await computeInputsHash(
    reads,
    call.baseDir,
    call.workspaceRoot,
    cache.treeIndex,
  );
  const inputsHash = withEnv(baseInputsHash, envDigest);
  const moduleResolved = await resolveModuleSrcDir(call.command, call.workspaceRoot);
  const moduleHash = await getOrComputeModuleHash(
    moduleResolved,
    call.command,
    cache.moduleHashCache ?? new Map<string, string>(),
    call.workspaceRoot,
  );

  const key: CommandResultCacheKey = {
    schemaVersion: COMMAND_RESULT_CACHE_SCHEMA_VERSION,
    commandName: call.command.name,
    siteName: call.siteName,
    inputsHash,
    moduleHash,
    flagsHash,
  };

  await setCachedCommandResult(cache.layer, key, report, metadata);

  // RFC-0685: store metadata-to-inputsHash mapping for the mtime fast path.
  // RFC-1303: envDigest suffixes the sidecar key so alternating env states
  // keep their own mapping instead of overwriting each other.
  if (metadata.length > 0) {
    const metadataHash = stableJsonHash({ metadata });
    const metaKey = `meta:${call.command.name}:${call.siteName ?? ""}:${metadataHash}${metaKeyEnvSuffix(envDigest)}`;
    await cache.layer.set("command_results_meta", metaKey, inputsHash, Date.now(), inputsHash);
  }
}

/**
 * RFC-1193: bounded module-source resolution. A real directory is returned
 * only when `modulePath` carries a `/src/` or `/os/` segment and resolves to an
 * existing directory strictly below the workspace root. Every other outcome is
 * a deterministic constant-hash fallback — caching degrades gracefully instead
 * of fingerprinting unbounded trees (previously the whole workspace or an
 * arbitrary package src).
 */
export type ModuleHashFallbackReason = "unpatched" | "workspace-root" | "missing" | "unrooted";

export type ResolvedModuleSrcDir =
  | { srcDir: string; fallbackReason?: never }
  | { srcDir?: never; fallbackReason: ModuleHashFallbackReason };

/**
 * RFC-1028 + RFC-1193: resolve the module source dir from command.modulePath.
 * Returns a descriptor, never `workspaceRoot`: `unpatched` when modulePath is
 * absent, `unrooted` when modulePath has neither `/src/` nor `/os/`,
 * `workspace-root`/`missing` when the derived path degenerates or is absent.
 */
export async function resolveModuleSrcDir(
  command: KernelCommandDefinition,
  workspaceRoot: string,
  io: WorkspaceIO = createDefaultIO().io,
): Promise<ResolvedModuleSrcDir> {
  if (!command.modulePath) return { fallbackReason: "unpatched" };
  const basePath = deriveModuleBasePath(command.modulePath);
  if (!basePath) return { fallbackReason: "unrooted" };
  const srcDir = join(workspaceRoot, basePath);
  if (srcDir === workspaceRoot) return { fallbackReason: "workspace-root" };
  if (!(await io.exists(srcDir))) return { fallbackReason: "missing" };
  return { srcDir };
}

/**
 * RFC-0637 + RFC-1193: resolve the module hash from the per-run cache,
 * computing on miss. Unresolvable moduleSrcDir yields a deterministic
 * `module-hash-fallback:<reason>` constant (plus a warning) — never a
 * whole-workspace fingerprint.
 */
export async function getOrComputeModuleHash(
  resolved: ResolvedModuleSrcDir,
  command: KernelCommandDefinition,
  moduleHashCache: Map<string, string>,
  workspaceRoot: string,
  io: WorkspaceIO = createDefaultIO().io,
): Promise<string> {
  // RFC-1194: declared modulePaths anchor to dirname(modulePath) — the module
  // file's own directory — independent of the /src//os/ module root. A command
  // in an unrooted location can still declare modulePaths and get real
  // invalidation; only modulePath-less commands stay on the fallbacks.
  const anchorBase =
    (command.modulePaths?.length ?? 0) > 0 && command.modulePath
      ? dirname(join(workspaceRoot, command.modulePath))
      : resolved.srcDir;
  const moduleHashCacheKey = `${anchorBase ?? `fallback:${resolved.fallbackReason}`}:${command.modulePaths?.join(",") ?? ""}`;
  let moduleHash = moduleHashCache.get(moduleHashCacheKey);
  if (!moduleHash) {
    if (anchorBase && (await io.exists(anchorBase))) {
      moduleHash = await computeModuleHash(anchorBase, command.modulePaths, workspaceRoot, io);
    } else {
      const reason = anchorBase ? "missing" : (resolved.fallbackReason ?? "unpatched");
      moduleHash = byteHash(`module-hash-fallback:${reason}`);
      console.warn(
        `Command "${command.name}": module hash base unresolved (${reason}) — using constant module hash fallback (RFC-1193/RFC-1194).`,
      );
    }
    moduleHashCache.set(moduleHashCacheKey, moduleHash);
  }
  return moduleHash;
}

/**
 * RFC-0685: mtime fast path. Computes current file metadata from the tree
 * index, looks up the stored inputsHash via the command_results_meta sidecar,
 * and returns the cached report without fingerprinting when it matches.
 */
async function tryMtimeFastPath(
  cache: KernelResultCacheContext,
  command: KernelCommandDefinition,
  reads: string[],
  baseDir: string,
  workspaceRoot: string,
  siteName: string | null,
  moduleSrcDir: string,
  moduleHashCache: Map<string, string>,
  flagsHash: string,
  envDigest: string,
  io: WorkspaceIO = createDefaultIO().io,
): Promise<KernelExecutionReport | null> {
  const treeIndex = cache.treeIndex;
  if (!treeIndex) return null;

  // Expand globs using tree index to get current file list with metadata.
  const files = filterTreeIndex(treeIndex, reads, baseDir, workspaceRoot);
  if (files.length === 0) return null;

  // Build current metadata from tree index.
  const currentMetadata: InputsMetadataEntry[] = [];
  for (const abs of files) {
    const rel = relative(workspaceRoot, abs).split(sep).join("/");
    const entry = treeIndex.get(rel);
    if (!entry) return null; // file not in index — can't use fast path
    currentMetadata.push({ path: rel, mtimeMs: entry.mtimeMs, size: entry.size });
  }
  currentMetadata.sort((a, b) => a.path.localeCompare(b.path));

  // Compute a hash from the metadata to use as a lookup key.
  const metadataHash = stableJsonHash({ metadata: currentMetadata });

  // Look up the metadata-to-inputsHash mapping.
  const metaKey = `meta:${command.name}:${siteName ?? ""}:${metadataHash}${metaKeyEnvSuffix(envDigest)}`;
  const metaEntry = await cache.layer.get("command_results_meta", metaKey);
  if (!metaEntry) return null;

  const storedInputsHash = metaEntry.data as string;
  if (typeof storedInputsHash !== "string") return null;

  const moduleHash = await getOrComputeModuleHash(
    { srcDir: moduleSrcDir },
    command,
    moduleHashCache,
    workspaceRoot,
    io,
  );

  const key: CommandResultCacheKey = {
    schemaVersion: COMMAND_RESULT_CACHE_SCHEMA_VERSION,
    commandName: command.name,
    siteName,
    inputsHash: storedInputsHash,
    moduleHash,
    flagsHash,
  };

  const entry = await getCachedCommandResult(cache.layer, key);
  if (!entry?.report) return null;

  // RFC-1057: verify output files still exist before trusting cache.
  const writes = command.writes ?? [];
  if (writes.length > 0) {
    for (const writePath of writes) {
      const abs = join(baseDir, writePath);
      if (!(await io.exists(abs))) {
        return null;
      }
    }
  }

  return entry.report;
}