memind/session-reconcile.mjs

import path from 'node:path';
import { developerToolsFromPolicy } from './capabilities.mjs';
import { applyContextBudgetToHarnessEntries } from './context-budget.mjs';
import { mindSpaceMcpScopedTokenNeedsRotation } from './mindspace-mcp-scoped-token.mjs';
import { buildSessionMemoryEntries } from './user-memory-profile.mjs';
import { buildSandboxSessionConstraints } from './user-publish.mjs';

export const SESSION_MCP_ENV_STALE = 'SESSION_MCP_ENV_STALE';
export const MCP_SCOPED_TOKEN_ROTATE_AFTER_MS = 50 * 60 * 1000;

export function parseGooseTimestampMs(value) {
  if (value == null || value === '') return null;
  if (typeof value === 'number' && Number.isFinite(value)) {
    return value < 1e12 ? value * 1000 : value;
  }
  const text = String(value).trim();
  if (!text) return null;
  if (/^\d+$/.test(text)) {
    const n = Number(text);
    if (!Number.isFinite(n) || n <= 0) return null;
    return n < 1e12 ? n * 1000 : n;
  }
  const parsed = Date.parse(text);
  return Number.isFinite(parsed) ? parsed : null;
}

function desiredExtensionsWantMcpToken(desiredExtensions) {
  return (desiredExtensions ?? []).some(
    (extension) => String(extension?.envs?.MINDSPACE_MCP_SCOPED_TOKEN ?? '').trim(),
  );
}

export function sessionMcpEnvRequiresFreshSession(
  session,
  currentExtensions,
  desiredExtensions,
  { now = Date.now() } = {},
) {
  if (!desiredExtensionsWantMcpToken(desiredExtensions)) return false;

  for (const desired of desiredExtensions ?? []) {
    const desiredToken = String(desired?.envs?.MINDSPACE_MCP_SCOPED_TOKEN ?? '').trim();
    if (!desiredToken) continue;
    const name = extensionName(desired);
    const current = (currentExtensions ?? []).find((item) => extensionName(item) === name);
    const currentToken = String(current?.envs?.MINDSPACE_MCP_SCOPED_TOKEN ?? '').trim();
    if (currentToken && mindSpaceMcpScopedTokenNeedsRotation(currentToken, { now })) {
      return true;
    }
  }

  const createdAt = parseGooseTimestampMs(
    session?.created_at ?? session?.createdAt ?? session?.created,
  );
  if (createdAt == null) return false;
  return Number(now) - createdAt >= MCP_SCOPED_TOKEN_ROTATE_AFTER_MS;
}

function staleMcpEnvError(sessionId) {
  const err = new Error(
    `session ${sessionId} MindSpace MCP scoped token cannot be refreshed in-place`,
  );
  err.code = SESSION_MCP_ENV_STALE;
  err.retryable = false;
  return err;
}

function extensionName(config) {
  return config?.name ?? null;
}

export function allowedExtensionNames(extensionOverrides) {
  if (!extensionOverrides) return null;
  return new Set(extensionOverrides.map((item) => item.name).filter(Boolean));
}

export function extensionToolsKey(config) {
  const tools = config?.available_tools ?? config?.availableTools ?? [];
  return [...tools].sort().join('\0');
}

function extensionExecutionConfig(config) {
  const type = String(config?.type ?? '');
  const executionConfig = {
    name: extensionName(config),
    type,
    availableTools: extensionToolsKey(config),
  };

  // Platform/builtin extensions are identified by their name and tool grant.
  // Stdio extensions also carry the executable, arguments, and environment that
  // define their filesystem and service boundaries. Ignoring those fields lets
  // a resumed session keep a stale SANDBOX_ROOT after the Portal runtime moves.
  if (type === 'stdio') {
    executionConfig.cmd = String(config?.cmd ?? '');
    executionConfig.args = Array.isArray(config?.args)
      ? config.args.map((item) => String(item))
      : [];
    executionConfig.envs = Object.fromEntries(
      Object.entries(config?.envs ?? {})
        .sort(([left], [right]) => left.localeCompare(right))
        .map(([key, value]) => [key, String(value)]),
    );
  }

  return executionConfig;
}

export function extensionConfigsMatch(sessionExt, desiredConfig) {
  return (
    JSON.stringify(extensionExecutionConfig(sessionExt))
    === JSON.stringify(extensionExecutionConfig(desiredConfig))
  );
}

export function extensionsNeedingRefresh(currentExtensions, desiredExtensions) {
  const current = currentExtensions ?? [];
  const desired = desiredExtensions ?? [];
  const toRemove = [];
  const toAdd = [];

  for (const ext of current) {
    const name = extensionName(ext);
    if (!name) continue;
    const wanted = desired.find((item) => extensionName(item) === name);
    if (!wanted) {
      toRemove.push(name);
      continue;
    }
    if (!extensionConfigsMatch(ext, wanted)) {
      const currentExec = extensionExecutionConfig(ext);
      const wantedExec = extensionExecutionConfig(wanted);
      const stdioCmdArgsMatch = currentExec.type === 'stdio'
        && wantedExec.type === 'stdio'
        && currentExec.cmd === wantedExec.cmd
        && JSON.stringify(currentExec.args) === JSON.stringify(wantedExec.args);
      if (stdioCmdArgsMatch) {
        // goosed cannot hot-refresh stdio MCP env/tool grants after session start.
        continue;
      }
      toRemove.push(name);
      toAdd.push(wanted);
    }
  }

  for (const config of desired) {
    const name = extensionName(config);
    if (!name) continue;
    const exists = current.some((ext) => extensionName(ext) === name);
    if (!exists) {
      toAdd.push(config);
    }
  }

  return { toRemove, toAdd };
}

function stdioListingOmissionAllowed(currentExt, desiredConfig) {
  if (String(desiredConfig?.type ?? '') !== 'stdio') return false;
  if (!currentExt) return false;
  const currentExec = extensionExecutionConfig(currentExt);
  const desiredExec = extensionExecutionConfig(desiredConfig);
  return currentExec.type === 'stdio'
    && desiredExec.type === 'stdio'
    && currentExec.cmd === desiredExec.cmd
    && JSON.stringify(currentExec.args) === JSON.stringify(desiredExec.args);
}

export function extensionPolicyViolations(currentExtensions, desiredExtensions) {
  const current = currentExtensions ?? [];
  const desired = desiredExtensions ?? [];
  const desiredByName = new Map();
  for (const config of desired) {
    const name = extensionName(config);
    if (name && !desiredByName.has(name)) desiredByName.set(name, config);
  }

  const unexpected = [];
  const duplicate = [];
  const currentByName = new Map();
  for (const ext of current) {
    const name = extensionName(ext);
    if (!name) continue;
    if (!desiredByName.has(name)) unexpected.push(name);
    const configs = currentByName.get(name) ?? [];
    configs.push(ext);
    currentByName.set(name, configs);
    if (configs.length === 2) duplicate.push(name);
  }

  const missingOrMismatched = [];
  for (const [name, desiredConfig] of desiredByName) {
    const configs = currentByName.get(name) ?? [];
    if (stdioListingOmissionAllowed(configs[0] ?? null, desiredConfig)) {
      continue;
    }
    if (
      configs.length !== 1
      || !extensionConfigsMatch(configs[0], desiredConfig)
    ) {
      missingOrMismatched.push(name);
    }
  }

  return {
    unexpected: [...new Set(unexpected)].sort(),
    duplicate: [...new Set(duplicate)].sort(),
    missingOrMismatched: [...new Set(missingOrMismatched)].sort(),
  };
}

function assertExtensionPolicyApplied(currentExtensions, desiredExtensions, sessionId) {
  const violations = extensionPolicyViolations(currentExtensions, desiredExtensions);
  if (
    violations.unexpected.length === 0
    && violations.duplicate.length === 0
    && violations.missingOrMismatched.length === 0
  ) {
    return;
  }

  const err = new Error(
    `session ${sessionId} extension policy mismatch: ${JSON.stringify(violations)}`,
  );
  err.code = 'SESSION_EXTENSION_POLICY_MISMATCH';
  err.retryable = false;
  throw err;
}

function samePath(left, right) {
  if (!left || !right) return false;
  return path.resolve(left) === path.resolve(right);
}

async function readJson(upstream) {
  const text = await upstream.text();
  if (!upstream.ok) {
    throw new Error(text || `upstream ${upstream.status}`);
  }
  return text ? JSON.parse(text) : null;
}

function isInvalidDirectoryPathError(err) {
  return err instanceof Error && /Invalid directory path/i.test(err.message);
}

/**
 * Re-apply H5 session policy after resume so stale extension sets cannot bypass capability limits.
 */
export async function reconcileAgentSession(
  apiFetch,
  sessionId,
  {
    workingDir,
    sessionPolicy,
    sandboxConstraints = null,
    userContext = null,
    userMemories = null,
    tolerateInvalidWorkingDir = false,
  },
) {
  // Unrestricted users skip capability gating, but Memind still injects sandbox-fs
  // extension_overrides that must be reconciled onto the Goose session.
  if (sessionPolicy?.unrestricted && !(sessionPolicy?.extensionOverrides?.length)) return;

  const session = await readJson(await apiFetch(`/sessions/${sessionId}`));
  let needsRestart = false;
  if (workingDir && !samePath(session?.working_dir, workingDir)) {
    try {
      await readJson(
        await apiFetch('/agent/update_working_dir', {
          method: 'POST',
          body: JSON.stringify({ session_id: sessionId, working_dir: workingDir }),
        }),
      );
    } catch (err) {
      if (!(tolerateInvalidWorkingDir && isInvalidDirectoryPathError(err))) {
        throw err;
      }
      console.warn(
        `[session-reconcile] skip invalid working_dir during resume for session ${sessionId}: ${
          err.message
        }`,
      );
    }
  }

  if (sessionPolicy?.gooseMode && session?.goose_mode !== sessionPolicy.gooseMode) {
    await readJson(
      await apiFetch('/agent/update_session', {
        method: 'POST',
        body: JSON.stringify({
          session_id: sessionId,
          goose_mode: sessionPolicy.gooseMode,
        }),
      }),
    );
    needsRestart = true;
  }

  const desired = sessionPolicy?.extensionOverrides ?? [];
  const allowed = allowedExtensionNames(desired);
  const currentPayload = await readJson(await apiFetch(`/sessions/${sessionId}/extensions`));
  const current = currentPayload?.extensions ?? [];
  if (sessionMcpEnvRequiresFreshSession(session, current, desired)) {
    throw staleMcpEnvError(sessionId);
  }
  let removedAny = false;

  for (const ext of current) {
    const name = extensionName(ext);
    if (!name || allowed.has(name)) continue;
    await readJson(
      await apiFetch('/agent/remove_extension', {
        method: 'POST',
        body: JSON.stringify({ session_id: sessionId, name }),
      }),
    );
    removedAny = true;
  }

  const refreshed = removedAny
    ? ((await readJson(await apiFetch(`/sessions/${sessionId}/extensions`)))?.extensions ?? [])
    : current;
  const { toRemove, toAdd } = extensionsNeedingRefresh(refreshed, desired);

  for (const name of toRemove) {
    await readJson(
      await apiFetch('/agent/remove_extension', {
        method: 'POST',
        body: JSON.stringify({ session_id: sessionId, name }),
      }),
    );
    needsRestart = true;
  }

  for (const config of toAdd) {
    await readJson(
      await apiFetch('/agent/add_extension', {
        method: 'POST',
        body: JSON.stringify({ session_id: sessionId, config }),
      }),
    );
    needsRestart = true;
  }

  if (needsRestart) {
    await readJson(
      await apiFetch('/agent/restart', {
        method: 'POST',
        body: JSON.stringify({ session_id: sessionId }),
      }),
    );
  }

  // Fail closed: the upstream may acknowledge add/remove/restart calls without
  // persisting their effective session configuration. Never let the agent run
  // with default or stale tools after a restricted policy was requested.
  const effectiveExtensions = needsRestart
    ? (
        (
          await readJson(await apiFetch(`/sessions/${sessionId}/extensions`))
        )?.extensions ?? []
      )
    : refreshed;
  assertExtensionPolicyApplied(effectiveExtensions, desired, sessionId);

  const sandboxText = sandboxConstraints?.trim()
    ? buildSandboxSessionConstraints({
        baseConstraints: sandboxConstraints,
        developerTools: developerToolsFromPolicy(sessionPolicy),
      })
    : null;

  const rawMemoryEntries = buildSessionMemoryEntries({
    workingDir,
    sessionPolicy,
    sandboxConstraints: sandboxText,
    userContext,
    userMemories,
  });
  const { entries: memoryEntries, plan: harnessBudgetPlan } = applyContextBudgetToHarnessEntries(
    rawMemoryEntries,
  );
  if (harnessBudgetPlan?.mode === 'shadow' && harnessBudgetPlan.duplicateItems.length > 0) {
    console.log('[session-reconcile] context budget shadow', {
      duplicateCount: harnessBudgetPlan.duplicateItems.length,
      droppedCount: harnessBudgetPlan.droppedItems.length,
      keptBlockCount: harnessBudgetPlan.keptBlockCount,
    });
  }

  if (memoryEntries.length > 0) {
    for (const entry of memoryEntries) {
      if (!entry.content?.trim()) continue;
      await apiFetch('/agent/harness_remember', {
        method: 'POST',
        body: JSON.stringify({
          sessionId,
          content: entry.content,
          title: entry.title,
        }),
      });
    }
    await apiFetch('/agent/harness_bootstrap', {
      method: 'POST',
      body: JSON.stringify({
        sessionId,
        force: true,
      }),
    });
  }
}