/**
 * Session continuity for DEVIN_CONNECT.
 *
 * Derives a stable protobuf session_id (#16) from the conversation's own completed
 * request/response pair fingerprints, so one multi-turn dialog reuses ONE session_id
 * instead of minting a fresh uuid per turn (which the upstream velocity limiter reads
 * as N brand-new sessions). Identity is the client-visible input/output only — never
 * requestId / session / IP / timestamp / provider signature.
 *
 * Architecture:
 *   1. Canonicalize OpenAI messages → events (role:kind:payload)
 *   2. Normalize tool linkage (semantic slot matching, not transport IDs)
 *   3. Split into completed pairs: [input] → [output]
 *   4. HMAC-hash each pair by scopeId
 *   5. Resolve by overlap scoring against stored pair windows
 *   6. Commit after response with idempotency guard
 *
 * Edge cases handled:
 *   - Tool call ID regeneration (retry/combo switch)
 *   - Orphan tool_result (barrier → blocks resolve)
 *   - System prompt changes (excluded from identity)
 *   - Parallel dialogs on same API key (different pair chains)
 *   - Output drift (client truncates response text)
 *   - Idempotent commits (same state → same stateId)
 */
import crypto from 'crypto';

// ─── Configuration ─────────────────────────────────────────────────────────

const SECRET = crypto.randomBytes(32);
const DEFAULT_TTL = 30 * 60 * 1000;
const DEFAULT_MAX_STATES = 500;
const PAIR_WINDOW_SIZE = 10;
const CLEANUP_INTERVAL = 5 * 60 * 1000;

function getTtl(env = process.env) {
  const v = Number(env.DEVIN_CONNECT_SESSION_TTL_MS);
  return Number.isFinite(v) && v > 0 ? v : DEFAULT_TTL;
}

function getMaxStates(env = process.env) {
  const v = Number(env.DEVIN_CONNECT_SESSION_MAX_STATES);
  return Number.isFinite(v) && v > 0 ? v : DEFAULT_MAX_STATES;
}

// ─── State Store ───────────────────────────────────────────────────────────

const statesById = new Map();
const pairIndex = new Map();       // `${scopeId}:${pairHash}` → Set<stateId>
const commitIndex = new Map();     // commitKey → stateId
// stateId → every key this state wrote into pairIndex/commitIndex, so eviction
// touches its own keys instead of scanning the (unbounded) whole index.
const ownedIndexKeys = new Map();

export function isSessionReuseEnabled(env = process.env) {
  const v = String(env.DEVIN_CONNECT_SESSION_REUSE ?? '').trim().toLowerCase();
  return v === '1' || v === 'true' || v === 'yes' || v === 'on';
}

// ─── Canonicalizer ─────────────────────────────────────────────────────────

function deriveScopeId(callerKey) {
  return crypto.createHmac('sha256', SECRET).update(callerKey || '').digest('hex');
}

function messageText(msg) {
  if (typeof msg?.content === 'string') return msg.content;
  if (Array.isArray(msg?.content)) return msg.content.map((p) => p?.text || '').join('');
  return '';
}

function canonicalizeArgs(args) {
  if (typeof args === 'string') {
    try { return JSON.stringify(JSON.parse(args)); } catch { return args; }
  }
  if (args && typeof args === 'object') return JSON.stringify(args);
  return '';
}

/**
 * Canonicalize OpenAI messages into a normalized event stream.
 * System/developer messages are EXCLUDED (they don't participate in pair identity).
 */
export function canonicalize(messages) {
  const events = [];
  for (const msg of (messages || [])) {
    if (!msg) continue;
    const role = msg.role || '';
    if (role === 'system' || role === 'developer') continue;

    if (role === 'tool') {
      events.push({
        role: 'tool',
        kind: 'tool_result',
        payload: { rawLink: msg.tool_call_id || '', content: messageText(msg) },
      });
      continue;
    }

    const text = messageText(msg);
    if (text) events.push({ role, kind: 'text', payload: text });

    if (Array.isArray(msg.tool_calls)) {
      for (const tc of msg.tool_calls) {
        const name = tc.function?.name || tc.name || '';
        const args = canonicalizeArgs(tc.function?.arguments ?? tc.arguments);
        events.push({
          role,
          kind: 'tool_call',
          payload: { name, arguments: args, rawLink: tc.id || '' },
        });
      }
    }

    // Anchor empty assistant turns (no text, no tool_calls) so they still form a pair boundary
    if (role === 'assistant' && !text && !Array.isArray(msg.tool_calls)) {
      events.push({ role: 'assistant', kind: 'empty_output', payload: '' });
    }
  }
  return events;
}

// ─── Tool Linkage Normalizer ───────────────────────────────────────────────

function isAssistantOutput(ev) {
  return ev.role === 'assistant' && (ev.kind === 'text' || ev.kind === 'tool_call' || ev.kind === 'empty_output');
}

/**
 * Normalize tool_call ↔ tool_result linkage by semantic content (name + args + slot),
 * not by opaque transport IDs. Survives tool_call_id regeneration across retries.
 */
export function normalizeToolLinkage(events) {
  if (!Array.isArray(events)) return [];
  const rawLinkToDescriptor = new Map();
  let inRun = false;
  let batch = -1;
  let batchSlot = 0;
  let lastBatchDescriptors = [];

  return events.map((ev) => {
    if (isAssistantOutput(ev)) {
      if (!inRun) { inRun = true; batch++; batchSlot = 0; lastBatchDescriptors = []; }
      if (ev.kind !== 'tool_call') return ev;

      const p = ev.payload || {};
      const descriptor = { slot: `slot:${batchSlot++}`, name: p.name || '', arguments: p.arguments || '', pending: true };
      if (p.rawLink) rawLinkToDescriptor.set(p.rawLink, descriptor);
      lastBatchDescriptors.push(descriptor);
      return { ...ev, payload: { slot: descriptor.slot, name: descriptor.name, arguments: descriptor.arguments } };
    }

    inRun = false;
    if (ev.role === 'tool' && ev.kind === 'tool_result') {
      const p = ev.payload || {};
      const rawLink = p.rawLink || '';
      // Try raw link first
      let desc = rawLink ? rawLinkToDescriptor.get(rawLink) : null;
      if (desc && !desc.pending) desc = null; // already consumed
      // Fallback: name-hint (single pending match by name)
      if (!desc && p.nameHint) {
        const matches = lastBatchDescriptors.filter((d) => d.pending && d.name === p.nameHint);
        if (matches.length === 1) desc = matches[0];
      }
      if (!desc) {
        return { ...ev, payload: { unlinked: true, content: p.content || '' } };
      }
      // Consume
      desc.pending = false;
      if (rawLink) rawLinkToDescriptor.delete(rawLink);
      return { ...ev, payload: { slot: desc.slot, name: desc.name, arguments: desc.arguments, content: p.content || '' } };
    }

    return ev;
  });
}

// ─── Barrier Detection ─────────────────────────────────────────────────────

function hasUnlinkedResult(events) {
  return events.some((ev) => ev.role === 'tool' && ev.kind === 'tool_result' && ev.payload?.unlinked === true);
}

/**
 * Analyze history: detect barriers, compute post-barrier pair chain.
 */
export function analyzeHistory(events, scopeId) {
  const normalized = normalizeToolLinkage(events);
  const { records, trailingInput } = splitWithTrailing(normalized);

  const hasTrailingBarrier = hasUnlinkedResult(trailingInput);

  let lastBarrierIndex = -1;
  for (let i = 0; i < records.length; i++) {
    if (hasUnlinkedResult(records[i].input)) lastBarrierIndex = i;
  }

  const postBarrierRecords = hasTrailingBarrier ? [] : records.slice(lastBarrierIndex + 1);
  const hashes = postBarrierRecords.map((r) => hashPair(r, scopeId));
  const canResolve = !hasTrailingBarrier && hashes.length > 0;

  return { records, trailingInput, hasTrailingBarrier, lastBarrierIndex, postBarrierRecords, hashes, canResolve };
}

function splitWithTrailing(events) {
  const pairs = [];
  let input = [];
  let output = [];
  let inOutput = false;

  for (const ev of events) {
    if (isAssistantOutput(ev)) {
      inOutput = true;
      output.push(ev);
    } else {
      if (inOutput) {
        pairs.push({ input, output });
        input = [];
        output = [];
        inOutput = false;
      }
      input.push(ev);
    }
  }
  if (inOutput && output.length > 0) {
    pairs.push({ input, output });
    return { records: pairs, trailingInput: [] };
  }
  return { records: pairs, trailingInput: input };
}

// ─── Pair Hashing ──────────────────────────────────────────────────────────

function stableJson(value) {
  if (value === undefined) return '';
  if (value === null) return 'null';
  if (Array.isArray(value)) return '[' + value.map(stableJson).join(',') + ']';
  if (typeof value === 'object') {
    const keys = Object.keys(value).sort();
    return '{' + keys.map((k) => JSON.stringify(k) + ':' + stableJson(value[k])).join(',') + '}';
  }
  return JSON.stringify(value);
}

function hashPair(pair, scopeId) {
  const obj = { version: 'pair-v1', scopeId, input: pair.input, output: pair.output };
  return crypto.createHmac('sha256', SECRET).update(stableJson(obj)).digest('hex');
}

/**
 * Derive a per-dialog "root anchor" key from the dialog's FIRST input turn — the
 * only identifying signal available before any completed pair exists. This is what
 * lets the session_id be stable from turn 1: the same first input yields the same
 * key at resolve time (turn 1, trailing input) and at commit time (turn 1, first
 * pair's input) and on every later turn (first pair's input). Two dialogs with a
 * byte-identical opener share this key until they diverge — an unavoidable ceiling
 * (there is no other signal at turn 1) — and the pair chain then forks them at
 * commit via the provisional-claim rule. Returns null when even the opener is empty.
 */
function rootAnchorKey(scopeId, analysis) {
  const firstPairInput = analysis.records?.[0]?.input;
  const firstInput = (Array.isArray(firstPairInput) && firstPairInput.length)
    ? firstPairInput
    : analysis.trailingInput;
  if (!Array.isArray(firstInput) || firstInput.length === 0) return null;
  const digest = crypto.createHmac('sha256', SECRET)
    .update(stableJson({ version: 'root-v1', scopeId, input: firstInput }))
    .digest('hex');
  return `${scopeId}:root:${digest}`;
}

/**
 * Build completed pair hashes for a conversation.
 */
export function buildPairHashes(callerKey, messages) {
  const scopeId = deriveScopeId(callerKey);
  const events = canonicalize(messages);
  const analysis = analyzeHistory(events, scopeId);
  return { scopeId, hashes: analysis.hashes, canResolve: analysis.canResolve };
}

// ─── Resolver ──────────────────────────────────────────────────────────────

/**
 * Overlap score: the contiguous run of candidate-window hashes ending at the
 * candidate's LAST hash (a retained tail is always a SUFFIX of the committed
 * chain — compaction cuts the head, never the tail). Anchoring at the tail is
 * what keeps a divergent dialog that merely shares an early pair from
 * hijacking the session: a prefix-only run scores 0 here.
 */
function overlapScore(incoming, candidateWindow) {
  if (!candidateWindow.length || !incoming.length) return 0;
  const lastCand = candidateWindow[candidateWindow.length - 1];
  let anchor = -1;
  for (let i = incoming.length - 1; i >= 0; i--) {
    if (incoming[i] === lastCand) { anchor = i; break; }
  }
  if (anchor < 0) return 0;
  let score = 0;
  for (let k = 0; k < candidateWindow.length && anchor - k >= 0; k++) {
    if (incoming[anchor - k] !== candidateWindow[candidateWindow.length - 1 - k]) break;
    score++;
  }
  return score;
}

/**
 * Output drift tolerance: when exact hash doesn't match but the input is
 * identical and the output is a prefix/superset of the stored output.
 */
function driftScore(incomingRecords, candidateRecords, incomingHashes, candidateHashes) {
  if (!Array.isArray(incomingRecords) || !Array.isArray(candidateRecords)) return 0;
  const max = Math.min(10, incomingRecords.length, candidateRecords.length);
  let score = 0;
  for (let k = 1; k <= max; k++) {
    const iIdx = incomingRecords.length - k;
    const cIdx = candidateRecords.length - k;
    // Exact hash match takes priority
    if (incomingHashes?.[incomingHashes.length - k] === candidateHashes?.[candidateHashes.length - k]) {
      score = k;
      continue;
    }
    const inc = incomingRecords[iIdx];
    const cand = candidateRecords[cIdx];
    if (stableJson(inc?.input) !== stableJson(cand?.input)) break;
    // Output: check if one is a prefix of the other (text truncation/extension)
    if (!outputsCompatible(inc?.output, cand?.output)) break;
    score = k;
  }
  return score;
}

function outputsCompatible(a, b) {
  if (!Array.isArray(a) || !Array.isArray(b) || a.length !== b.length) return false;
  for (let i = 0; i < a.length; i++) {
    if (a[i]?.role !== b[i]?.role || a[i]?.kind !== b[i]?.kind) return false;
    const aText = typeof a[i]?.payload === 'string' ? a[i].payload : null;
    const bText = typeof b[i]?.payload === 'string' ? b[i].payload : null;
    if (aText !== null || bText !== null) {
      if (aText === null || bText === null) return false;
      if (aText === bText) continue;
      const shorter = aText.length <= bText.length ? aText : bText;
      const longer = aText.length > bText.length ? aText : bText;
      if (shorter.length < 24 || shorter.length / Math.max(1, longer.length) < 0.35) return false;
      if (!longer.includes(shorter)) return false;
      continue;
    }
    if (stableJson(a[i]?.payload) !== stableJson(b[i]?.payload)) return false;
  }
  return true;
}

// ─── Store Operations ──────────────────────────────────────────────────────

function evictState(stateId) {
  // Only dead-state membership is redundant: historical live hashes still gate
  // drift lookup, and old commit keys still implement historical idempotency.
  // Walk the state's own key set instead of the whole index: the index still
  // grows with live history, so a full scan makes every eviction O(turns) and it
  // is paid by the sweep and by create-at-capacity (measured 3.4 ms per evict at
  // 100k index entries before this change).
  const drop = (key) => {
    const set = pairIndex.get(key);
    if (set) { set.delete(stateId); if (set.size === 0) pairIndex.delete(key); }
    if (commitIndex.get(key) === stateId) commitIndex.delete(key);
  };
  const state = statesById.get(stateId);
  const owned = ownedIndexKeys.get(stateId);
  if (owned) { for (const key of owned) drop(key); ownedIndexKeys.delete(stateId); }
  // Belt and braces for the paths that index a state without going through
  // indexState: the re-association branch writes the new window's keys directly,
  // the root anchor is indexed at creation, and commitKey is written besides the
  // tracked commit keys. Deriving them from the state itself keeps eviction O(own)
  // without depending on every writer remembering to call ownIndexKey.
  if (state) {
    for (const hash of state.pairWindow || []) drop(`${state.scopeId}:${hash}`);
    if (state.commitKey) drop(state.commitKey);
    if (state.rootKey) drop(state.rootKey);
  }
  statesById.delete(stateId);
}

function enforceCapacity(env) {
  const max = getMaxStates(env);
  if (statesById.size < max) return;
  let oldest = null; let oldestTs = Infinity;
  for (const [id, s] of statesById) {
    if (s.lastSeen < oldestTs) { oldestTs = s.lastSeen; oldest = id; }
  }
  if (oldest) evictState(oldest);
}

function clearExpired(env) {
  const ttl = getTtl(env);
  const now = Date.now();
  for (const [id, s] of statesById) {
    if (now - s.lastSeen > ttl) evictState(id);
  }
}

function ownIndexKey(stateId, key) {
  let owned = ownedIndexKeys.get(stateId);
  if (!owned) { owned = new Set(); ownedIndexKeys.set(stateId, owned); }
  owned.add(key);
}

// Every pairIndex insertion goes through here. A key that is written without an
// owner outlives the state that wrote it: once the state's rolling window moves
// past it, nothing can find the membership again and eviction leaves it behind
// (2026-09-17 review, F4: the resolve re-association branch indexed its new
// window directly). Ownership is what makes eviction able to drop exactly the
// state's own keys — membership in a key another live state still holds stays.
function linkPairIndex(stateId, key) {
  let set = pairIndex.get(key);
  if (!set) { set = new Set(); pairIndex.set(key, set); }
  set.add(stateId);
  ownIndexKey(stateId, key);
}

function indexState(stateId, state) {
  for (const ph of state.pairWindow) linkPairIndex(stateId, `${state.scopeId}:${ph}`);
}

// ─── Public API ────────────────────────────────────────────────────────────

/**
 * Resolve a stable session_id for the given conversation.
 * Called BEFORE upstream dispatch.
 */
// ─── ModelConfig stability (companion to PR #226) ──────────────────────────
// Genuine devin.exe keeps ModelConfig #15.1 stable within a session and
// increments #15.2 every turn (calibrated on live capture 9501aa2c). With
// SESSION_REUSE on this gateway is no longer stateless either, so both values
// can be derived from the same session state. Opt-in gate, default OFF.

export function isModelConfigStableEnabled(env = process.env) {
  const v = String(env.DEVIN_CONNECT_MODEL_CONFIG_STABLE ?? '').trim().toLowerCase();
  return v === '1' || v === 'true' || v === 'yes' || v === 'on';
}

// ─── T1: server-side reasoning continuity (Thinking-core design v0.3) ──────
// The reasoning we already emit on outbound is kept per session as a queue
// of tail digests and re-injected as a system-prompt suffix on the next turn.
// Text channel only (A3 matrix: #11/#9 accepted but not consumed by upstream);
// never as an assistant message (self-reflection-loop anti-pattern).

// Total injection budget AND per-digest cap. Default 4000; 0 disables capture.
// Ceiling-clamped (same post-merge lesson as #241's DIGEST_MAX_CEILING): `1e9`
// passes isFinite, and an uncapped digest here would let one 30KB reasoning turn
// ride the system prompt whole — exactly the unbounded-body failure mode.
const SESSION_REASONING_MAX_CHARS_CEILING = 32000;
const SESSION_REASONING_COUNT_CEILING = 32;
export function getSessionReasoningMaxChars(env = process.env) {
  // '' and 0 < n < 1 fall back to the default: Number('') === 0 would read an
  // empty assignment as opt-out, and Math.floor(0.5) === 0 would silently turn
  // the feature off — count's rule applied to chars: only a literal 0 opts out.
  const raw = String(env.DEVIN_CONNECT_SESSION_REASONING_MAX_CHARS ?? '').trim();
  if (raw === '') return 4000;
  const n = Number(raw);
  if (!Number.isFinite(n) || n < 0 || (n > 0 && n < 1)) return 4000;
  if (n === 0) return 0;
  return Math.min(Math.floor(n), SESSION_REASONING_MAX_CHARS_CEILING);
}

// Queue length in turns (how many per-turn digests the state keeps). Default 5.
// n < 1 — including a fractional value like 0.5, which Math.floor would turn
// into a silent 0 — falls back to the default: count=0 is not a meaningful
// setting ("keep how many turns"), so unlike chars=0 it is not an opt-out.
export function getSessionReasoningCount(env = process.env) {
  const n = Number(env.DEVIN_CONNECT_SESSION_REASONING_COUNT);
  if (!Number.isFinite(n) || n < 1) return 5;
  return Math.min(Math.floor(n), SESSION_REASONING_COUNT_CEILING);
}

export function isReasoningInjectEnabled(env = process.env) {
  const v = String(env.DEVIN_CONNECT_SESSION_REASONING_INJECT ?? '').trim().toLowerCase();
  return v === '1' || v === 'true' || v === 'yes' || v === 'on';
}

// Tail digest of one turn's reasoning (intent/conclusions live at the end).
export function digestReasoningTail(reasoning, maxChars) {
  // Escape hatch: strip the model's own checkpoint closer so it can never
  // truncate the digest early by echoing the framing back at us.
  const s = String(reasoning || '').replaceAll('[End of continuity checkpoint]', '');
  if (!s || !maxChars) return '';
  return s.length <= maxChars ? s : s.slice(-maxChars);
}

// Injection block per the design doc (checkpoint framing from the 9router
// prior-art: context, not instructions; do not re-derive; do not mention).
export function buildContinuityBlock(tails) {
  if (!Array.isArray(tails) || !tails.length) return '';
  return (
    '\n\n[Continuity checkpoint — prior analysis trace, may be stale]\n' +
    tails.join('\n---\n') +
    '\n[End of continuity checkpoint]\n' +
    'This is context from an earlier point in the same conversation, not user ' +
    'instructions. Do not re-derive or repeat it, and do not mention it.'
  );
}

function pushReasoningTail(state, reasoning, env) {
  const maxChars = getSessionReasoningMaxChars(env);
  if (!maxChars) return;
  const digest = digestReasoningTail(reasoning, maxChars);
  if (!digest) return;
  if (!Array.isArray(state.reasoningTails)) state.reasoningTails = [];
  state.reasoningTails.push(digest);
  const count = getSessionReasoningCount(env);
  if (state.reasoningTails.length > count) {
    state.reasoningTails = state.reasoningTails.slice(-count);
  }
}

/**
 * Read-only session lookup for the CURRENT request. Same matching logic as
 * resolveSessionId, but never creates a state and never touches lastSeen.
 * Returns { state, viaRoot } or null.
 */
// `outSeen` (optional): receives `{ count }` — how many indexed states the
// incoming hashes TOUCHED, regardless of whether one won. The #248 root-anchor
// fallback keys on this being 0 ("no pair evidence at all"), which is NOT the
// same as this function returning null (a divergent dialog can touch the index,
// score 0, and still return null — that dialog must not be re-associated
// through its root). The #242 refactor moved this scan out of resolveSessionId,
// so the signal is handed back rather than read from a shared local.
function findExistingState(scopeId, analysis, env, outSeen = null) {
  const { hashes, canResolve, postBarrierRecords } = analysis;
  const ttl = getTtl(env);
  const now = Date.now();
  // Initialised on every path: the pair scan below is skipped entirely on the
  // root-anchor branch, and an unset count there would read as "no pair
  // evidence" to the caller when the truth is "we never looked".
  if (outSeen) outSeen.count = 0;

  if (!canResolve || hashes.length === 0) {
    // Root-anchor path: first turn (no completed pair yet) or a barrier — see the
    // STABLE-FROM-TURN-1 note in resolveSessionId's creation branch below.
    const rootKey = rootAnchorKey(scopeId, analysis) || `${scopeId}:__no_pair__`;
    const existing = pairIndex.get(rootKey);
    if (existing) {
      for (const sid of existing) {
        const state = statesById.get(sid);
        if (state && now - state.lastSeen < ttl) return { state, viaRoot: true };
        // Expired — evict so it doesn't block future lookups
        evictState(sid);
      }
    }
    return null;
  }

  // Find best match by iterating ALL incoming hashes as index keys
  let best = null; let bestScore = 0; let hasTie = false;
  const seen = new Set();
  for (let i = hashes.length - 1; i >= 0; i--) {
    const idxKey = `${scopeId}:${hashes[i]}`;
    const candidates = pairIndex.get(idxKey);
    if (!candidates) continue;
    for (const stateId of candidates) {
      if (seen.has(stateId)) continue;
      seen.add(stateId);
      const state = statesById.get(stateId);
      if (!state || now - state.lastSeen > ttl) continue;
      let score = overlapScore(hashes, state.pairWindow);
      // Drift fallback
      if (score === 0 && state.pairRecords?.length) {
        score = driftScore(postBarrierRecords, state.pairRecords, hashes, state.pairWindow);
      }
      if (score > bestScore) { bestScore = score; best = state; hasTie = false; }
      else if (score === bestScore && score > 0) hasTie = true;
    }
  }
  // `seenCount` rides on the RESULT rather than replacing the null contract:
  // three callers test `if (!found)` / `if (found)`, so returning an object on
  // the miss path would make every miss look like a hit. The #248 root fallback
  // needs the miss-path count too, so it is exposed via the out-param below.
  if (outSeen) outSeen.count = seen.size;
  return (best && !hasTie) ? { state: best, viaRoot: false } : null;
}

/**
 * ModelConfig for the current request: stable per-session config UUID (#15.1)
 * and 1-based turn counter (#15.2), mirroring genuine devin.exe wire behaviour.
 * Read-only — never creates state, never touches lastSeen. Null when either
 * gate is off or no session resolves for this request.
 */
export function getSessionModelConfig(callerKey, messages, env = process.env) {
  if (!isModelConfigStableEnabled(env) || !isSessionReuseEnabled(env)) return null;
  const scopeId = deriveScopeId(callerKey);
  const analysis = analyzeHistory(canonicalize(messages), scopeId);
  const found = findExistingState(scopeId, analysis, env);
  if (!found) return null;
  return { configId: found.state.configId, turn: (found.state.turnCount || 0) + 1 };
}

/**
 * T1 read side: the continuity block to append to the system prompt for the
 * current request, or null when nothing applies (gate off, no session, empty
 * queue). Budget: whole tail digests from the newest, while they fit into
 * DEVIN_CONNECT_SESSION_REASONING_MAX_CHARS total. Read-only like
 * getSessionModelConfig.
 */
export function getSessionReasoningTrail(callerKey, messages, env = process.env) {
  if (!isReasoningInjectEnabled(env) || !isSessionReuseEnabled(env)) return null;
  const budget = getSessionReasoningMaxChars(env);
  if (!budget) return null;
  const scopeId = deriveScopeId(callerKey);
  const analysis = analyzeHistory(canonicalize(messages), scopeId);
  const found = findExistingState(scopeId, analysis, env);
  if (!found) return null;
  const tails = found.state.reasoningTails;
  if (!Array.isArray(tails) || !tails.length) return null;
  const picked = [];
  let total = 0;
  for (let i = tails.length - 1; i >= 0; i--) {
    if (total + tails[i].length > budget) break;
    picked.unshift(tails[i]);
    total += tails[i].length;
  }
  if (!picked.length) return null;
  return buildContinuityBlock(picked);
}

export function resolveSessionId(callerKey, messages, env = process.env) {
  if (!isSessionReuseEnabled(env)) return null;

  const scopeId = deriveScopeId(callerKey);
  const events = canonicalize(messages);
  const analysis = analyzeHistory(events, scopeId);
  const { hashes, canResolve, postBarrierRecords } = analysis;
  const now = Date.now();

  const ttl = getTtl(env);
  const seenProbe = { count: 0 };
  const found = findExistingState(scopeId, analysis, env, seenProbe);
  if (found) {
    found.state.lastSeen = now;
    if (!found.viaRoot) {
      const newWindow = hashes.slice(-PAIR_WINDOW_SIZE);
      // Re-index with new hashes (stateId stored on state object for O(1) access)
      if (found.state.stateId) {
        for (const ph of newWindow) linkPairIndex(found.state.stateId, `${scopeId}:${ph}`);
      }
      found.state.pairWindow = newWindow;
      found.state.pairRecords = postBarrierRecords.slice(-PAIR_WINDOW_SIZE);
    }
    return found.state.sessionId;
  }

  if (!canResolve || hashes.length === 0) {
    // First turn (no completed pair yet) or a barrier. Anchor the session on the
    // dialog's first input turn so the id is STABLE FROM TURN 1 — the only signal
    // that exists before a pair does. Falls back to a shared bucket only when even
    // the opener is empty. Two dialogs with an identical opener share this id until
    // they diverge; commitAfterResponse then forks them (provisional-claim rule).
    const rootKey = rootAnchorKey(scopeId, analysis) || `${scopeId}:__no_pair__`;
    enforceCapacity(env);
    const sessionId = crypto.randomUUID();
    const stateId = crypto.randomUUID();
    const state = { stateId, scopeId, sessionId, pairWindow: [], pairRecords: [], lastSeen: now, commitKey: null, dialogAnchor: null, rootKey, configId: crypto.randomUUID(), turnCount: 0, reasoningTails: [] };
    statesById.set(stateId, state);
    linkPairIndex(stateId, rootKey);
    return sessionId;
  }

  // Root fallback (compaction survival). Measured on a real kimi compaction:
  // 0 of 31 retained pairs survive byte-for-byte or canonically — the client
  // rewrites the retained tail — BUT the dialog's first input turn survives
  // verbatim. States are indexed by that rootKey at creation, so re-associate
  // through it when pair evidence is gone. Several live candidates with no pair
  // evidence = ambiguous -> assign to none (collision rule), let a new id form.
  // Guard: only when NO incoming hash matched any stored index (seen empty) — a
  // divergent dialog whose prefix pair still hits the index must NOT be
  // re-associated through the root (pair evidence exists; it just scores 0).
  if (seenProbe.count === 0) {
  const rootKey = rootAnchorKey(scopeId, analysis);
  if (rootKey) {
    const rootSet = pairIndex.get(rootKey);
    if (rootSet) {
      const live = [];
      for (const stateId of rootSet) {
        const state = statesById.get(stateId);
        if (!state) continue;
        if (now - state.lastSeen > ttl) { evictState(stateId); continue; }
        live.push(state);
      }
      if (live.length === 1) {
        const state = live[0];
        state.lastSeen = now;
        const newWindow = hashes.slice(-PAIR_WINDOW_SIZE);
        state.pairWindow = newWindow;
        indexState(state.stateId, state);
        state.pairRecords = postBarrierRecords.slice(-PAIR_WINDOW_SIZE);
        return state.sessionId;
      }
    }
  }
  }

  // No match — create new state
  clearExpired(env);
  enforceCapacity(env);
  const sessionId = crypto.randomUUID();
  const stateId = crypto.randomUUID();
  const pairWindow = hashes.slice(-PAIR_WINDOW_SIZE);
  const dialogAnchor = pairWindow[0]?.slice(0, 16) || null;
  const state = { stateId, scopeId, sessionId, pairWindow, pairRecords: postBarrierRecords.slice(-PAIR_WINDOW_SIZE), lastSeen: now, commitKey: null, dialogAnchor, configId: crypto.randomUUID(), turnCount: 0, reasoningTails: [] };
  statesById.set(stateId, state);
  indexState(stateId, state);
  return sessionId;
}

/**
 * Commit state after a successful response. Idempotent.
 * Called AFTER upstream response completes.
 */
export function commitAfterResponse(callerKey, messagesWithResponse, env = process.env, opts = {}) {
  if (!isSessionReuseEnabled(env)) return null;

  const scopeId = deriveScopeId(callerKey);
  const events = canonicalize(messagesWithResponse);
  const analysis = analyzeHistory(events, scopeId);
  const { hashes, postBarrierRecords } = analysis;
  if (hashes.length === 0) return null;

  const pairWindow = hashes.slice(-PAIR_WINDOW_SIZE);
  const ttl = getTtl(env);
  const now = Date.now();

  // Compute commitKey for idempotency
  const commitKey = crypto.createHmac('sha256', SECRET)
    .update(JSON.stringify({ scopeId, pairWindow }))
    .digest('hex');

  const existingId = commitIndex.get(commitKey);
  if (existingId) {
    const existing = statesById.get(existingId);
    if (existing && now - existing.lastSeen < ttl) {
      existing.lastSeen = now;
      return existing.sessionId;
    }
  }

  // Find existing state to update (by overlap)
  let target = null;
  const seen = new Set();
  for (let i = hashes.length - 1; i >= 0; i--) {
    const idxKey = `${scopeId}:${hashes[i]}`;
    const candidates = pairIndex.get(idxKey);
    if (!candidates) continue;
    for (const stateId of candidates) {
      if (seen.has(stateId)) continue;
      seen.add(stateId);
      const state = statesById.get(stateId);
      if (!state || now - state.lastSeen > ttl) continue;
      const score = overlapScore(hashes, state.pairWindow);
      if (score > 0) { target = { state, stateId }; break; }
    }
    if (target) break;
  }

  if (target) {
    // Update existing state
    target.state.lastSeen = now;
    target.state.pairWindow = pairWindow;
    target.state.pairRecords = postBarrierRecords.slice(-PAIR_WINDOW_SIZE);
    target.state.commitKey = commitKey;
    target.state.turnCount = (target.state.turnCount || 0) + 1;
    pushReasoningTail(target.state, opts.reasoning, env);
    commitIndex.set(commitKey, target.stateId);
    ownIndexKey(target.stateId, commitKey);
    indexState(target.stateId, target.state);
    return target.state.sessionId;
  }

  // Turn-1 commit: no pair matched yet, but resolveSessionId already minted a
  // provisional state under this dialog's root anchor. Claim it so the id chosen
  // on turn 1 carries into the pair chain (stable from turn 1). Only an UNCLAIMED
  // provisional (empty pairWindow) is reusable — a second dialog with the same
  // opener finds this one already claimed and forks into a fresh state below.
  const rootKey = rootAnchorKey(scopeId, analysis);
  if (rootKey) {
    const rootCandidates = pairIndex.get(rootKey);
    if (rootCandidates) {
      for (const sid of rootCandidates) {
        const state = statesById.get(sid);
        if (!state || now - state.lastSeen > ttl) continue;
        if (state.pairWindow.length > 0) continue; // already claimed by another dialog
        state.lastSeen = now;
        state.pairWindow = pairWindow;
        state.pairRecords = postBarrierRecords.slice(-PAIR_WINDOW_SIZE);
        state.commitKey = commitKey;
        state.turnCount = (state.turnCount || 0) + 1;
        pushReasoningTail(state, opts.reasoning, env);
        if (!state.dialogAnchor) state.dialogAnchor = pairWindow[0]?.slice(0, 16) || null;
        commitIndex.set(commitKey, sid);
        ownIndexKey(sid, commitKey);
        indexState(sid, state);
        return state.sessionId;
      }
    }
  }

  // Create new state (shouldn't normally happen if resolve was called first)
  clearExpired(env);
  enforceCapacity(env);
  const sessionId = crypto.randomUUID();
  const stateId = crypto.randomUUID();
  const dialogAnchor = pairWindow[0]?.slice(0, 16) || null;
  // Root-index the fork too: a forked dialog shares its opener's root anchor, so
  // a later compacted resolve must see it as a SECOND live root candidate —
  // otherwise the root fallback would re-associate the fork into the ORIGINAL
  // session (hijack). The fallback's ambiguity rule (several live -> assign to
  // none) only works when every root-sharing state is visible under the anchor.
  const state = { stateId, scopeId, sessionId, pairWindow, pairRecords: postBarrierRecords.slice(-PAIR_WINDOW_SIZE), lastSeen: now, commitKey, dialogAnchor, rootKey, configId: crypto.randomUUID(), turnCount: 1, reasoningTails: [] };
  pushReasoningTail(state, opts.reasoning, env);
  statesById.set(stateId, state);
  commitIndex.set(commitKey, stateId);
  ownIndexKey(stateId, commitKey);
  indexState(stateId, state);
  if (rootKey) linkPairIndex(stateId, rootKey);
  return sessionId;
}

// ─── Periodic Cleanup ──────────────────────────────────────────────────────

const _cleanupInterval = setInterval(() => clearExpired(), CLEANUP_INTERVAL);
if (_cleanupInterval.unref) _cleanupInterval.unref();

// ─── Test Helpers ──────────────────────────────────────────────────────────

export function _resetForTests() {
  statesById.clear();
  pairIndex.clear();
  commitIndex.clear();
  ownedIndexKeys.clear();
}

export function _getStoreSize() {
  return statesById.size;
}

// Index health, for the eviction invariant: no index key may still reference a
// state that eviction dropped (a membership like that can never be removed
// again, because the state that owns it is gone from statesById), and a key two
// live states share must keep the surviving one.
export function _getIndexStats() {
  let pairKeys = 0; let pairMemberships = 0; let sharedKeys = 0; let deadMemberships = 0;
  for (const set of pairIndex.values()) {
    pairKeys++;
    pairMemberships += set.size;
    if (set.size > 1) sharedKeys++;
    for (const id of set) if (!statesById.has(id)) deadMemberships++;
  }
  let deadCommitKeys = 0;
  for (const id of commitIndex.values()) if (!statesById.has(id)) deadCommitKeys++;
  return { pairKeys, pairMemberships, sharedKeys, deadMemberships, commitKeys: commitIndex.size, deadCommitKeys, ownedStates: ownedIndexKeys.size };
}