stack/packages/mosaic/framework/tools/_scripts/mosaic-ensure-sequential-thinking

#!/usr/bin/env bash
set -euo pipefail

MOSAIC_HOME="${MOSAIC_HOME:-$HOME/.config/mosaic}"
MODE="apply"
RUNTIME="all"
STRICT_CHECK=0

PKG="@modelcontextprotocol/server-sequential-thinking"

err() { echo "[mosaic-seq] ERROR: $*" >&2; }
log() { echo "[mosaic-seq] $*"; }

while [[ $# -gt 0 ]]; do
  case "$1" in
    --check)
      MODE="check"
      shift
      ;;
    --runtime)
      if [[ $# -lt 2 ]]; then
        err "--runtime requires a value: claude|codex|opencode|all"
        exit 2
      fi
      RUNTIME="$2"
      shift 2
      ;;
    --strict)
      STRICT_CHECK=1
      shift
      ;;
    *)
      err "Unknown argument: $1"
      exit 2
      ;;
  esac
done

case "$RUNTIME" in
  all|claude|codex|opencode) ;;
  *)
    err "Invalid runtime: $RUNTIME (expected claude|codex|opencode|all)"
    exit 2
    ;;
esac

require_binary() {
  local name="$1"
  if ! command -v "$name" >/dev/null 2>&1; then
    err "Required binary missing: $name"
    return 1
  fi
}

check_software() {
  require_binary node
  require_binary npx
}

warm_package() {
  local timeout_sec="${MOSAIC_SEQ_WARM_TIMEOUT_SEC:-15}"
  if command -v timeout >/dev/null 2>&1; then
    timeout "$timeout_sec" npx -y "$PKG" --help >/dev/null 2>&1
  else
    npx -y "$PKG" --help >/dev/null 2>&1
  fi
}

check_claude_config() {
  python3 - <<'PY'
import json
from pathlib import Path
p = Path.home() / ".claude" / "settings.json"
if not p.exists():
    raise SystemExit(1)
try:
    data = json.loads(p.read_text(encoding="utf-8"))
except Exception:
    raise SystemExit(1)
mcp = data.get("mcpServers")
if not isinstance(mcp, dict):
    raise SystemExit(1)
entry = mcp.get("sequential-thinking")
if not isinstance(entry, dict):
    raise SystemExit(1)
if entry.get("command") != "npx":
    raise SystemExit(1)
args = entry.get("args")
if args != ["-y", "@modelcontextprotocol/server-sequential-thinking"]:
    raise SystemExit(1)
PY
}

apply_claude_config() {
  python3 - <<'PY'
import json
from pathlib import Path
p = Path.home() / ".claude" / "settings.json"
p.parent.mkdir(parents=True, exist_ok=True)
if p.exists():
    try:
        data = json.loads(p.read_text(encoding="utf-8"))
    except Exception:
        data = {}
else:
    data = {}
mcp = data.get("mcpServers")
if not isinstance(mcp, dict):
    mcp = {}
mcp["sequential-thinking"] = {
    "command": "npx",
    "args": ["-y", "@modelcontextprotocol/server-sequential-thinking"]
}
data["mcpServers"] = mcp
p.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
PY
}

check_codex_config() {
  local cfg="$HOME/.codex/config.toml"
  [[ -f "$cfg" ]] || return 1
  grep -Eq '^\[mcp_servers\.(sequential-thinking|sequential_thinking)\]' "$cfg" && \
  grep -q '^command = "npx"' "$cfg" && \
  grep -q '@modelcontextprotocol/server-sequential-thinking' "$cfg"
}

apply_codex_config() {
  local cfg="$HOME/.codex/config.toml"
  mkdir -p "$(dirname "$cfg")"
  [[ -f "$cfg" ]] || touch "$cfg"

  local tmp
  tmp="$(mktemp)"
  awk '
    BEGIN { skip = 0 }
    /^\[mcp_servers\.(sequential-thinking|sequential_thinking)\]/ { skip = 1; next }
    skip && /^\[/ { skip = 0 }
    !skip { print }
  ' "$cfg" > "$tmp"
  mv "$tmp" "$cfg"

  {
    echo ""
    echo "[mcp_servers.sequential-thinking]"
    echo "command = \"npx\""
    echo "args = [\"-y\", \"@modelcontextprotocol/server-sequential-thinking\"]"
  } >> "$cfg"
}

check_opencode_config() {
  python3 - <<'PY'
import json
from pathlib import Path
p = Path.home() / ".config" / "opencode" / "config.json"
if not p.exists():
    raise SystemExit(1)
try:
    data = json.loads(p.read_text(encoding="utf-8"))
except Exception:
    raise SystemExit(1)
mcp = data.get("mcp")
if not isinstance(mcp, dict):
    raise SystemExit(1)
entry = mcp.get("sequential-thinking")
if not isinstance(entry, dict):
    raise SystemExit(1)
if entry.get("type") != "local":
    raise SystemExit(1)
if entry.get("command") != ["npx", "-y", "@modelcontextprotocol/server-sequential-thinking"]:
    raise SystemExit(1)
if entry.get("enabled") is not True:
    raise SystemExit(1)
PY
}

apply_opencode_config() {
  python3 - <<'PY'
import json
from pathlib import Path
p = Path.home() / ".config" / "opencode" / "config.json"
p.parent.mkdir(parents=True, exist_ok=True)
if p.exists():
    try:
        data = json.loads(p.read_text(encoding="utf-8"))
    except Exception:
        data = {}
else:
    data = {}
mcp = data.get("mcp")
if not isinstance(mcp, dict):
    mcp = {}
mcp["sequential-thinking"] = {
    "type": "local",
    "command": ["npx", "-y", "@modelcontextprotocol/server-sequential-thinking"],
    "enabled": True
}
data["mcp"] = mcp
p.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
PY
}

check_runtime_config() {
  case "$RUNTIME" in
    all)
      check_claude_config
      check_codex_config
      check_opencode_config
      ;;
    claude)
      check_claude_config
      ;;
    codex)
      check_codex_config
      ;;
    opencode)
      check_opencode_config
      ;;
  esac
}

apply_runtime_config() {
  case "$RUNTIME" in
    all)
      apply_claude_config
      apply_codex_config
      apply_opencode_config
      ;;
    claude)
      apply_claude_config
      ;;
    codex)
      apply_codex_config
      ;;
    opencode)
      apply_opencode_config
      ;;
  esac
}

if [[ "$MODE" == "check" ]]; then
  check_software
  check_runtime_config

  # Runtime launch checks should be local/fast by default.
  if [[ "$STRICT_CHECK" -eq 1 || "${MOSAIC_SEQ_CHECK_WARM:-0}" == "1" ]]; then
    if ! warm_package; then
      err "sequential-thinking package warm-up failed in strict mode"
      exit 1
    fi
  fi

  log "sequential-thinking MCP is configured and available (${RUNTIME})"
  exit 0
fi

check_software
if ! warm_package; then
  err "Unable to warm sequential-thinking package (npx timeout/failure)"
  exit 1
fi
apply_runtime_config
log "sequential-thinking MCP configured (${RUNTIME})"